apache-spark - Spark Dataframe 到 Kafka

Question

我正在尝试将 Spark Dataframe 流式传输到 Kafka 消费者。我做不到，请你指教。

我能够从 Kafka producer 中选择数据到 Spark ，并且我已经执行了一些操作，在操作数据之后，我有兴趣将它流回 Kafka(Consumer)。

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

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