私はカフカのトピックから受信して処理が完了した後にストリームの処理を終了したいと思います。停止は、（awaitTerminationOrTimeout）のような特定の時間であってはなりません。トピックが消耗した後にsparkstreamingcontextを停止する方法はありますか？ Dstream [T]をT値と比較して制御フローを指示する方法はありますか？stop sparkストリーミングコンテクストkafkaDirectStream
私はisEmptyがtrueを返すと、ストリームが空の場合headOptionはになしてはならないことを約80％cetainです。
ストリームの読み込みを開始する前に、トピック内のすべてのパーティションの最新のオフセットを取得してから、受信したオフセットがいつ取得されたかを確認するのが最善の方法です。トピックのオフセットを取得する方法については、previous answerを参照してください。
フローはビーイングを終わる：
OffsetRequestを行い、各パーティションについてSimpleConsumerOffsetRequestあなたがkafkaStream.foreachRDDの
rdd.isEmptyはtrueを返しません行われにreceieved 最新のと同じです。しかし、sparkStreamingContext.stopでアプリケーションを停止することはありません。 –
'future（{（）=> while（！kafkaMessageStream.isEmpty）{ Thread.sleep（100）} sparkTreamingContext。 （真） }） ' これは100ミリ秒ごとにメッセージがあるかどうかをチェックし、メッセージがない場合は停止します。 –
偽陽性を得ることができます。たとえば、ブローカがバッチ間隔よりも長い時間忙しくなった場合などです。 –