Spark Dataframe 到 Kafka

Question

我正在尝试将 Spark Dataframe 流式传输到 Kafka 消费者。我做不到，请您给我建议。

我能够从 Kafka 生产者那里挑选数据到 Spark，并且我已经执行了一些操作，在处理数据之后，我有兴趣将其流式传输回 Kafka（消费者）。

Answer 1

这是一个流式生产到 kafka 的示例，但批处理版本几乎相同

从源流式传输到 kafka：

 val ds = df
      .selectExpr("CAST(key AS STRING)", "CAST(value AS STRING)")
      .writeStream
      .format("kafka")
      .option("kafka.bootstrap.servers", "host1:port1,host2:port2")
      .option("topic", "topic1")
      .start()

将静态数据帧（不是从源流式传输）写入 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()

请记住

1. 每一行都是一条消息。
2. 数据帧必须是流数据帧。如果您有静态数据框，请使用静态版本。

查看基本文档：https://spark.apache.org/docs/2.2.0/structured-streaming-kafka-integration.html

听起来您有一个静态数据帧，它不是从源流式传输的。