KStream と KTable: ストリームとテーブルの抽象を正しく使い分ける

KStream はイベントの連続（追記ログ）、KTable はキーごとの最新状態（チェンジログ）を表します。両者の違いを理解すると、ジョインや集約、トピック設計で迷いません。

CCDAK では、KStream/KTable の選択、ジョインの時間セマンティクス、トピックのコンパクション、トゥームストーン（null 値）などが頻出論点です。

KStream と KTable の基礎

KStream は、トピックに到着する各レコードをそのまま時間順に扱う抽象です。各レコードは独立しており、ストリーム処理（map、filter、branch、windowed aggregation など）に向きます。KStream は「事実のイベント列」を表現します。

KTable は、キーごとの最新の値を持つテーブルの抽象で、チェンジログ（更新の流れ）から最新像をマテリアライズします。KTable の入力にはログコンパクション対象のトピックが適しています。null 値はトゥームストーンとして扱われ、そのキーの行を削除します。

GlobalKTable は全パーティションを各タスクに複製する読み取り専用のテーブルで、KStream とのキー以外での結合に使えます（全ノードに全データを展開するためサイズに注意）。

• KStream: 追記ログ（append-only）、各レコードは不変のイベント
• KTable: キーの最新値、アップサートと削除（null）で表現
• トピックの cleanup.policy=compact は KTable の正確な最新像に有効
• KTable の更新は同じキーのレコードを上書きし、下流には変更のみを伝播

KStream と KTable の概念的な位置づけ

KStream/KTable の基本的な組み立て（Java, Kafka Streams DSL）

Properties props = new Properties();
props.put(StreamsConfig.APPLICATION_ID_CONFIG, "app-kstream-ktable");
props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
props.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass());
props.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass());

StreamsBuilder builder = new StreamsBuilder();
KStream<String, String> events = builder.stream("events");
// KTable のソースは通常 compacted topic を推奨
KTable<String, String> profiles = builder.table("user-profiles");

KStream<String, String> enriched = events.leftJoin(
    profiles,
    (eventVal, profileVal) -> profileVal == null ? eventVal : eventVal + "|" + profileVal
);

enriched.to("events-enriched");

KafkaStreams streams = new KafkaStreams(builder.build(), props);
streams.start();

ソース、シリアライズ、パーティション設計

KStream/KTable の正確な振る舞いはキーとパーティション整合性に依存します。ジョインや集約では、同一キーが同一パーティションに存在することが前提です。selectKey や groupBy を行うと、内部リパーティション・トピックが自動作成されます。

KTable を正しく最新像として保つには、入力トピックの cleanup.policy=compact を設定し、キーは安定（不変）であることが望まれます。値が null のレコードはトゥームストーンとなり、そのキーが削除されます。

Serdes（シリアライザ/デシリアライザ）はデータの互換性に直結します。デフォルト Serdes を設定しつつ、異なる型を扱う演算では明示的に Serdes を渡してください。

• join/aggregate ではキー整合（同一ハッシュ -> 同一パーティション）が必須
• selectKey/groupBy は内部リパーティションを生成（コストとスループットに影響）
• KTable の入力には compacted topic を推奨（最新像が得やすく、ストレージ効率も良い）
• null 値は削除の意味（トゥームストーン）。スキーマ進化時は null の扱いに注意

パーティション整合とタスク配置

Topic A (3 partitions)          Topic B (3 partitions)
P0 -----> Task 0 <----- P0       P0 -----> Task 0 <----- P0
P1 -----> Task 1 <----- P1       P1 -----> Task 1 <----- P1
P2 -----> Task 2 <----- P2       P2 -----> Task 2 <----- P2

同じキーは同じハッシュ -> 同じパーティション -> 同じ Task でジョイン可能

トピック作成（コンパクション）とキー整形の例

AdminClient admin = AdminClient.create(Map.of("bootstrap.servers", "localhost:9092"));
NewTopic compacted = new NewTopic("user-profiles", 3, (short) 3)
    .configs(Map.of(
        "cleanup.policy", "compact",
        "min.cleanable.dirty.ratio", "0.1",
        "segment.ms", "604800000" // 例: 7日
    ));
