ストリーム処理のウォーターマークと窓

無限のデータをどう区切るか

バッチ処理は「有限のデータ集合」を前提にします。全件がそろっているから、合計も平均も一意に決まる。ところがストリーム処理が扱うのは終わりのないイベントの列です。センサー、クリックログ、決済、IoTのテレメトリ——止まる時刻がないデータに対して「1時間の売上」を出すには、まず無限を有限に区切る必要があります。この区切りが**窓（ウィンドウ）**です。

区切りの基準となる「時刻」に、実は二つの候補があります。

• イベント時間（event time）：そのイベントが実際に発生した時刻。データ自身に刻まれたタイムスタンプ。
• 処理時間（processing time）：そのイベントが処理システムに到達し処理された時刻。壁時計の時刻。

処理時間で区切るのは簡単です。今の壁時計だけ見ればよい。しかしモバイル端末が圏外に入る、ネットワークが詰まる、リトライで再送される——こうした理由で、イベント時間と処理時間は必ずズレます。このズレをスキュー（skew）と呼びます。スキューがあると、処理時間で「14時台の窓」を作っても、そこには14時に発生したイベントと、13時台に発生したのに遅れて届いたイベントが混ざり、集計がいつ実行したかで結果が変わる非決定的なものになります。

ウォーターマーク：「もう来ない」をどう判断するか

イベント時間で窓を切ると、避けられない問いが生まれます。14時台の窓は、いつ閉じてよいのか。 イベントは発生順に届くとは限りません。13時59分発生のデータが14時03分に、ときには14時10分に届くかもしれない。永遠に待てば完全な結果が出ますが、結果は永遠に出ません。有限時間で窓を閉じるには、「イベント時間のこの点までは、もうデータは来ないだろう」という判断が要ります。これを表現するのが**ウォーターマーク（watermark）**です。

ウォーターマークは、ストリームに差し込まれる単調増加のタイムスタンプ付きマーカーです。「ウォーターマーク W(t) が流れた」とは、イベント時間 t 以下のイベントはもう到着しないという宣言に相当します。処理エンジンは各ソースのウォーターマークを追跡し、イベント時間の進行を表す論理時計として使います。壁時計ではなく、データが運んでくる時刻でシステムの時間が進むのが要点です。

窓の終端を E とすると、ロジックは単純です。

ウォーターマーク W(t) を受信:
  すべての窓 [start, E] について
    if t >= E:              # イベント時間が窓の終端を越えた
      窓を閉じて集計結果を出力（トリガ発火）

問題は、ウォーターマークをどう決めるかです。理想は完全（perfect）ウォーターマーク——入力を完全に把握でき、W(t) を出したら t 以下のデータは絶対に来ないと保証できる場合で、静的なログファイルを昇順に読むようなケースに限られます。現実の分散ストリームでは、どこまで届いたかを全ソースについて正確に知ることはできません。そこで実務はヒューリスティック（推測）ウォーターマークを使います。たとえば「観測した最大イベント時間から一定のスキュー分だけ引いた値」を W(t) とする。この「引く一定分」が**許容する遅れ（bounded delay）**の見積もりです。

窓の三形態：タンブリング・スライディング・セッション

窓は「イベントをどう時間で束ねるか」の戦略です。代表的な三つを、意味論と状態コストの観点で並べます。

タンブリング窓は最も素直です。時間軸を固定長で隙間なく敷き詰め、各イベントは属する窓がただ一つに決まります。「5分ごとの売上」のような定期集計に向き、同時に開く窓が実質一つなので状態も軽い。

スライディング窓（ホッピング窓）は窓長とスライド幅の二つのパラメータを持ちます。スライド幅が窓長より短いと窓どうしが重なり、一つのイベントが複数の窓に同時に属します。「1時間の移動平均を1分ごとに更新」のような、なめらかに動く指標に使います。代償は状態で、任意の瞬間に 窓長 ÷ スライド幅 個の窓が同時に開くため、上の例なら60窓分の集計状態を並行して保持します。

