重新启动Spark Structured Streaming Job会消耗数百万条Kafka消息并死亡

Question

我们有一个运行在Spark2.3.3上的Spark流应用程序

基本上，它开启了一条Kafka流：

  kafka_stream = spark \
  .readStream \
  .format("kafka") \
  .option("kafka.bootstrap.servers", "mykafka:9092") \
  .option("subscribe", "mytopic") \
  .load()

   stream = apply_operations(kafka_stream)

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

   spark.streams.awaitAnyTermination()

我们尝试：

>

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作业会再次崩溃。我们不能简单地扩大内存或内核来用于spark工作。

在控制一个流批处理中处理的事件的数量方面，我们是否遗漏了什么？

Answer 1

您在注释中写道，您使用的是spark-streaming-kafka-0-82.11，并且该api版本不能处理maxOffsetPerTrigger（或据我所知的任何其他减少已消耗消息数量的机制），因为它只是为较新的api spark-streaming-kafka-0-102.11实现的。根据文档，这个较新的api也适用于您的kafka版本0.10.2.2。