admin.createTopics(List.of(compacted));

StreamsBuilder b = new StreamsBuilder();
KStream<String, String> raw = b.stream("pageviews-raw");
KStream<String, String> keyed = raw.selectKey((k, v) -> extractUserId(v));
// キー変更 -> 内部リパーティション（throughで明示も可能）
KStream<String, String> repartitioned = keyed.through("pageviews-by-user");

変換、集約、ウィンドウの基本

KStream はレコード単位の変換（map、filter、branch）に加え、ウィンドウ集約が可能です。ウィンドウ集約の結果は KTable（ウィンドウ付きの最新値）として表現されます。時間セマンティクスはイベント時刻を基本とし、遅延到着は許容期間（grace）内で取り込めます。

KTable は非ウィンドウのキー集約に向き、各更新で下流へ再計算の差分を流します（update 設定）。KTable 同士の結合・集約は、テーブルの変更イベントをトリガに再評価されます。

• KStream の windowedBy は Tumbling/Hopping/Sliding/Sessions を選択可能
• 遅延到着は grace で制御。閉じたウィンドウは更新されない
• KTable の集約は常に最新像（非ウィンドウ）。再計算の差分が流れる
• ウィンドウ集約の出力は KTable（Windowed<K> キー）として表現

ウィンドウ時間線図（Tumbling の例）

time --->
|----[W1]----|----[W2]----|----[W3]----|
イベントは所属ウィンドウに集約され、ウィンドウ境界後に確定（grace を考慮）

イベント時刻ベースのウィンドウ集約（Tumbling）

StreamsBuilder builder = new StreamsBuilder();
KStream<String, Long> events = builder.stream("clicks", Consumed.with(Serdes.String(), Serdes.Long()));
TimeWindows windows = TimeWindows.ofSizeWithNoGrace(Duration.ofMinutes(1)); // 例: grace なし
KTable<Windowed<String>, Long> counts = events
    .groupByKey()
    .windowedBy(windows)
    .count();

counts.toStream().to("clicks-per-minute", Produced.with(WindowedSerdes.timeWindowedSerdeFrom(String.class), Serdes.Long()));

ジョインの選択肢とセマンティクス

KStream-KTable は参照（ルックアップ）型の結合で、ストリーム側レコードの処理時点における KTable の最新値と結合します。時間ウィンドウは不要です。KTable の後続更新は過去の結合結果を遡及しません。

KStream-KStream はウィンドウ必須の結合で、同一キーのレコードがウィンドウ内に現れたときに結合します。左右どちらのストリームにも到着順序や遅延の影響を受けます。

KTable-KTable はテーブル同士の結合で、どちらかのテーブルでキーの値が更新されると結合結果が再計算され、差分が下流に流れます。

• KStream-KTable: ウィンドウ不要、処理時点の最新像で参照
• KStream-KStream: ウィンドウ必須、遅延・順序に敏感
• KTable-KTable: 双方の更新イベントで再評価、差分伝播
• leftJoin/rightJoin/outerJoin の可否と null の扱いを確認（特に CCDAK）

3 種のジョインの直感図

KStream --lookup--> KTable   =>  参照時点の最新値
KStream <==window==> KStream =>  時間窓内で結合
KTable  <==update==> KTable  =>  更新のたびに再計算

KStream-KTable と KStream-KStream の例

StreamsBuilder b = new StreamsBuilder();
KStream<String, String> pageviews = b.stream("pageviews");
KTable<String, String> users = b.table("user-profiles");

// 参照型（ウィンドウ不要）
KStream<String, String> enriched = pageviews.leftJoin(users, (pv, u) -> u == null ? pv : pv + "|user=" + u);
enriched.to("pageviews-enriched");

