【发布时间】：2019-08-23 21:09:38
【问题描述】：
我们有一个在 Spark 2.3.3 上运行的 Spark Streaming 应用程序
基本上，它会打开一个 Kafka Stream：
kafka_stream = spark \
.readStream \
.format("kafka") \
.option("kafka.bootstrap.servers", "mykafka:9092") \
.option("subscribe", "mytopic") \
.load()
kafka 主题有 2 个分区。之后，还有一些基本的过滤操作，一些 Python UDF 和一个列上的 explode()，比如：
stream = apply_operations(kafka_stream)
apply_operations 对数据进行所有工作。最后，我们想将流写入接收器，即。 e.：
stream.writeStream \
.format("our.java.sink.Class") \
.option("some-option", "value") \
.trigger(processingTime='15 seconds') \
.start()
为了让这个流操作永远运行，我们应用：
spark.streams.awaitAnyTermination()
最后。
到目前为止，一切都很好。一切都运行了几天。但是由于网络问题，job 死了几天，现在 kafka 流中有 百万条消息等待被追上。
当我们使用 spark-submit 重新启动流数据作业时，第一批会太大，需要很长时间才能完成。我们认为可能有一种方法可以通过一些参数来限制第一批的大小，但我们没有找到任何帮助。
我们试过了：
-
spark.streaming.backpressure.enabled=true 以及 spark.streaming.backpressure.initialRate=2000 和 spark.streaming.kafka.maxRatePerPartition=1000 和 spark.streaming.receiver.maxrate=2000
将 spark.streaming.backpressure.pid.minrate 设置为较低的值也没有效果
设置选项("maxOffsetsPerTrigger", 10000) 也没有效果
现在，在我们重新启动管道后，整个 Spark Job 迟早会再次崩溃。我们不能简单地扩大用于 spark 作业的内存或内核。
我们是否遗漏了什么来控制在一个流批处理中处理的事件数量？
【问题讨论】：
-
你试过批处理吗？
-
您的意思是，将流应用程序重写为“普通”批处理应用程序？我希望可以避免重写并调整一些参数？
-
backpressure.initialRate 不适用于结构化流，但 maxOffsetsPerTrigger 应该可以。他为第一批收集了多少条消息？
-
@cronoik 当我查看 Spark UI 的作业选项卡时，收集了数百万条消息，还有数百万条记录要写入接收器。通常，当流启动并运行时，每个触发器会处理大约 2000 条消息。但是，当流落后时，它似乎正在接收所有可用的消息，而不管任何尝试过的选项。
-
为了确定，您为读取流设置了 maxOffsetsPerTrigger？您能否在 pysparkshell 中尝试以下操作：
kafka_stream = spark.readStream.format("kafka").option("kafka.bootstrap.servers", "mykafka:9092").option("subscribe", "mytopic").option('maxOffsetsPerTrigger', 4).load()和query = df.writeStream.queryName("bla").outputMode("append").format("console").start()您得到多少行？
标签： apache-spark pyspark spark-streaming spark-structured-streaming