ストリーム処理｜データエンジニアリング

🎓 レベル：発展　|　重要度：A（必須）

📎 前提：メッセージキューとイベントストリーミング・バッチ処理とストリーム処理　|　関連：ウィンドウ関数

要点（BLUF）

ストリーム処理は、終わりのないイベント列を**ウィンドウ（時間の区切り）**に分けて集計する。タンブリング（重ならない固定幅）・スライディング（重なる）・セッション（活動の切れ目）が基本。
難所は時間。イベントが発生したイベント時間と、システムが処理した処理時間がずれ、ネットワーク遅延で順序も狂う。正しい集計はイベント時間で行う。
遅れて届くデータ（遅延データ）にどう対処するかが核心。**watermark（ここまでは出揃ったとみなす境界）**で「いつ集計を確定するか」を決め、それを超えた遅延データは隔離・別処理する。

概念 ── 無限の流れを区切る

バッチは「有限のデータの塊」を処理しますが、ストリームは「終わらない流れ」。そのままでは集計できないので、ウィンドウで時間を区切って「この10秒分の合計」を出します。

flowchart LR
    E["イベント列（無限）"] --> W0["window 0-10秒"]
    E --> W1["window 10-20秒"]
    E --> W2["window 20-30秒"]
    W0 --> A0["集計"]
    W1 --> A1["集計"]
    W2 --> A2["集計"]

仕組み ── イベント時間ウィンドウと遅延データ

10秒幅のタンブリングウィンドウで集計し、watermarkを超えて遅れて届いたデータは隔離します。

WIN = 10
def window_start(t): return (t // WIN) * WIN
# watermark（処理済みの最大イベント時間）より十分古いものは遅延として隔離
if t < watermark - WIN:
    late.append((t, v))
else:
    windows[window_start(t)] ...  # 該当ウィンドウに加算

実行結果（実機・純Pythonで再現）:

ウィンドウ別集計（イベント時間基準）:
  [ 0,10): 件数=3 合計=60
  [10,20): 件数=3 合計=150
  [20,30): 件数=1 合計=70
遅延データ（watermark超過で隔離）: [(5, 99)]

イベント時間で各ウィンドウに正しく振り分けられ、最後に届いた (5, 99)（本来は[0,10)のはずが、watermarkが22まで進んだ後に到着）は遅延データとして隔離されました。「いつまで待って集計を確定するか」を決めるのがwatermarkの役割です。

設計の勘所 ── ウィンドウと時間の扱い

種類	形	用途
タンブリング	重ならない固定幅	1分ごとの売上
スライディング	重なる固定幅	直近5分の移動平均
セッション	活動の切れ目で区切る	ユーザーの操作セッション

イベント時間＞処理時間：正しさはイベント時間で。処理時間は「いつ計算したか」に過ぎず、遅延で結果が変わる。
watermarkのトレードオフ：早く確定すれば鮮度が上がるが遅延データを取りこぼす。長く待てば正確だが遅い。鮮度と正確さの綱引き。
状態管理：ウィンドウごとの途中集計は「状態」。これを耐障害に保つのがストリームエンジン（Flink等）の仕事。

なぜそうするか ── 鮮度のための難しさを引き受ける

なぜこの複雑さを引き受けるのか。**秒単位の鮮度がビジネス価値を持つ場面（不正検知・即時モニタリング）**があるからです。バッチでは「1時間後に異常が分かる」では遅い。ただしバッチ処理とストリーム処理の通り、鮮度が要らないならバッチが断然簡単。ストリームは「時間・順序・遅延・状態」という固有の難しさを抱えるので、鮮度要件が明確なときだけ選びます。なお、ウィンドウ集計の発想自体はウィンドウ関数の枠付き集計と同じ——「区切って、その中で集める」が両者の共通形です。

⚠️ よくある落とし穴

処理時間で集計 → 遅延・再処理で結果が変わり再現しない。イベント時間で。
watermarkを設けず「全部待つ」 → 永遠に確定しない。境界を決めて確定する。
遅延データを黙って捨てる → 集計が過少に。隔離して別途反映するか、許容遅延を明示。
状態を無制限に保持 → メモリ枯渇。古いウィンドウは確定後に破棄。

対応ラボ

data-engineering-study/labs/07_streaming_orchestration.py（PYTHONIOENCODING=utf-8 で実行・ウィンドウ集計と遅延データ隔離を確認済み）。