混合 Spark Structured Streaming API 和 DStream 写入 Kafka

Posted 2023-04-18

【中文标题】混合 Spark Structured Streaming API 和 DStream 写入 Kafka

【英文标题】：Mixing Spark Structured Streaming API and DStream to write to Kafka

【发布时间】：2020-07-13 03:38:17

【问题描述】：

我最近注意到我对 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' 只能在流数据集/数据帧上调用

这让我很惊讶，因为 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 有何不同。

【参考方案1】：

但我不知道这与仅在 DStream 上调用 foreach 然后将每个 RDD 转换为 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。 但这应该让您基本了解如何使用以下方法将数据写入 Kafka 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 混为一谈，而是坚持使用其中一种。