Spark结构流媒体-联合两个或多个流媒体源

Question

val finalDF = flatDF1
      .union(flatDF2)
      .union(flatDF3)

val query = finalDF.writeStream
      .format("parquet")
      .outputMode("append")
      .option("path", hdfsLocation)
      .option("checkpointLocation", checkpointLocation)
      .option("failOnDataLoss", false)
      .start()

    query.awaitTermination()

Caused by: java.lang.AssertionError: assertion failed
    at scala.Predef$.assert(Predef.scala:156)
    at org.apache.spark.sql.execution.streaming.OffsetSeq.toStreamProgress(OffsetSeq.scala:42)
    at org.apache.spark.sql.execution.streaming.MicroBatchExecution.org$apache$spark$sql$execution$streaming$MicroBatchExecution$$populateStartOffsets(MicroBatchExecution.scala:185)
    at org.apache.spark.sql.execution.streaming.MicroBatchExecution$$anonfun$runActivatedStream$1$$anonfun$apply$mcZ$sp$1.apply$mcV$sp(MicroBatchExecution.scala:124)
    at org.apache.spark.sql.execution.streaming.MicroBatchExecution$$anonfun$runActivatedStream$1$$anonfun$apply$mcZ$sp$1.apply(MicroBatchExecution.scala:121)
    at org.apache.spark.sql.execution.streaming.MicroBatchExecution$$anonfun$runActivatedStream$1$$anonfun$apply$mcZ$sp$1.apply(MicroBatchExecution.scala:121)

case class OffsetSeq(offsets: Seq[Option[Offset]], metadata: Option[OffsetSeqMetadata] = None) {

assert(sources.size == offsets.size)

这是因为检查点只存储了其中一个数据流的偏移量吗？浏览Spark结构流文档，似乎可以在Spark 2.2或>中进行流源的联接/联合

Answer 1

首先，请定义您的case类OffsetSeq如何与具有数据规则联合的代码相关联。

接下来，在执行这个联合并用WriteStream写到Kafka时，检查点是一个真正的问题。分离成多个WriteStream--每个writestreams都有自己的检查点--会因为联合操作而混淆批ID。使用相同的writestream与union of dataframes失败，因为检查点似乎查找union之前生成dataframes的所有模型，并且无法区分哪些行/记录来自哪些Dataframe/Model。

为了写到Kafka，从结构化sql流统一了Dataframes-最好使用writestream与foreach和ForEachWriter，包括进程方法中的Kafka生产者。不需要检查点；应用程序只使用临时检查点文件，这些文件被设置为在适当的时候删除--将“ForceDeleteTempCheckPointLocation”设置为true--在会话生成器中。

val output = dataFrameModelArray.reduce(_ union _)
val stream: StreamingQuery = output
  .writeStream.foreach(new ForeachWriter[Row] {

    def open(partitionId: Long, version: Long): Boolean = {
      true
    }

    def process(row: Row): Unit = {
      val producer: KafkaProducer[String, String] = new KafkaProducer[String, String](props)
      val record = new ProducerRecord[String, String](producerTopic, row.getString(0), row.getString(1))
      producer.send(record)
    }

    def close(errorOrNull: Throwable): Unit = {
    }
  }
).start()

如果需要，可以在过程方法中添加更多的逻辑。

注在联合之前，所有要联合的数据集都已转换为键值字符串列。值是要通过Kafka生成器发送的消息数据的json字符串。这对于在尝试联合之前获得write也是非常重要的。

svcModel.transform(query)
    .select($"key", $"uuid", $"currentTime", $"label", $"rawPrediction", $"prediction")
    .selectExpr("key", "to_json(struct(*)) AS value")
    .selectExpr("CAST(key AS STRING)", "CAST(value AS STRING)")

其中svcModel是DataFrameModelArray中的dataframe。