Apache Sparkストリーミング - タイムアウトの長時間実行のバッチ

Question

InputDStreamを使用して（並列化されていない）ストリーミングを実行するために、Apache Sparkの長時間実行されるストリーミングジョブを設定しています。

私が達成しようとしているのは、キューのバッチが（ユーザー定義のタイムアウトに基づいて）時間がかかり過ぎると、バッチをスキップして完全に放棄できるようにしたいということです。実行。

spark APIまたはオンラインでこの問題の解決策を見つけることができませんでした - StreamingContext awaitTerminationOrTimeoutを調べましたが、これはタイムアウト時にStreamingContext全体を強制終了しますが、私がしたいのはスキップ/キルです現在のバッチ。

また、mapWithStateを使用すると考えられましたが、これはこの使用例には当てはまりません。最後に、私はStreamingListenerを設定し、バッチが開始されたときにタイマーを開始し、一定のタイムアウトしきい値に達したときにバッチを停止/スキップ/終了させることを検討していましたが、バッチを終了する方法はまだありません。

ありがとうございます！

Answer 1

私はyelpからいくつかのドキュメントを見ましたが、私はそれを自分でやっていません。イベントが最初に見られていると状態は、それが

def update_function(new_events, current_state): 
    if current_state is None: 
     current_state = init_state() 
     attach_expire_datetime(new_events) 
     ...... 
    if is_expired(current_state): 
     return None //current_state drops? 
    if new_events: 
     apply_business_logic(new_events, current_state) 

を満了した場合

これはそのように見える状態をドロップ初期化されるときUpdateStateByKey(update_func)またはmapWithState(stateSpec)、

1. を使用して

   はタイムアウトを添付します構造化ストリーミングウォーターマークは、タイムアウトしたときにイベントをドロップします。これがジョイに適用される場合bs/stagesタイムアウトの低下。