// ストリーム同士の結合（ウィンドウ必須）
KStream<String, String> clicks = b.stream("clicks");
JoinWindows jw = JoinWindows.ofTimeDifferenceWithNoGrace(Duration.ofSeconds(30));
KStream<String, String> joined = pageviews.join(
    clicks,
    (pv, cl) -> pv + "&" + cl,
    jw
);
joined.to("pv-clicks-joined");

状態管理とマテリアライズ

Kafka Streams は状態ストア（既定では RocksDB ローカル）に集約・結合などの状態を保持し、変更は内部チェンジログ・トピックに複製されます。障害時にはチェンジログから再構築され、スタンバイ・レプリカを有効化するとフェイルオーバ時間を短縮できます。

マテリアライズでストア名を明示すると、インタラクティブクエリや外部キャッシュ統合にも備えられます。発行頻度はキャッシュ設定と commit.interval に影響されます。トランザクションと冪等プロデューサにより、状態更新と出力の一貫性（いわゆる exactly-once 処理セマンティクス）が提供されます。

• 状態ストアはローカルに保持、チェンジログに複製して復元可能
• Materialized.as でストア名・Serdes・キャッシュを指定
• スタンバイレプリカでより速い復旧（コストとトレードオフ）
• processing.guarantee の設定で整合性レベルを選択（環境互換性に合わせる）

状態ストアとチェンジログの関係

マテリアライズと EOS（環境互換に注意）の設定例

Properties props = new Properties();
props.put(StreamsConfig.PROCESSING_GUARANTEE_CONFIG, StreamsConfig.EXACTLY_ONCE_V2); // 環境により EXACTLY_ONCE を選択
props.put(StreamsConfig.CACHE_MAX_BYTES_BUFFERING_CONFIG, 10485760); // キャッシュ 10MB（例）

StreamsBuilder b = new StreamsBuilder();
KTable<String, Long> counts = b.stream("events", Consumed.with(Serdes.String(), Serdes.String()))
    .groupByKey()
    .count(Materialized.<String, Long, KeyValueStore<Bytes, byte[]>>as("event-counts")
        .withKeySerde(Serdes.String())
        .withValueSerde(Serdes.Long())
    );

counts.toStream().to("events-counts");

試験対策チェックポイントと実務の落とし穴

CCDAK では、KStream/KTable の役割、結合ごとの時間セマンティクス、トピックのコンパクション、トゥームストーンの意味、リパーティションの発生条件、状態管理と処理保証の設定が問われます。実務でも同じ点で障害・過負荷・不整合が起きやすいため、設計時に明確化しておきます。

特に KTable での null 値は削除を意味するため、スキーマ進化やデシリアライズ失敗で意図せず null を生成しないように注意してください。キャッシュが有効だと下流への出力が遅延するため、テストでは一時的にキャッシュを無効化して観測性を高める手もあります。

• KStream-KTable はウィンドウ不要、処理時点の参照。KStream-KStream はウィンドウ必須
• KTable 入力には compacted topic を推奨。null は削除（トゥームストーン）
• selectKey/groupBy 後は内部リパーティションが発生しうる
• キャッシュと commit.interval が下流出力頻度に影響
• processing.guarantee を環境に合わせて設定（EOS 利用時はブローカ設定も要件）
• GlobalKTable はサイズ注意（全ノードに複製）

トゥームストーン（null）による削除の流れ

トゥームストーンの送出と観測性向上（テスト用）

Properties p = new Properties();
p.put(StreamsConfig.CACHE_MAX_BYTES_BUFFERING_CONFIG, 0); // テスト時は即時フラッシュ

StreamsBuilder b = new StreamsBuilder();
KStream<String, String> updates = b.stream("user-updates");
// 削除条件を満たすと null を出力 -> compacted topic でトゥームストーン
updates
  .mapValues((v) -> shouldDelete(v) ? null : v)
  .to("user-profiles"); // cleanup.policy=compact 推奨

// KTable 側で null を検知して監視に送る
KTable<String, String> profiles = b.table("user-profiles");
profiles.toStream()
  .filter((k, v) -> v == null)
  .mapValues(v -> "deleted")
  .to("audit-deletes");

問題で確認

よくある質問