Kafka Streams の Join 完全攻略: Stream-Stream / Stream-Table / GlobalKTable

Kafka Streams の Join は、リアルタイム処理の表現力を大きく左右します。試験(CCDAK)でも出題頻度が高く、実務でも「どの Join を選ぶか」「どのようにキーとパーティションを合わせるか」でパフォーマンスと正確性が決まります。

本稿では、Stream-Stream、Stream-Table、GlobalKTable を比較し、ウィンドウ、タイムスタンプ伝播、再パーティショニング、tombstone など公式挙動に基づく注意点を整理します。

Join の種類と選択基準

Kafka Streams の Join は大きく3つ。イベント同士を一定の時間窓で突き合わせる Stream-Stream、ストリームに最新状態を付与する Stream-Table、分散を意識せず参照できる GlobalKTable です。選択の軸は「時間整合が必要か」「参照は最新1行でよいか」「同一パーティションに揃えられるか」です。

CCDAK 観点では、どの Join がウィンドウ必須か、どの結合種(inner/left/outer)が使えるか、結果レコードのタイムスタンプが何に由来するか、再パーティショニングの必要性と GlobalKTable のトレードオフを問われやすいです。

• Stream-Stream は時間窓が必須。inner/left/outer を選べる
• Stream-Table は最新状態の参照。inner/left のみ。窓は不要
• GlobalKTable は全ノードにフル複製。外部キー参照に最適だがメモリ/ストレージ増

基本スケルトン（共通 Serde とビルダー）

StreamsBuilder builder = new StreamsBuilder();
// 共通の Serde 設定はプロパティ or StreamJoined/Joined で明示
// 以降のセクションで具体的な join を示します。

Stream-Stream Join の実務ポイント

Stream-Stream は両入力をウィンドウにバッファし、同一キーで時間が重なるペアを照合します。JoinWindow は前後幅(before/after)を持ち、状態ストアに保持された相手側レコードとマッチします。到着順が前後しても、ウィンドウ保持内かつ遅延許容範囲にあれば結合されます。

結果レコードのタイムスタンプは2入力の最大値です。join 種は inner/left/outer が選べ、左外部は左側のイベントを基準に相手が無い場合も出力、完全外部はどちらか片方のみでも出力します。内部的には両側に RocksDB の状態ストアが作られ、保持期間はウィンドウ幅（before+after）に遅延許容分を足した値が必要です。

• ウィンドウが必須。幅は業務SLAと遅延分を見込む
• 結果タイムスタンプ = max(left.ts, right.ts)
• キーが null のレコードは結合対象外（再パーティション時に破棄され得る）
• 状態ストア保持期間はウィンドウ幅+遅延許容ぶんを確保
• inner/left/outer を選択可能

Stream-Stream 窓付き結合の概念図

time --->

left (orders):   --- o@t5 ----- o@t8 -------- o@t14 -----
                               [<-- before 2m -->|<-- after 3m -->]
right(payments): ---- p@t6 ---- p@t10 -- p@t13 -------------

例: before=2m, after=3m の場合
- o@t8 は [t6..t11] の支払いと突合 → p@t6, p@t10 が候補
- o@t14 は [t12..t17] の支払いと突合 → p@t13 が候補
結果レコード ts = max(o.ts, p.ts)

KStream-KStream 窓付き結合の例 (Java DSL)

KStream<String, Order> orders = builder.stream("orders");
KStream<String, Payment> payments = builder.stream("payments");

// 5分窓、前2分・後3分（メソッド名はバージョンで異なる場合あり）
JoinWindows win = JoinWindows.of(Duration.ofMinutes(5))
                              .before(Duration.ofMinutes(2))
                              .after(Duration.ofMinutes(3));

KStream<String, Enriched> joined = orders.join(
    payments,
    (o, p) -> Enriched.from(o, p),
    win,
    StreamJoined.with(Serdes.String(), orderSerde, paymentSerde)
);

// left/outer も同様に leftJoin / outerJoin を使用

Stream-Table Join (KStream-KTable)

Stream-Table は「イベント + 最新状態」の付与です。窓は不要で、結合時点のテーブルの最新値を参照します。inner と left が利用可能で、outer はありません。テーブル側の値が null（tombstone）であれば、inner はドロップ、left は右値 null で出力されます。

結果レコードのタイムスタンプはストリーム側のものが引き継がれます。実行時にはテーブルの changelog 由来のローカルストアを参照します。パーティション整合が必要なため、必要に応じてストリーム側に内部再パーティショニングが入ります。

