如何将 Spark 流合并到 Kafka 的数组中答案

【问题标题】：How to consolidate the Spark streaming into array to Kafka如何将 Spark 流合并到 Kafka 的数组中
【发布时间】：2018-10-28 19:28:12
【问题描述】：

目前，我有以下 df

+-------+--------------------+-----+
|    key|          created_at|count|
+-------+--------------------+-----+
|Bullish|[2017-08-06 08:00...|   12|
|Bearish|[2017-08-06 08:00...|    1|
+-------+--------------------+-----+

我使用以下内容将数据流式传输到 Kafka

df.selectExpr("CAST(key AS STRING) AS key", "to_json(struct(*)) AS value")
  .writeStream
  .format("kafka")
  .option("kafka.bootstrap.servers", "localhost:9092").option("topic","chart3").option("checkpointLocation", "/tmp/checkpoints2")
  .outputMode("complete")
  .start()

这里的问题是，对于DataFrame中的每一行，都会一一写入Kafka。我的消费者会一一收到消息。

有没有办法将所有行合并到一个数组中并流式传输到 Kafka，以便我的消费者可以一次性获取全部数据。

感谢您的建议。

【问题讨论】：

标签： apache-spark apache-kafka spark-structured-streaming

【解决方案1】：

我的消费者会一一收到消息。

不完全是。它可能取决于 Kafka 属性。您可以指定自己的属性并使用例如：

props.put("batch.size", 16384);

Spark 在后台使用普通缓存的 KafkaProducer。它将使用您在提交查询时在选项中提供的属性。

另见Java Doc。请注意，它可能无法正确缩放

【讨论】：