在Spark Structured Streaming中应用消息级别的模式而不是数据帧级别的模式

case class UserEvent(userId: Int, action: String)
case class SystemEvent(eventType: String, details: Map[String, String])

val userEvents = MemoryStream[UserEvent]
val systemEvents = MemoryStream[SystemEvent]

val events = userEvents.union(systemEvents)

events.writeStream.foreachBatch { (batchDF: DataFrame, batchId: Long) =>
  batchDF.foreach {
    case userEvent: UserEvent => // 处理用户事件
    case systemEvent: SystemEvent => // 处理系统事件
  }
}.start()

events.writeStream.foreachBatch { (batchDF: DataFrame, batchId: Long) =>
  batchDF.withColumn("eventType", expr("typeOf(this)")).as[(Any, String)].foreach {
    case (userEvent: UserEvent, _) => // 处理用户事件
    case (systemEvent: SystemEvent, _) => // 处理系统事件
    case _ => // 处理未知类型的事件或错误
  }
}.start()