• サポート結合種: inner / left（outer は不可）
• テーブルの null は tombstone。inner はドロップ、left は右 null
• 結果タイムスタンプ = ストリーム側の ts
• キー整合が必要。必要に応じてストリーム側を再パーティション
• KTable のソースは compacted topic が推奨

KStream-KTable の例（ユーザー属性の付与）

KStream<String, PageView> views = builder.stream("pageviews");
KTable<String, UserAttr> users = builder.table("users-changelog");

KStream<String, EnrichedView> enriched = views.leftJoin(
    users,
    (view, user) -> EnrichedView.of(view, user) // user が null の場合も考慮
);

enriched.to("pageviews-with-user");

GlobalKTable Join と外部キー参照

GlobalKTable は、テーブルトピックの全パーティションを各タスクに複製し、ローカルで即時参照できる仕組みです。KStream-GlobalKTable は、ストリームレコードから参照キーを導出するための KeyValueMapper を渡し、外部キー参照（例: 注文の productId から製品カテゴリを引く）を容易にします。

利点は再パーティショニング不要・パーティション不一致でも動作する点。欠点は各タスクが全データを保持するためメモリ/ローカルストレージの負荷が増える点です。大規模ディメンションではサイズ見積もりと RocksDB・changelog の運用を前提に設計します。

• サポート結合種: inner / left
• 参照キーは KeyValueMapper で導出（ストリームキーと一致不要）
• 複製によりメモリ/ストレージ負荷が増える。サイズ上限を考慮
• 結果タイムスタンプ = ストリーム側の ts
• トピックは compacted が望ましい（最新状態を維持）

KStream-GlobalKTable の例（外部キー参照）

KStream<String, Order> orders = builder.stream("orders");
GlobalKTable<String, Product> products = builder.globalTable("products-changelog");

KStream<String, EnrichedOrder> enriched = orders.leftJoin(
    products,
    // ストリームレコードから GlobalKTable の参照キーを導出
    (orderKey, order) -> order.getProductId(),
    (order, product) -> EnrichedOrder.of(order, product) // product が null もあり得る
);

enriched.to("orders-enriched");

再パーティショニング、キー設計、状態保持

Join の多くはキー整合が前提です。Kafka Streams は必要に応じて内部の再パーティショニングトピックを作成し、キーに基づいてデータを並び替えます。これには追加のネットワーク・ストレージコストがかかるため、可能ならソーストピック段階で「同じキー関数・同じパーティション数」に揃えるのがベストです。

Stream-Stream の状態ストアは左右に作られ、保持期間はウィンドウ幅と遅延許容分に依存します。保持が短すぎると遅延イベントが結合されず、長すぎるとストレージを圧迫します。Stream-Table/GlobalKTable はテーブル側が changelog によって常に最新へ更新され、tombstone により削除が伝播します。

• selectKey で明示的にキー整形すると不要な再パーティションを回避しやすい
• 内部再パーティショニングトピックはコンパクション+保持に注意
• ウィンドウ長と遅延許容（grace 設定が可能なバージョンあり）は運用観点で重要
• 巨大テーブルは GlobalKTable より KTable でのコパーティションを優先検討

キー整形と再パーティショニングの明示例

KStream<String, Event> in = builder.stream("input");
// 外部キー join を見据えて、参照キーで再キー化
KStream<String, Event> byRefKey = in.selectKey((k, v) -> v.getRefId());
// 以降の join は byRefKey を使うと内部再パーティションを減らせる場合がある

CCDAK 対策: よく出る論点と落とし穴

試験では、各 Join の適用条件と結果のタイムスタンプ、サポートされる結合種、tombstone の影響、GlobalKTable の特性が頻出です。実装 API 名はバージョン差分があるため、概念と挙動を優先的に押さえてください。

• 窓が必要なのは Stream-Stream のみ。Stream-Table/GlobalKTable は不要
• 結果タイムスタンプ: Stream-Stream は max(left,right)、他はストリーム側
• KStream-KTable は inner/left のみ。outer は不可
• GlobalKTable は再パーティショニング不要だがメモリとローカルストレージを消費
• KTable の null は tombstone。inner では出力されず、left は右 null で出力
• キーが null のレコードは join できない

Serdes/Joined の明示で型・キーを安定化

KStream<String, L> left = builder.stream("left");
KStream<String, R> right = builder.stream("right");
KStream<String, Out> out = left.join(
    right,
    (l, r) -> combine(l, r),
    JoinWindows.of(Duration.ofMinutes(3)),
    StreamJoined.with(Serdes.String(), lSerde, rSerde) // 既定Serdeの誤適用を防ぐ
);

問題で確認

よくある質問