数据倾斜处理：KeyBy之后一个SubTask累死

keyBy(userId)：
  SubTask[0] 处理 userId="系统用户" → 1000万条/分钟
  SubTask[1] 处理其他用户 → 1万条/分钟
  SubTask[2] 处理其他用户 → 8000条/分钟

→ SubTask[0]是其他的1000倍，直接反压甚至OOM

原始方案：keyBy(city) → sum(amount)
  北京：1000万条 → 一个SubTask处理

两阶段方案：
  第一阶段：keyBy(city + 随机后缀) → 预聚合
    北京_0：100万条
    北京_1：100万条
    ...
    北京_9：100万条
  
  第二阶段：keyBy(city) → 最终聚合
    北京：10条预聚合结果 → 轻松处理

// 第一阶段：加盐预聚合
stream
    .map(e -> {
        e.setCityKey(e.getCity() + "_" + ThreadLocalRandom.current().nextInt(10));
        return e;
    })
    .keyBy(e -> e.getCityKey())
    .window(TumblingEventTimeWindows.of(Time.minutes(1)))
    .sum("amount")
    // 第二阶段：去盐最终聚合
    .map(e -> {
        e.setCityKey(e.getCity()); // 去掉随机后缀
        return e;
    })
    .keyBy(e -> e.getCityKey())
    .window(TumblingEventTimeWindows.of(Time.minutes(1)))
    .sum("amount");

方案二：预聚合（攒批再发）

在keyBy之前做本地预聚合，减少发往下游的数据量。

// 用Process Function在本地攒1000条或每5秒做一次预聚合
// 把 1000万条/分钟 → 攒成 1万条预聚合结果 → 再keyBy

方案三：过滤热点Key单独处理

// 把热点Key拎出来单独一条链路
DataStream<Event> hotStream = stream.filter(e -> e.isHotKey());
DataStream<Event> normalStream = stream.filter(e -> !e.isHotKey());

// 热点Key用更高并行度或特殊逻辑处理
hotStream.keyBy(...).process(new HotKeyHandler());
normalStream.keyBy(...).sum("amount");

// 最后union
hotStream.union(normalStream);

最终我们用了两阶段聚合。系统用户的数据打散到10个SubTask预聚合，再汇总。延迟从5分钟降到10秒。

怎么发现倾斜？

Flink Web UI → 看每个SubTask的指标：
  - Records Received：如果某个SubTask是其他的100倍 → 倾斜
  - Back Pressured：如果只有个别SubTask反压 → 倾斜
  - Busy Time：如果某个SubTask busy 99%其他20% → 倾斜

面试加分提示

💡 场景+发现+方案：

「我遇到过按user_id做keyBy统计在线时长，一个系统用户的数据量是其他的1000倍。通过Web UI看到一个SubTask的Records Received远大于其他。

用两阶段聚合解决：第一阶段给Key加随机后缀打散到10个SubTask预聚合，第二阶段去掉后缀做最终聚合。延迟从5分钟降到10秒。」

📌 总结

记忆点	内容
倾斜原因	某些Key数据量远大于其他
发现方法	Web UI看SubTask的Records和Busy不均匀
两阶段聚合	加盐预聚合→去盐最终聚合（最通用）
预聚合	keyBy前本地攒批，减少下游数据量
热点Key分离	filter拆分热点和普通，分别处理

▶ 下一篇预告：DAY 9 内存模型与调优

---

Wood聊大数据 | 9年大数据老兵，聊技术、聊职场、聊成长

---

小红书正文

KeyBy之后一个SubTask累死其他SubTask闲着——这就是Flink数据倾斜。

怎么发现？Web UI看SubTask的Records Received和Busy Time。如果某个是其他的100倍就是倾斜。

三种解法：① 两阶段聚合（最通用）——加盐打散预聚合→去盐最终聚合 ② 预聚合——keyBy前本地攒批减少数据量 ③ 热点Key分离——filter拆开单独处理。

我遇到过一个系统用户数据量是其他的1000倍。两阶段聚合后延迟从5分钟→10秒。