重新启动Spark结构化流作业会消耗数百万个Kafka消息并消亡

Question

我们有一个运行在Spark 2.3.3上的Spark Streaming应用程序

基本上，它将打开一个Kafka流：

  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将对数据进行所有工作。最后，我们想将流写入接收器，即：

   stream.writeStream \
   .format("our.java.sink.Class") \
   .option("some-option", "value") \
   .trigger(processingTime='15 seconds') \
   .start()

为了让此流操作永久运行，我们应用：

   spark.streams.awaitAnyTermination()

到底。

到目前为止，一切都很好。一切都会持续数天。但是由于网络问题，该工作终止了几天，现在卡夫卡流中有数百万条消息正在等待追赶。

当我们使用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设置为较低的值也没有效果

• 设置option（“ maxOffsetsPerTrigger”，10000）也不起作用

现在，在我们重新启动管道之后，迟早整个Spark Job都会再次崩溃。我们不能简单地扩展用于火花作业的内存或内核。

我们是否有任何东西无法控制一个流批量中处理的事件数量？

Answer 1

您在评论中写道，您正在使用spark-streaming-kafka-0-8_2.11并且该 api 版本无法处理 maxOffsetPerTrigger （或据我所知任何其他减少消耗消息数量的机制），因为它只实现了新的API火花流-卡夫卡0-10_2.11。根据文档，这个较新的 api 也适用于您的 kafka 版本 0.10.2.2 。