PySpark Structured Streaming with Kafka - 为具有不同负载的多个主题扩展消费者答案

【问题标题】：PySpark Structured Streaming with Kafka - Scaling Consumers for multiple topics with different loadsPySpark Structured Streaming with Kafka - 为具有不同负载的多个主题扩展消费者
【发布时间】：2022-10-20 21:00:44
【问题描述】：

我们订阅了7个主题与spark.readStream 在1单跑步火花应用. 转换事件有效负载后，我们将它们与spark.writeStream 一起保存到我们的数据库中。

对于其中一个主题，数据仅以非常高的负载分批（每天一次）插入。这会延迟我们的阅读全部其他主题也是如此。例如（grafana），所有主题的生产记录和消费记录之间的延迟全天保持在 1m 以下。当 bulk-topic 收到它的事件时，我们的延迟增加长达 2 小时全部（！）话题.

我们如何解决这个问题？我们已经尝试了 2 个连续的 readStreams（单独的批量主题），但它没有帮助。
更多信息：我们使用 6 个执行器，2 个执行器核心。主题有不同数量的分区（3 到 30）。结构化流式处理 Kafka 集成 v0.10.0。
一般问题：我们如何在 Spark 结构化流中扩展消费者？ 1 个 readStream 是否等于 1 个消费者？或1个执行人？还是什么？

【问题讨论】：

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

【解决方案1】：

分区是Kafka并行的主要来源，所以我建议你增加分区的数量（至少对于有性能问题的主题）。您还可以调整doc 中提到的一些消费者缓存选项。尽量保持分区数为 2^n。最后，如果可能的话，你可以增加驱动机器的大小。
我不完全确定，但我认为 Spark 会尽量保持与每个主题的分区数量相同的消费者数量。另外我认为实际上流总是从 Spark 驱动程序中获取的（而不是从工作人员那里获取的）。

【讨论】：

【解决方案2】：

我们找到了解决问题的方法：
更改后我们的grafana 显示，批处理数据主题仍然达到峰值，但没有阻塞其他主题的消费。

我们做了什么：
我们还有 1 个 spark 应用程序。我们使用了 2 个连续的 spark.readStreams，但也为每个添加了一个接收器。

在代码中：

priority_topic_stream = spark.readStream.format('kafka')
     .options(..).option('subscribe', ','.join([T1, T2, T3])).load()
bulk_topic_stream = spark.readStream.format('kafka')
     .options(..).option('subscribe', BULK_TOPIC).load()

priority_topic_stream.writeStream.foreachBatch(..).trigger(..).start()
bulk_topic_stream.writeStream.foreachBatch(..).trigger(..).start()
spark.streams.awaitAnyTermination()

为了最大限度地减少批量流的峰值，我们将尝试增加其分区，如@partlov 建议的那样。但这只会加快批量流的消耗，但不能解决阻止我们从优先主题读取的问题。

【讨论】：