apache-spark - 混合 Spark Structured Streaming API 和 DStream 写入 Kafka

Question

我最近注意到我对 Spark Streaming 感到困惑(我目前正在学习 Spark)。

我正在从这样的 Kafka 主题中读取数据:

val stream = KafkaUtils.createDirectStream[String, String](
      streamingContext,
      PreferConsistent,
      Subscribe[String, String](topics, kafkaParams)
    )

创建一个 DStream。

为了处理事件时间(而不是处理时间)，我这样做了:

outputStream
      .foreachRDD(rdd => {
          rdd.toDF().withWatermark("timestamp", "60 seconds")
            .groupBy(
              window($"timestamp", "60 seconds", "10 seconds")
            )
            .sum("meterIncrement")
            .toJSON
            .toDF("value")
            .writeStream
            .format("kafka")
            .option("kafka.bootstrap.servers", "localhost:9092")
            .option("topic", "taxi-dollar-accurate")
            .start()
        )
      })

我得到了错误

'writeStream' can be called only on streaming Dataset/DataFrame

这让我很吃惊，因为 DF 的源是一个 DStream。不管怎样，我设法通过将 .writeStream 更改为 .write 并将 .start() 更改为 .save() 来解决这个问题。

但我感觉我以某种方式失去了 foreach 的流处理能力。显然这就是我写这个问题的原因。这是正确的方法吗？我见过其他使用的脚本

val df = spark
  .readStream
  .format("kafka")
  .option("kafka.bootstrap.servers", "host1:port1,host2:port2")
  .option("subscribe", "topic1")
  .load()
df.selectExpr("CAST(key AS STRING)", "CAST(value AS STRING)")
  .as[(String, String)]

但我不知道这与仅在 DStream 上调用 foreach 然后将每个 RDD 转换为 DF 有何不同。

Answer 1

But I don't know how different is this from just calling foreach on the DStream and then transforming each RDD to DF.

当你打电话时:

outputStream
      .foreachRDD(rdd => {
          rdd.toDF()
            .[...]
            .toJSON
            .toDF("value")
            .writeStream
            .format("kafka")

您的变量 rdd(或 Dataframe)变成了一个单个 RDD，它不再是一个流。因此，rdd.toDF.[...].writeStream 将不再有效。

继续RDD

如果您选择使用 DSream 方法，您可以发送调用 KafkaProducer API 的那些单个 RDD。

一个例子:

dstream.foreachRDD { rdd =>
  rdd.foreachPartition { partitionOfRecords =>
    val producer = new KafkaProducer[String, String](kafkaParameters)
    partitionOfRecords.foreach { message =>
      producer.send(message)
    }
    producer.close()
  }
}

但是，这不是推荐的方法，因为您在每个执行程序的每个批处理间隔中创建和关闭 KafkaProducer。但这应该让您基本了解如何使用DirectStream API。

要进一步优化将数据发送到 Kafka，您可以遵循给定的指导 here .

继续数据框

但是，您也可以将 RDD 转换为 Dataframe，但要确保调用面向批处理的 API 将数据写入 Kafka:

df.selectExpr("CAST(key AS STRING)", "CAST(value AS STRING)")
  .write
  .format("kafka")
  .option("kafka.bootstrap.servers", "host1:port1,host2:port2")
  .option("topic", "topic1")
  .save()

有关如何将批处理 Dataframe 写入 Kafka 的所有详细信息，请参阅 Spark Structured Streaming + Kafka Integration Guide

注意事项

不过，最重要的是，我强烈建议在这种情况下不要混淆 RDD 和结构化 API，而是坚持使用其中一个。