ストリーム処理のexactly-once

三つの配信保証：何が「一度」なのか

長時間走り続けるストリーム処理は、途中で必ず落ちる。ノード障害、デプロイ、ネットワーク分断。落ちた後にどこから再開するかで、各入力レコードが結果に反映される回数が変わる。これが配信保証（delivery guarantee）で、三段階ある。

• at-most-once（高々一度）: 再送しない。落ちた瞬間に処理中だったレコードは失われる。重複はないが欠落する。実装は最も簡単で、メトリクスのサンプリングのように取りこぼしを許せる用途向け。
• at-least-once（少なくとも一度）: 未確定のレコードを再送する。欠落しないが重複する。多くのストリーム基盤の既定値で、下流が重複に強ければこれで足りる。
• exactly-once（ちょうど一度）: 各レコードの効果が結果にちょうど一度だけ現れる。欠落も重複もない。カウント・課金・残高のように一件のずれが誤りになる集計で要る。

なぜ状態が問題を難しくするのか

ストリーム処理が単純なメッセージ転送と決定的に違うのは、演算子が状態を持つ点にある。ウィンドウ集計のカウンタ、ストリーム結合のバッファ、重複排除のセット。入力レコード1件は、この内部状態を更新し、下流へ結果を出す。

at-least-once で再送すると、この状態更新が二度走る。「注文数を+1」を二度適用すればカウントは狂う。つまりストリーム処理の exactly-once とは、再開したときに内部状態が『ちょうど一度だけ各入力を反映した値』に戻り、かつ外部への出力もそれと一致することに他ならない。問題は次の三つに分解できる。

(1) 入力位置   : どのレコードまで取り込んだか（ソースのオフセット）
(2) 内部状態   : 各演算子が持つカウンタ・バッファ・集約値
(3) 外部出力   : シンク（DB・別トピック）へ書いた結果
この三者を「同一時点」に揃えて復元できれば effectively-once になる。

素朴には「全処理を止めて(1)(2)(3)を一斉に保存」すれば揃うが、分散した多数の演算子を同時停止するのは高コストで、スループットを殺す。止めずに整合したスナップショットを撮る——ここに分散スナップショットのアルゴリズムが要る。

チェックポイントとバリア：分散スナップショットの応用

Flinkに代表される方式は、チェックポイント（checkpoint）と呼ぶ全演算子の整合したスナップショットを定期的に撮り、障害時はそこへ全体を巻き戻す。中核は Chandy-Lamport の分散スナップショットアルゴリズムを、データフローの向きに合わせて単純化したバリア（barrier）方式だ。

仕組みはこうだ。チェックポイント・コーディネータが各ソースへ「チェックポイントn を開始せよ」と指示する。ソースは現在の入力オフセットを記録し、データストリームの中にバリアという特殊なマーカーを挿入する。バリアは通常のレコードと同じ経路を、追い越しも追い越されもせず流れていく。

ソース ──[レコード][レコード]<バリアn>[レコード]──▶ 演算子A ──▶ 演算子B ──▶ シンク
                                  ↑
      バリアnより前のレコードは状態に反映済み、後のレコードはまだ、という境界線

演算子はバリアを受け取ると、その瞬間の自分の状態を永続ストレージ（分散ファイルシステムやオブジェクトストア）へ書き出し、バリアを下流へ流す。全演算子がバリアn を通し切り、シンクまで到達すると、チェックポイントn が完成する。こうして撮れた {ソースのオフセット, 全演算子の状態} は、「バリアより前の全レコードだけを反映した」整合したカット（consistent cut）になっている。バリアは実際のデータと一緒に流れるだけなので、処理は基本的に止まらない。

障害が起きたら、最後に完成したチェックポイントn を使い、全演算子の状態を復元し、ソースを記録済みオフセットまで巻き戻して再開する。バリアn 以降に処理していたレコードは再度流れてくるが、状態はバリアn 時点に戻っているので、結果として各レコードはちょうど一度だけ反映される。ここまでがジョブ内部の exactly-once だ。

内部だけでは足りない：外部シンクの壁

チェックポイントが守るのは(1)(2)、すなわちオフセットと内部状態である。しかし(3)の外部への出力は基本的にロールバックできない。データベースへ INSERT した行、別のKafkaトピックへ送ったメッセージは、ジョブが巻き戻っても消えない。

時刻t1: レコードr を処理 → 外部DBへ結果を書き込み（コミット済み）
時刻t2: チェックポイント完成前にクラッシュ
時刻t3: 最後のチェックポイントへ巻き戻し、r を再処理 → DBへ再度書き込み
        → 外部DBに r の効果が二度残る（内部はexactly-onceなのに）

内部がどれだけ厳密でも、この継ぎ目から重複が漏れる。したがって exactly-once をエンド・ツー・エンドで成立させるには、シンク側で二つのうちいずれかの規律が要る。冪等書き込みか、**トランザクショナルシンク（2相コミット）**である。

