hadoop - ネットワーク接続 (TCP IP) が閉じているときに Spark Streaming コンテキストを停止するには?

Question

つまり、Spark ストリーミング コンテキストで「期間」を値に設定する代わりに、(ソケット クローズ時間 - ソケット オープン時間) に設定したい

Answer 1

StreamingListnerインターフェイスを使用して、切断されているレシーバーをリッスンし、ストリーミング コンテキストをシャットダウンできます。

これは次のように使用されます

// define listener
class MyListener extends StreamingListener {
  override def onReceiverStopped(...) {
    streamingContext.stop()
  }
} 

// attach listener
streamingContext. addStreamingListener(new MyListener())

Answer 2

ストリーミング リスナーで ssc.stop() を実行すると、ストリーミングがエラーになるか終了したときに、この関数がトリガーされて停止します。バッチの開始時、終了時など、別のトリガーがあります。

    ssc.addStreamingListener(new StreamingListener() {

    @Override
      public void onReceiverError(StreamingListenerReceiverError receiverError) {
      System.out.println("Do what u want");
      ssc.stop();
      }

    @Override
      public void onReceiverStopped(StreamingListenerReceiverStopped receiverStopped) {
      System.out.println("Do what u want");
      ssc.stop(true,true); 
      }
    }

    ssc.start();