KStream：记录流
KStream 表示无界记录流。每条记录是独立的，相同 key 不会互相覆盖。
KStream<String, String> stream:
("alice", "order-1")
("alice", "order-2") ← 两条记录都保留
("bob", "order-3")
适用：事件日志、点击流、订单流——每条记录都需要处理。
KTable：更新流（物化视图）
KTable 表示changelog 流。相同 key 的新记录会覆盖旧值，相当于数据库表的变化日志。
KTable<String, String> table:
("alice", "active") ← 插入
("bob", "active") ← 插入
("alice", "inactive") ← 更新（覆盖 alice 的旧值）
最终状态：
| key | value |
|---|---|
| alice | inactive |
| bob | active |
适用：用户状态、配置表、最新余额——只关心最新值。
创建 KTable
// 方式 1：直接从 Topic 读取（作为 changelog）
KTable<String, String> table = builder.table("user-status-topic");
// 方式 2：从 KStream 聚合得到
KTable<String, Long> counts = stream
.groupByKey()
.count(Materialized.as("counts-store"));
Changelog Topic
物化后的 KTable 才会把自己的状态变更写进 changelog Topic；builder.table("topic") 直接读源 Topic 时不额外生成 changelog，因为输入 Topic 本身就是按 key 更新的日志。聚合得到的 KTable（如 count()）默认会物化并生成 changelog：
用户修改状态 → 写入 "user-status-topic"
↓
Kafka Streams 应用读取
↓
本地 RocksDB 状态存储（物化视图）
↓
状态变更写入 __changelog Topic（用于恢复）
当 Kafka Streams 实例重启时，从 changelog Topic 恢复本地状态存储，无需重新消费完整输入 Topic。
KStream vs KTable 对比
| 维度 | KStream | KTable |
|---|---|---|
| 语义 | 记录流（append-only） | 更新流（upsert） |
| 相同 key | 保留所有记录 | 只保留最新值 |
| 状态存储 | 无（或有） | 只有物化时才有；table(topic) 不加 Materialized 时不建本地存储 |
| changelog | 可选 | 聚合/物化时自动生成；直接读 Topic 时，输入 Topic 本身就是 changelog |
| 适用 | 事件、日志、订单 | 状态、配置、余额 |
join 语义差异
KStream 和 KTable 的 join 行为不同：
KStream + KStream（Windowed Join）
stream1.join(stream2,
(v1, v2) -> v1 + "-" + v2,
JoinWindows.ofTimeDifferenceWithNoGrace(Duration.ofMinutes(5))
);
必须指定窗口。两条流在窗口内匹配 key 的记录才会 join。
KStream + KTable（非窗口 Join）
stream.leftJoin(table,
(streamValue, tableValue) -> streamValue + "user=" + tableValue
);
不需要窗口。KStream 的每条记录与 KTable 的当前最新值 join。
KTable + KTable（Changelog Join）
table1.join(table2,
(v1, v2) -> v1 + v2
);
两个 KTable 的更新相互触发 join 结果更新。
KStream 转 KTable
KStream<String, String> stream = builder.stream("input");
// 方式 1：groupBy + aggregate
KTable<String, Long> countTable = stream
.groupByKey()
.count(Materialized.<String, Long, KeyValueStore<Bytes, byte[]>>as("count-store")
.withKeySerde(Serdes.String())
.withValueSerde(Serdes.Long()));
// 方式 2：toTable()
KTable<String, String> table = stream.toTable();
Repartition
重新分区不是由 groupByKey() 本身决定的，而是看上游有没有动过 key：
builder.stream("input").groupByKey() → 不插入 repartition
builder.stream("input").map(...).groupByKey() → 插入 repartition
因为 map 可能把 key 换掉，相同 key 已经不在同一分区，
聚合前必须经 repartition Topic 重新按 key 分区
groupBy()（带 key 映射）则一定标记为需要重新分区。
Kafka Streams 自动创建 <application-id>-<topology-node-name>-repartition Topic。不需要手动管理，但要注意它会增加 Kafka 的 Topic 数量和 I/O。
小结
| 操作 | KStream 结果 | KTable 结果 |
|---|---|---|
filter | KStream | KTable |
mapValues | KStream | KTable |
groupByKey | — | KGroupedStream |
count | — | KTable |
join(KStream) | KStream（需窗口） | — |
join(KTable) | KStream | KTable |
下一节 2.3 Kafka Streams DSL 基础 ：filter、map、flatMap、groupBy、reduce、aggregate。
继续阅读
探索更多技术文章
浏览归档，发现更多关于系统设计、工具链和工程实践的内容。