冪等な書き込み：同じ結果を二度書いても同じ

冪等（idempotent）とは、同じ操作を何度適用しても結果が変わらない性質をいう。出力を冪等にすれば、再処理で同じ書き込みが二度走っても最終状態は一度と同じになるので、重複が実質的に無害化される。

• 決定的キーで上書きする。追記（INSERT）ではなく、レコードから決まる主キーで UPSERT する。結果キー = f(入力キー, ウィンドウ) のように決定的に決め、同じ入力からは必ず同じキー・同じ値を書く。二度書いても同じ行を同じ値で上書きするだけなので効果は一度。
• キー・バリュー／ドキュメントストアと相性が良い。キー指定の PUT はまさに冪等な上書きだからだ。逆に「カウンタを+1」のような相対更新は冪等でないため、そのままでは再処理で壊れる。

トランザクショナルシンク：2相コミットで出力を状態に縛る

冪等にできない出力——追記ログや、別のメッセージブローカへの publish——では、出力のコミットをチェックポイントの成否に同期させる。Flinkの TwoPhaseCommitSinkFunction や Kafka の トランザクション（transactional.id + read_committed）がこれにあたる。分散トランザクションの2相コミット（2PC）と同じ構造を、チェックポイントのライフサイクルに重ねる。

[事前準備 pre-commit]  チェックポイント中：出力をトランザクション/一時領域へ「まだ見えない」形で書く
        │
        ▼
[チェックポイント完成]  コーディネータが全演算子の状態保存の成功を確認
        │
        ▼
[コミット commit]      チェックポイント確定の通知を受けて、貯めた出力を一括コミット（可視化）

要は、チェックポイントが確定した出力だけを外から見える状態にする。チェックポイント完成前にクラッシュすれば、まだコミットしていない出力トランザクションはアボートされ（あるいは可視化されず）、巻き戻し後に作り直される。逆にチェックポイントが完成していれば出力もコミット済みなので、両者は常に一致する。Kafka では、コンシューマ側を isolation.level=read_committed にすることでアボートされたトランザクションのメッセージを読み飛ばす——これがなければ未コミット出力が漏れ見えて重複する。

レイテンシとの取引：ただではない

エンド・ツー・エンドの exactly-once は無料ではない。トランザクショナルシンクでは、出力が見えるのは次のチェックポイントが完成した後なので、下流から見た遅延は最短でもチェックポイント間隔ぶん増える。間隔を縮めれば遅延は減るが、スナップショットの頻度が上がりオーバーヘッドが増す。逆に冪等書き込みは出力を即座に可視化できるため低遅延だが、出力を決定的な上書きに落とせる場合に限られる。

チェックポイント間隔 大 → オーバーヘッド小・トランザクショナル出力の遅延 大
チェックポイント間隔 小 → 出力遅延 小・スナップショットのコスト（IO/CPU）大

だからまず「本当に exactly-once が要るのか」を問う。多くの集計は下流を冪等にした at-least-once で十分で、そのほうが速く単純だ。厳密なちょうど一度が要るのは、金額・在庫・課金カウントのように一件の重複／欠落が業務上の誤りに直結するところに絞る。配信保証は正確性と遅延・複雑さのトレードオフであり、要件から逆算して最小限を選ぶのが定石だ。分散システムの一貫性の議論（/devops/）や、下流DB側の整合の扱い（/database/）と地続きの設計判断になる。

まとめ

• 配信保証は at-most-once（欠落あり）／at-least-once（重複あり）／exactly-once（欠落も重複もなし） の三段階。ストリームの exactly-once はメッセージ配送回数ではなく、各入力の効果が結果に一度だけ反映される（effectively-once）ことを指す。
• 難しさの根は演算子が状態を持つこと。再開時に「入力オフセット・内部状態・外部出力」を同一時点へ揃えて復元できれば成立する。
• Flink系は バリア方式の分散スナップショット（Chandy-Lamport応用） でチェックポイントを撮る。バリアはデータと共に流れ、各演算子はバリア到達時の状態を保存するので、処理を止めずに整合カットが得られる。複数入力はバリアアライメントで待つ。
• 障害時は完成済みの最新チェックポイントへ全体を巻き戻し、ソースを記録オフセットまで戻す。これで内部状態は各入力を一度だけ反映した値に復元される。
• 内部の保証は外部シンクには及ばない。エンドツーエンドには 冪等書き込み（決定的キーで上書き） か トランザクショナルシンク（2PCでチェックポイント確定と出力コミットを一致、消費側は read_committed） が要る。
• exactly-once は遅延・オーバーヘッドと引き換え。多くの集計は冪等な at-least-once で足り、厳密なちょうど一度は一件のずれが誤りになる箇所に絞るのが定石。