ストリーム結合と状態管理

なぜストリームの結合は「状態」を要求するのか

バッチの結合は簡単です。両テーブルが有界で全件そろっているので、ハッシュ結合なら片方でハッシュ表を作り、もう片方を流して突き合わせれば終わります。ところがストリーム処理が扱うのは終端の来ない無界データです。左ストリームのレコード L が到着した瞬間、それに対応する右ストリームのレコード R は、まだ来ていないかもしれないし、1秒後かもしれないし、10分後かもしれない。全件がそろう「その時」が原理的に存在しません。

だから結合演算子は、片側のレコードを内部状態に保持し、反対側が到着したときに照合するしかありません。これがストリーム結合が本質的に**ステートフル（状態を持つ）**である理由です。無界データを時間で区切る発想は集計と同じで、結合でも「いつまで相手を待つか」を必ず区切ります（/data-engineering/stream-watermarks-windowing/）。区切らなければ、状態は流入したレコードすべてを永久に抱え込み、いずれリソースを食い尽くします。

三つの結合：ストリーム-ストリーム、ストリーム-テーブル、テーブル-テーブル

分散ストリーム処理での結合は、両辺が「変化し続けるイベントの列」なのか「最新値のスナップショット」なのかで、意味論が大きく変わります。ここは**ストリーム（追記される事実の列）とテーブル（キーごとの最新状態）**の双対性が効くところです。

ストリーム-ストリーム結合は両辺がイベントの列で、片方が来たらもう片方の到着を待つため、両辺を状態に溜めます。無界どうしなのでウィンドウが必須です。ストリーム-テーブル結合は、テーブル側をキーごとの最新値として状態にマテリアライズしておき、ストリーム側のレコードが来るたびに現在のテーブル値を引いて結合します。ウィンドウは要りませんが、テーブル側の状態はキー空間ぶん常駐します。このテーブルの実体は、多くの場合変更ログ（changelog）を畳んで作った最新スナップショットであり、CDCで業務DBの変更を流し込んで組み立てるのが定石です（/data-engineering/cdc-log-based/）。

状態ストアとRocksDB：メモリを超える状態をどう置くか

結合の状態は、キー付きのレコード集合を高速に読み書きできる**状態ストア（state store）**に置きます。素直にはJVMヒープ上のハッシュ表ですが、これには限界があります。数千万キー×保持期間ぶんのレコードは容易に数十GBに達し、ヒープに載せればGCが破綻し、OOMで落ちます。

そこで大きな状態を扱う実装は、状態バックエンドにRocksDBを選びます。Kafka Streams は永続的な状態ストアの既定がRocksDBで、Flink も既定こそヒープ上の HashMapStateBackend ですが、大規模状態やHA構成では EmbeddedRocksDBStateBackend が事実上の標準的な選択肢になります。RocksDBはFacebook製の組込みKVストアで、内部構造は**LSMツリー（Log-Structured Merge-tree）**です。書き込みをまずメモリ上の表（memtable）に貯め、満杯になるとディスクへソート済みファイル（SSTable）としてフラッシュし、バックグラウンドのコンパクションで階層をマージしていきます。この構造の利点はストリーム処理と噛み合っています。

• 状態がメモリを超えても動く：ホットなデータはブロックキャッシュとmemtableに載り、それを超えたぶんはローカルディスクへ逃がされる（スピル）。状態量がRAMに収まる前提を外せる。
• 書き込みに強い：ストリーム結合は到着ごとに状態を更新する書き込み主体のワークロード。LSMは追記中心で書き込みスループットが高い。
• 範囲スキャンが速い：キーがソート保持されるため、ウィンドウ結合で「あるキーの、ある時間範囲の相手」を範囲スキャンで引ける。

障害復旧：チェンジログとco-partition

ローカルRocksDBは落ちれば消えるので、状態の更新を別の永続ログへ複製しておき、障害時に読み直して再構築します。Kafka Streams はこれを**チェンジログトピック（changelog topic）**と呼ぶ専用のKafkaトピックで実現します。状態ストアへの各更新（キーと新しい値）を、同じ内容でチェンジログにも書き込むのです。

到着レコード ─▶ 結合演算子 ─▶ RocksDB（ローカル・作業コピー）
                     └────────▶ changelogトピック（Kafka・永続バックアップ）

障害 → 別ノードで、changelogを先頭から再生してRocksDBを再構築 → 処理再開

