Spark 流式处理待处理批次

Question

我正在运行一个Spark Streaming 应用程序，它从Kafka 读取数据（使用Direct Stream 方法）并将结果发布回Kafka。应用程序的输入速率以及应用程序的吞吐量在大约一两个小时内保持稳定。之后，我开始看到在Active Batches 队列中保留很长时间（30 分钟+）的批次。 Spark driver日志指出了以下两种错误，并且这些错误的发生时间与卡住的批次的开始时间吻合良好：

第一个错误类型

ERROR LiveListenerBus: Dropping SparkListenerEvent because no remaining room in event queue. This likely means one of the SparkListeners is too slow and cannot keep up with the rate at which tasks are being started by the scheduler.

第二种错误类型

ERROR StreamingListenerBus: Listener StreamingJobProgressListener threw an exception
java.util.NoSuchElementException: key not found: 1501806558000 ms
  at scala.collection.MapLike$class.default(MapLike.scala:228)
  at scala.collection.AbstractMap.default(Map.scala:59)
  at scala.collection.mutable.HashMap.apply(HashMap.scala:65)
  at org.apache.spark.streaming.ui.StreamingJobProgressListener.onOutputOperationCompleted(StreamingJobProgressListener.scala:134)
  at org.apache.spark.streaming.scheduler.StreamingListenerBus.doPostEvent(StreamingListenerBus.scala:67)
  at org.apache.spark.streaming.scheduler.StreamingListenerBus.doPostEvent(StreamingListenerBus.scala:29)
  at org.apache.spark.util.ListenerBus$class.postToAll(ListenerBus.scala:63)
  at org.apache.spark.streaming.scheduler.StreamingListenerBus.postToAll(StreamingListenerBus.scala:29)
  at org.apache.spark.streaming.scheduler.StreamingListenerBus.onOtherEvent(StreamingListenerBus.scala:43)
  at org.apache.spark.scheduler.SparkListenerBus$class.doPostEvent(SparkListenerBus.scala:75)
  at org.apache.spark.scheduler.LiveListenerBus.doPostEvent(LiveListenerBus.scala:36)
  at org.apache.spark.scheduler.LiveListenerBus.doPostEvent(LiveListenerBus.scala:36)
  at org.apache.spark.util.ListenerBus$class.postToAll(ListenerBus.scala:63)
  at org.apache.spark.scheduler.LiveListenerBus.postToAll(LiveListenerBus.scala:36)
  at org.apache.spark.scheduler.LiveListenerBus$$anon$1$$anonfun$run$1$$anonfun$apply$mcV$sp$1.apply$mcV$sp(LiveListenerBus.scala:94)
  at org.apache.spark.scheduler.LiveListenerBus$$anon$1$$anonfun$run$1$$anonfun$apply$mcV$sp$1.apply(LiveListenerBus.scala:79)
  at org.apache.spark.scheduler.LiveListenerBus$$anon$1$$anonfun$run$1$$anonfun$apply$mcV$sp$1.apply(LiveListenerBus.scala:79)
  at scala.util.DynamicVariable.withValue(DynamicVariable.scala:58)
  at org.apache.spark.scheduler.LiveListenerBus$$anon$1$$anonfun$run$1.apply$mcV$sp(LiveListenerBus.scala:78)
  at org.apache.spark.util.Utils$.tryOrStopSparkContext(Utils.scala:1279)
  at org.apache.spark.scheduler.LiveListenerBus$$anon$1.run(LiveListenerBus.scala:77)

但是，我不确定如何解释这些错误，尽管进行了广泛的在线搜索，但我找不到任何与此相关的有用信息。

问题

1. 这些错误是什么意思？它们是否表示资源限制（例如：CPU、内存等）？
2. 修复这些错误的最佳方法是什么？

提前致谢。

Answer 1

你的批处理时间不是比真正的批处理时间短吗？默认批处理队列大小为 1000，因此 Spark Streaming 批处理队列可能会溢出。