セッション窓は固定の時刻境界を持ちません。同一キー（多くはユーザー）の活動が続く限り一つの窓を延ばし続け、一定の非活動ギャップ（タイムアウト）が空いたら窓を閉じます。「ユーザーの一連の操作を1セッションにまとめる」といった、区切りがデータ自身の間隔で決まるケース向けです。実装は最も重く、キーごとに「開いている窓」と「最後の活動時刻」を保持し、遅れて届いたイベントが二つの既存セッションの隙間を埋めた場合は**窓のマージ（併合）**まで必要になります。

完全性と遅延のトレードオフ：設計として選ぶ

ここまでの要素——ウォーターマークの余裕、窓、late データ——は、一つの本質的なトレードオフに収束します。結果を早く出すほど、まだ届いていないデータを取りこぼす（完全性が下がる）。完全性を上げようと待つほど、結果が遅くなる。 これは実装の巧拙ではなく、無限ストリームを有限時間で区切ることの原理的な帰結で、消すことはできません。だからこそどこで妥協するかを明示的に選ぶのが上級者の設計です。制御に使う独立したつまみが四つあります。

1. ウォーターマークの余裕：観測最大イベント時間からどれだけ引くか。大きく引けば late は減るが窓が閉じるのが遅れる。

2. 許容遅延（allowed lateness）：ウォーターマークが窓の終端を越えて窓を一度閉じた後も、どれだけの間その窓の状態を保持しておくか。この期間内に届いた late データは、対応する窓にまだ振り分けられます。既定値がゼロのエンジンでは、ウォーターマークを越えて届いたデータは即座に破棄（またはサイド出力へ退避）されます。許容遅延を延ばすほど遅刻を救えますが、閉じたはずの窓の状態をメモリに抱え続けるコストが増えます。

3. トリガ（trigger）：窓の結果をいつ発火して出すかの規則。ウォーターマーク到達時に一度出すのが基本ですが、それを待たず途中経過を出す**早期発火（early firing）と、late データが届くたびに結果を更新する遅延発火（late firing）**を組み合わせられます。早期発火は速報を出し、遅延発火は正確さへ後から寄せていく。「速く、かつ正しく」を時間差で両立させる仕組みです。

4. 累積モード（accumulation mode）：同じ窓が複数回発火するとき、前回の結果をどう扱うか。

discarding（破棄）        : 発火ごとに前回分を捨て、その回の増分だけを出す
                           → 下流が自分で足し込む（合算はダウンストリーム側）
accumulating（累積）      : 発火のたびに窓の全データで計算し直し、確定値で上書き
                           → 下流は「同じキーの後勝ち」で最新を採用
accumulating & retracting : 累積に加え、前回値の取り消し（retraction）も発行
                           → 下流が前回を差し引いてから新値を足せる（結合の整合に有効）

これら四つを一枚の設計判断としてまとめると、次のようになります。低遅延の速報が要るなら、ウォーターマークを攻めに設定し早期発火を有効化し、正確さは遅延発火＋accumulating で後追いする。確定値の正確さが最優先なら、ウォーターマークに十分な余裕と長めの許容遅延を取り、発火はウォーターマーク到達時の一回に寄せる。どちらも正解ではなく、要件が決める。遅延・完全性・状態コストの三者を天秤にかけて意識的に選ぶことが、ストリーム集計の設計そのものです。

まとめ

• 無限ストリームは窓で有限に区切る。区切りの基準は、いつ処理したかの処理時間ではなく、いつ発生したかのイベント時間。両者のズレ（スキュー）と順序の乱れが遅延データ問題の根本。
• ウォーターマーク W(t) は「イベント時間 t までは出そろった」という単調増加の宣言で、窓の終端を越えた時点で窓を閉じる。現実のそれは**ヒューリスティック（推測）**なので、宣言より遅れて届く late データが必ず残る。
• 窓はタンブリング（固定長・非重複・状態小）、スライディング（重なりあり・1イベントが複数窓・状態中〜大）、セッション（活動をギャップで区切る・マージあり・状態大）。キー単位で並列化し、最も遅いパーティションが律速する。
• 早く出すほど取りこぼし、待つほど遅いという完全性と遅延のトレードオフは消せない。ウォーターマークの余裕・許容遅延・トリガ（早期／遅延発火）・累積モードの四つで、妥協点を明示的に設計する。
• 早期／遅延発火で窓は複数回発火し、配信も at-least-once が前提。下流は accumulating で後勝ち上書き、あるいは retracting で取り消し差し引きによる冪等な処理が必須。