チェンジログはlog compaction（同一キーは最新値だけ残す圧縮）を有効にでき、キーごとに最新の状態値だけを保持するので、テーブルの最新スナップショットと等価になります。これがまさに「ストリーム（更新の列）とテーブル（最新状態）の双対性」の実装です。Flink では同じ役割をチェックポイント機構が担い、バリア方式の分散スナップショットで状態を整合的に永続化して、障害時にそこへ巻き戻します（/data-engineering/exactly-once-streaming/）。

結合を成立させるもう一つの前提がco-partition（同一分割）です。結合演算子は各パーティションで独立に走り、自分のパーティションに来たレコードしか見ません。したがって結合キーが同じレコードは、両ストリームで必ず同じパーティション番号に来なければ、状態上で出会えず結合が漏れます。これには両辺が (1) 同じ結合キーでパーティション分割され、(2) 同じパーティション数を持つことが要ります。満たさない場合は結合前にキーで再分割（リパーティション、内部的なシャッフル）が入ります。この「同じキーを同じ場所へ集める」設計は、分散データ処理でシャッフルを制御する話と地続きです（/data-engineering/partitioning-bucketing/）。

状態の肥大化とTTL：保持期間で有界に保つ

ステートフルな結合の最大の運用リスクは、状態が際限なく膨らむことです。原因は結合の種類ごとに異なります。

ウィンドウ結合では、保持期間（retention）が状態量を直接決めます。結合ウィンドウを「イベント時刻±5分の相手と結合」と定義すると、演算子は各キーについて過去5分ぶんの相手レコードを状態に抱え続け、ウォーターマークがその範囲を追い越したぶんから期限切れとして掃除します。ここでウィンドウを広げるほど、あるいは遅延データを救うために保持期間を延ばすほど、同時に保持するレコード数が増え、状態が線形に膨らみます。遅延を救う完全性と、状態コストが正面から取引されるのです。

ウィンドウなしの結合（ストリーム-テーブル、テーブル-テーブル）はさらに危険です。テーブル側はキーごとの最新値を保持しますが、キーが二度と現れなくなっても、その状態は自動では消えません。ユーザーIDやデバイスIDのようにキー空間が時間とともに増え続ける（高カーディナリティ）と、状態は単調増加し続けます。これを止めるのが**TTL（Time To Live）**です。

ウィンドウ結合:      保持期間 = 結合ウィンドウ幅 + 許容遅延
                   → ウォーターマークが越えたら期限切れで削除（状態は有界）

ウィンドウなし結合:  TTLを設定しないと状態は単調増加
                   → 最終アクセス/更新から一定時間で状態エントリを退去（eviction）
                   → TTL経過後に来たキーは「相手なし」として扱われる点に注意

TTLは「そのキーが今後アクセスされる見込み」を時間で近似し、期限切れのエントリを退去させて状態量を有界に保ちます。代償は正確性で、TTLを短くしすぎると、まだ有効な相手を消してしまい結合が漏れます。逆に長くすればメモリ・ディスクを圧迫する。ウィンドウの保持期間もTTLも、本質は同じ「いつ相手を諦めるか」という時間の設計であり、完全性・遅延・状態コストの三者を天秤にかける点でウォーターマークの調整と同じ構図です。

まとめ

• ストリーム結合は、相手がいつ来るか分からないため片側を状態に保持して待つ。ゆえに本質的にステートフルで、無界データを扱う以上、ウィンドウと保持期間で「いつまで待つか」を必ず区切る。
• 結合はストリーム-ストリーム（両辺バッファ・ウィンドウ必須）、ストリーム-テーブル（テーブル側を最新値でマテリアライズ・ウィンドウ不要）、テーブル-テーブルの三種で、ストリームとテーブルの双対性が意味論を決める。
• 実務の状態ストアは**RocksDB（LSM系組込みKV）**で、状態がメモリを超えてもローカルディスクへスピルし、書き込み主体・範囲スキャンに強い。
• ローカル状態は揮発するので、更新をチェンジログトピック（Flinkはチェックポイント）へ複製して障害時に再構築する。結合の両辺は同一キー・同一パーティション数でco-partitionしていないと突き合わない。
• 最大の運用リスクは状態の肥大化。ウィンドウ結合は保持期間、ウィンドウなし結合はTTLで状態を有界に保つ。本質は「いつ相手を諦めるか」で、完全性・遅延・状態コストのトレードオフをウォーターマーク同様に明示的に設計する。