停止火花流上下文 kafkaDirectStream答案

【问题标题】：stop spark Streaming context kafkaDirectStream停止火花流上下文 kafkaDirectStream
【发布时间】：2016-05-03 21:24:45
【问题描述】：

我想在kafka主题接收和处理完成后结束流处理。停止不应像（ awaitTerminationOrTimeout ）那样特定于时间。有没有办法在主题耗尽后停止 sparkstreamingcontext。有没有办法将 Dstream[T] 与 T 值进行比较来指示控制流？

【问题讨论】：

【解决方案1】：

如果流为空，我有 80% 的把握认为 isEmpty 应该返回 true，headOption 应该在 KafkaMessageStream 上为 None。

【讨论】：

rdd.isEmpty of kafkaStream.foreachRDD 确实返回 true。但它不会使用 sparkStreamingContext.stop(true) 停止应用程序
您可以执行以下操作：Future({() => while(!kafkaMessageStream.isEmpty) { Thread.sleep(100)} sparkTreamingContext.stop(true) }) 这将每 100 毫秒检查一次是否有任何消息，如果没有则停止。
您可能会得到误报，例如，如果代理出现问题的时间长于您的批处理间隔。
我开始使用 sys.exit() 而不是 sparkStreamingContext.stop() 因为后者不会停止驱动程序中的应用程序。

【解决方案2】：

最好的方法是，在开始读取流之前，获取主题中所有分区的最新偏移量，然后检查接收到的偏移量何时到达那里。如果您想了解如何获取主题的偏移量，请参阅我的 previous answer。

流程最终是：

【讨论】：