混合 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 方法，您可以发送那些单个 RDD 调用 KafkaProducer API.

一个例子：

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

要进一步优化将数据发送到 Kafka，您可以按照 here.

给出的指导进行操作

继续使用 Dataframe

但是，您也可以将 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，而是坚持使用其中一种。

混合 Spark Structured Streaming API 和 DStream 写入 Kafka

Mixing Spark Structured Streaming API and DStream to write to Kafka

apache-spark

spark-streaming

spark-structured-streaming

继续使用 RDD

继续使用 Dataframe

备注