62.5 Flink / Kafka Streams
一句话
Flink 是「集群化的流批一体运行时」，Kafka Streams 是「嵌在应用里的流处理库」；选型先看部署边界、状态规模与是否需要多源/复杂事件处理，而不是看谁的 API 更熟。
两种形态，两种边界
| Apache Flink | Kafka Streams | |
|---|---|---|
| 形态 | 独立集群（JobManager + TaskManager） | 普通 Java 库，跑在业务应用进程里 |
| 数据源 | Kafka、JDBC、CDC、文件、MQ、Hive/Iceberg | 主要 Kafka（source/sink 都是 topic） |
| 状态 | Keyed/Operator State，RocksDB/Heap，checkpoint + savepoint | 本地 state store + changelog topic，standby |
| 扩缩容 | key group 重分配，可 rescale | 受 Kafka 分区数约束，rebalance |
| Exactly-once | checkpoint + 2PC sink / Kafka 事务 | Kafka 事务（exactly_once_v2） |
| SQL | Flink SQL / Table API | ksqlDB（独立部署） |
| 运维 | 集群、checkpoint 存储、作业管理 | 应用 + Kafka + 本地 state 目录 |
| 适合 | 多源、复杂 join/CEP、大状态、批流一体 | Kafka 内轻量转换、服务内嵌、团队不想养集群 |
🔑 最本质的差别是部署边界：Flink 是一个「流处理集群」，作业是提交给集群的；Kafka Streams 是「你的服务里的一个库」，生命周期跟应用走。前者适合平台团队统一运维，后者适合业务团队自己拥有。
并行度从哪来：key group vs partition
Flink 的 Keyed State 按 key group 分配：
key --hash--> keyGroup = hash(key) % maxParallelism
keyGroup --range--> subtask = keyGroup / (maxParallelism / parallelism)
maxParallelism 一旦设定（默认 128，或按状态规模估算），扩容只在 key group 的边界上搬，不会把所有 key 重新哈希。
Kafka Streams 的并行度直接等于 Kafka 分区数：
partition = hash(key) % partitions // 生产端决定
task = partition // 一个分区一个 task
Kafka 分区内的顺序与倾斜见 43.3 顺序与分区。
两条约束：
- Flink 的并行度上限是
maxParallelism；Kafka Streams 的上限是主题分区数，分区数不够，加实例也没用。 - 两者都不能拆单个热 key：同一个 key 永远落在同一个 key group/partition。
代码：热点 key 在两种模型下怎么分布
package main
import (
"fmt"
"hash/fnv"
"sort"
)
type Key struct {
Name string
Events int
}
// 一个明显的热点 key（K1 占 60%），其余是长尾。
var keys = []Key{
{"K1", 601}, {"K2", 180}, {"K3", 100}, {"K4", 17},
{"K5", 17}, {"K6", 17}, {"K7", 17}, {"K8", 17}, {"K9", 17}, {"K10", 17},
}
const maxParallelism = 8
func hash(s string) uint32 {
h := fnv.New32a()
_, _ = h.Write([]byte(s))
return h.Sum32()
}
func flinkTask(hv uint32, p int) int {
kg := int(hv % maxParallelism)
return kg / (maxParallelism / p) // key group 连续区间分给 task
}
func kafkaPartition(hv uint32, p int) int { return int(hv % uint32(p)) }
func main() {
fmt.Printf("总事件 %d；Flink maxParallelism=%d；Kafka 分区数随并行度变化\n\n", total(), maxParallelism)
fmt.Printf("%-5s %-7s %-12s %-4s %-8s %-8s %-9s %-9s\n",
"key", "events", "hash", "kg", "flink-p2", "flink-p4", "kafka-p2", "kafka-p4")
for _, k := range keys {
hv := hash(k.Name)
fmt.Printf("%-5s %-7d %-12d %-4d %-8d %-8d %-9d %-9d\n",
k.Name, k.Events, hv, hv%maxParallelism,
flinkTask(hv, 2), flinkTask(hv, 4), kafkaPartition(hv, 2), kafkaPartition(hv, 4))
}
fmt.Println("\n=== 每个 task 的负载（事件数） ===")
for _, p := range []int{2, 4} {
fl := map[int]int{}
kf := map[int]int{}
for _, k := range keys {
hv := hash(k.Name)
fl[flinkTask(hv, p)] += k.Events
kf[kafkaPartition(hv, p)] += k.Events
}
fmt.Printf("并行度 %d（Flink）：", p)
printLoad(fl, p)
fmt.Printf("并行度 %d（Kafka）：", p)
printLoad(kf, p)
}
fmt.Println("\n=== 从 2 扩到 4：哪些 key 换了 task ===")
moved := 0
for _, k := range keys {
hv := hash(k.Name)
a, b := flinkTask(hv, 2), flinkTask(hv, 4)
if a != b {
moved++
fmt.Printf(" %s: task%d -> task%d（%d 条）\n", k.Name, a, b, k.Events)
}
}
fmt.Printf("共 %d/%d 个 key 迁移；K1 仍是单 task 热点（%d 条）\n", moved, len(keys), keys[0].Events)
fmt.Println("\n结论：keyBy/分区决定状态归属；扩容只搬 key group/partition，不拆单个热 key。")
fmt.Println(" 热点 key 的吞吐上限 = 单个 task/partition 的上限，只能靠加盐/两阶段聚合解决。")
}
func total() int {
s := 0
for _, k := range keys {
s += k.Events
}
return s
}
func printLoad(m map[int]int, p int) {
ids := make([]int, 0, len(m))
for id := range m {
ids = append(ids, id)
}
sort.Ints(ids)
for _, id := range ids {
fmt.Printf("task%d=%d ", id, m[id])
}
fmt.Println()
}
本机实测（macOS，Apple M1，Go 1.27.1）：
总事件 1000；Flink maxParallelism=8；Kafka 分区数随并行度变化
key events hash kg flink-p2 flink-p4 kafka-p2 kafka-p4
K1 601 401512545 1 0 0 1 1
K2 180 351179688 0 0 0 0 0
K3 100 367957307 3 0 1 1 3
K4 17 451845402 2 0 1 0 2
K5 17 468623021 5 1 2 1 1
K6 17 418290164 4 1 2 0 0
K7 17 435067783 7 1 3 1 3
K8 17 250513974 6 1 3 0 2
K9 17 267291593 1 0 0 1 1
K10 17 4254713731 3 0 1 1 3
=== 每个 task 的负载（事件数） ===
并行度 2（Flink）：task0=932 task1=68
并行度 2（Kafka）：task0=231 task1=769
并行度 4（Flink）：task0=798 task1=134 task2=34 task3=34
并行度 4（Kafka）：task0=197 task1=635 task2=34 task3=134
=== 从 2 扩到 4：哪些 key 换了 task ===
K3: task0 -> task1（100 条）
K4: task0 -> task1（17 条）
K5: task1 -> task2（17 条）
K6: task1 -> task2（17 条）
K7: task1 -> task3（17 条）
K8: task1 -> task3（17 条）
K10: task0 -> task1（17 条）
共 7/10 个 key 迁移；K1 仍是单 task 热点（601 条）
结论：keyBy/分区决定状态归属；扩容只搬 key group/partition，不拆单个热 key。
热点 key 的吞吐上限 = 单个 task/partition 的上限，只能靠加盐/两阶段聚合解决。
读数：
| 现象 | 数字 | 含义 |
|---|---|---|
| Flink P=2 负载 | 932 / 68 | K1、K2 落在同一 task，热点叠加 |
| Flink P=4 负载 | 798 / 134 / 34 / 34 | 扩容后 K1 仍在一个 task，无法拆 |
| Kafka P=2 负载 | 231 / 769 | 分区哈希把 K2/K3 分到另一侧，热点更集中 |
| 从 2 扩到 4 | 7/10 key 迁移 | 扩容要搬状态；Flink 按 key group，Kafka 按 partition |
| 热 key | K1=601（60.1%） | 单 key 的吞吐上限 = 单 task/partition 上限 |
🔑 扩容只能解决「key 足够多、分布足够散」的倾斜；单个热 key 永远只能落在一个 task。解法是加盐（把 K1 拆成 K1#0..K1#9，两阶段聚合）或把热 key 的结果单独缓存/预聚合，而不是继续加并行度。
状态与 Exactly-once 的实现差异
Flink 的 barrier 对齐与恢复见 62.3 状态管理与容错；2PC 的提交顺序见 62.4 Exactly-once 语义。
| Flink | Kafka Streams | |
|---|---|---|
| 状态存储 | RocksDB/Heap，本地 + checkpoint | RocksDB，本地 + changelog topic |
| 故障恢复 | 从 checkpoint 恢复状态 + 重放 | 从 changelog 重建 state store |
| 一致性 | checkpoint barrier + 2PC sink | Kafka 事务 + 读已提交 |
| 升级 | savepoint | 停应用、改拓扑、重启（state 兼容性要测） |
| 扩缩容 | key group 迁移，状态随 key group 走 | 分区重分配，state 从 changelog 重建 |
⚠️ Kafka Streams 的「状态」不是靠 checkpoint，而是靠 changelog topic：每次 state 更新都写一条 changelog；实例崩溃后，新实例从 changelog 重放重建。changelog 的保留期必须覆盖「最坏恢复时间」，否则状态重建不完整。
延迟从哪来
| 来源 | Flink | Kafka Streams |
|---|---|---|
| 处理 | 算子链、网络 shuffle、序列化 | 同样，但通常进程内 |
| 提交/检查点 | checkpoint 间隔；2PC sink 在 checkpoint 时提交 | commit.interval.ms；EOS 下提交更频繁 |
| 缓存 | 状态后端缓存、网络缓冲 | statestore.cache.max.bytes / record cache |
| 反压 | checkpoint 对齐、慢 sink | rebalance、慢处理 |
💡 我的判断：延迟敏感且只做 Kafka 内转换 → Kafka Streams；需要多源 join、CDC、复杂 CEP、大状态或批流一体 → Flink。 不要因为「Flink 更强大」就把一个 200 行的 Kafka 转换搬进 Flink 集群；也不要因为「Kafka Streams 轻」就让它承担跨 10 个数据源的 join。
生产落地：指名 → 可抄 → 可验 → 可退
指名：Flink（自建或 Ververica / Amazon Managed Flink）；Kafka Streams（Confluent 生态、服务内嵌）；SQL 化用 ksqlDB 或 Flink SQL。
可抄（Flink SQL 1.17+，未验证：本机未安装 Flink）：
CREATE TABLE orders (
order_id STRING,
region STRING,
amount DECIMAL(18,2),
ts TIMESTAMP(3),
WATERMARK FOR ts AS ts - INTERVAL '5' SECOND
) WITH (
'connector' = 'kafka',
'topic' = 'orders',
'properties.bootstrap.servers' = 'localhost:9092',
'format' = 'json'
);
CREATE TABLE gmv_sink (
window_start TIMESTAMP(3),
window_end TIMESTAMP(3),
region STRING,
gmv DECIMAL(18,2),
PRIMARY KEY (window_start, region) NOT ENFORCED
) WITH (
'connector' = 'kafka',
'topic' = 'gmv',
'properties.bootstrap.servers' = 'localhost:9092',
'format' = 'json'
);
INSERT INTO gmv_sink
SELECT window_start, window_end, region, SUM(amount)
FROM TABLE(TUMBLE(TABLE orders, DESCRIPTOR(ts), INTERVAL '1' MINUTE))
GROUP BY window_start, window_end, region;
Kafka Streams（3.x，未验证）：
StreamsBuilder b = new StreamsBuilder();
b.stream("orders", Consumed.with(Serdes.String(), orderSerde))
.groupBy((k, v) -> v.region, Grouped.with(Serdes.String(), orderSerde))
.windowedBy(TimeWindows.ofSizeWithNoGrace(Duration.ofMinutes(1)))
.aggregate(() -> BigDecimal.ZERO,
(k, v, agg) -> agg.add(v.amount),
Materialized.as("gmv-store"))
.toStream()
.map((k, v) -> KeyValue.pair(k.key(), v))
.to("gmv", Produced.with(Serdes.String(), Serdes.serdeFrom(...)));
# Kafka Streams 关键配置
processing.guarantee=exactly_once_v2
commit.interval.ms=100
statestore.cache.max.bytes=10485760
可验：本节程序输出每个 key 的 task 归属与负载；Flink 侧看 Web UI 的 numRecordsIn/Out 与 checkpoint 指标；Kafka Streams 侧看 kafka-consumer-groups --describe 的 lag 与 state 目录大小。可退：Flink 用 savepoint 回滚；Kafka Streams 回滚要保证拓扑兼容（state store 名称/结构不变），否则先停写、重建 state；EOS 出问题可临时降级 at-least-once + 幂等 sink，但要记录「允许重复」的窗口。
反模式
| 反模式 | 后果 |
|---|---|
| Kafka 分区数不足，靠加实例扩容 | 实例闲着，分区仍是瓶颈 |
maxParallelism 设得过大 | key group 太多，小状态作业调度开销大 |
| 热 key 靠加并行度解决 | 单 key 永远在一个 task，扩容无效 |
| Kafka Streams 的 changelog 保留期太短 | 崩溃后状态重建不完整，结果错 |
| Flink 用 Heap 存 TB 级状态 | GC/ checkpoint 超时 |
| 把 Kafka Streams 当多源 ETL | 连接器与状态能力不够，最终自研一堆 |
| 只看吞吐不看 checkpoint/commit 延迟 | 端到端延迟被提交周期拖长 |
| 升级不做 state 兼容性测试 | 恢复失败或状态读错，只能重跑 |
自检
Q1 Flink 与 Kafka Streams 最本质的区别是什么？
▸ 答案：部署边界。Flink 是独立集群上的运行时，作业提交给集群，由平台团队统一运维；Kafka Streams 是嵌入业务应用的库，生命周期跟应用走，状态与 changelog 都依赖 Kafka。前者适合多源、大状态、复杂处理；后者适合 Kafka 内轻量转换、服务内嵌。
Q2 Flink 的 key group 与 Kafka 的 partition 在扩缩容上有什么不同？
▸ 答案：Flink 把 key 映射到固定数量的 key group（maxParallelism），扩容时按 key group 的连续区间重新分配给 subtask，状态随 key group 迁移；Kafka Streams 的并行度等于分区数，扩容时按 partition 重新分配，状态从 changelog 重建。前者扩容上限是 maxParallelism，后者上限是分区数。
Q3 本节 K1 有 601 条事件，把并行度从 2 加到 4 为什么还是热点？
▸ 答案：同一个 key 经过哈希永远落在同一个 key group/partition，扩容只改变「key group/partition 属于哪个 task」，不拆分单个 key。K1 的 601 条仍然在一个 task 上，单 task 的吞吐上限就是 K1 的上限。解法是加盐拆 key（两阶段聚合）或对热 key 单独预聚合/缓存。
Q4 从 2 扩到 4，7/10 个 key 迁移，意味着什么？
▸ 答案：意味着大部分 key 的状态要换 task。Flink 按 key group 迁移状态，Kafka Streams 按 partition 从 changelog 重建；迁移期间会有状态恢复/重分配开销，可能短暂反压。生产上要避免频繁扩缩容，扩之前先确认状态大小与恢复时间。
Q5 Flink 与 Kafka Streams 的 exactly-once 实现有什么不同？
▸ 答案：Flink 用 checkpoint barrier 保存算子状态，配合 2PC sink（或 Kafka 事务）在 checkpoint 完成之后提交；Kafka Streams 用 Kafka 事务把「消费位点 + 输出 + state changelog」放在同一个事务里（exactly_once_v2），下游用 read_committed 隔离。两者都依赖 sink/下游支持事务或幂等。
Q6 Kafka Streams 的 changelog 保留期为什么必须覆盖最坏恢复时间？
▸ 答案：Kafka Streams 的状态不是 checkpoint，而是本地 state store + changelog topic。实例崩溃后，新实例要从 changelog 最早可用的位置重放重建状态；如果 changelog 被过期清理，早期更新丢失，重建出的状态就不完整，结果会错。保留期要 ≥ 最坏恢复时间（含冷启动、积压、扩缩容）。
Q7 什么时候不该用 Flink，而该用 Kafka Streams 或更轻的方案？
▸ 答案：只做 Kafka 内的 filter/map/聚合、状态不大、团队不想维护 Flink 集群时，用 Kafka Streams；只做 stateless 转发/同步时，用 Kafka Connect 或普通 consumer；只有跨多源 join、CDC、复杂 CEP、大状态、批流一体时才值得上 Flink。选型先看「数据源与状态」，再看「团队运维能力」。