Flink实时作业背压报警，罪魁祸首是数据倾斜——怎么排查和解决？

# 步骤1：打开Flink Web UI（默认端口8081）
# 路径：Jobs → 点进报警的Job → 查看各算子的反压状态
 
# 反压颜色含义：
# 绿色（OK）：无反压，正常
# 黄色（LOW）：轻度反压，需关注
# 红色（HIGH）：严重反压，必须处理
 
# 步骤2：找到反压最严重的算子（通常是keyBy下游的第一个算子）
# 点进这个算子，查看各SubTask的指标：
# - numRecordsInPerSecond：每秒接收记录数
# - numRecordsOutPerSecond：每秒输出记录数
# - currentInputWatermark：水位线进度
 
# 步骤3：对比各SubTask的指标
# 如果某个SubTask的numRecordsIn是其他SubTask的5倍以上，
# 基本可以确认是数据倾斜
 
# 步骤4：查看倾斜SubTask对应的key分布
# 可以临时写一个侧输出流记录每个key的数量

// 场景：统计每个直播间实时在线人数，live_id极度倾斜
// 原始有倾斜的写法：
// stream.keyBy(event -> event.getLiveId())
//        .process(new CountFunction());  // 某个live_id数据量5000倍于其他
 
// 两阶段聚合：第一阶段加随机前缀做局部聚合
DataStream<Tuple2<String, Long>> partialAgg = stream
    .map(event -> {
        // 给live_id加0-9的随机前缀，打散成10份
        int salt = (int)(Math.random() * 10);
        String saltedKey = salt + "_" + event.getLiveId();
        return Tuple2.of(saltedKey, 1L);
    })
    .returns(Types.TUPLE(Types.STRING, Types.LONG))
    .keyBy(t -> t.f0)  // 按saltedKey分区，数据被打散
    .window(TumblingProcessingTimeWindows.of(Time.seconds(5)))
    .sum(1);  // 局部聚合：每5秒统计各打散key的count
 
// 第二阶段：去掉前缀，按真实live_id做全局聚合
DataStream<Tuple2<String, Long>> finalAgg = partialAgg
    .map(t -> {
        // 去掉随机前缀，还原真实live_id
        String realKey = t.f0.substring(t.f0.indexOf('_') + 1);
        return Tuple2.of(realKey, t.f1);
    })
    .returns(Types.TUPLE(Types.STRING, Types.LONG))
    .keyBy(t -> t.f0)  // 按真实live_id汇总
    .window(TumblingProcessingTimeWindows.of(Time.seconds(5)))
    .sum(1);  // 全局聚合：汇总10个打散key的局部结果

方案二：Flink SQL的两阶段聚合写法

Flink SQL里写两阶段聚合，比Java/Scala API更简洁，推荐在支持SQL的场景优先用这种写法：

-- Flink SQL：两阶段聚合处理keyBy倾斜
 
-- 第一阶段：加随机盐做局部聚合（5秒窗口）
CREATE VIEW partial_agg AS
SELECT
  CONCAT(CAST(MOD(HASH_CODE(user_id), 10) AS STRING), '_', live_id) AS salted_key,
  live_id,
  COUNT(*) AS partial_cnt,
  TUMBLE_START(ts, INTERVAL '5' SECOND) AS win_start
FROM live_events
GROUP BY
  CONCAT(CAST(MOD(HASH_CODE(user_id), 10) AS STRING), '_', live_id),
  live_id,
  TUMBLE(ts, INTERVAL '5' SECOND);
 
-- 第二阶段：去掉前缀，按真实live_id汇总
SELECT
  live_id,
  SUM(partial_cnt) AS total_cnt,  -- 汇总10个局部结果
  win_start
FROM partial_agg
GROUP BY live_id, win_start;
-- 效果：原来1个SubTask处理所有该live_id数据 → 10个SubTask并行处理

方案三：热点key分流，单独处理

如果倾斜的key是已知的固定热点（比如官方账号、超头部主播），可以在keyBy之前做分流：热点key走独立的处理链路，普通key走正常链路。

// 分流：热点live_id走侧输出流，普通live_id走主流
OutputTag<LiveEvent> hotTag = new OutputTag<>("hot-lives"){};
 
// 定义热点live_id列表（可以从配置中心动态加载）
Set<String> hotLiveIds = new HashSet<>(Arrays.asList(
    "live_official_001", "live_top_anchor_002"
));
 
SingleOutputStreamOperator<LiveEvent> mainStream = stream
    .process(new ProcessFunction<LiveEvent, LiveEvent>() {
        @Override
        public void processElement(LiveEvent event,
                Context ctx, Collector<LiveEvent> out) {
            if (hotLiveIds.contains(event.getLiveId())) {
                ctx.output(hotTag, event);  // 热点 → 侧输出流
            } else {
                out.collect(event);  // 普通 → 主流
            }
        }
    });
 
// 主流：普通live_id，正常keyBy处理
mainStream.keyBy(e -> e.getLiveId()).process(new CountFunction());
 
// 侧输出流：热点live_id，增大并行度处理，或者专用逻辑
mainStream.getSideOutput(hotTag)
    .keyBy(e -> e.getLiveId())  // 热点key数据量大，给更多并行度
    .process(new HotLiveCountFunction()).setParallelism(20);

三种方案对比

方案	适用场景	延迟影响	代码复杂度	缺点
两阶段聚合	倾斜key多、动态变化	增加一层窗口延迟	中	结果有窗口延迟，两次Shuffle
热点key分流	倾斜key少、相对固定	几乎无额外延迟	高	热点key需要提前知道或动态维护
提高并行度	轻度倾斜	无	低	不能根本消除严重倾斜

实际生产中，我倾向于先用提高并行度试一下轻度倾斜，效果不好再上两阶段聚合。热点key分流保留给倾斜key确定、业务逻辑也需要差异化处理的场景。

面试高频追问

追问一：Flink的两阶段聚合会不会影响Exactly-Once语义？

两阶段聚合本身不影响Exactly-Once，因为它只是改变了数据的分区方式，Checkpoint机制还是一样工作的。但要注意：两阶段聚合里有两层窗口，每层窗口都会有状态，Checkpoint时要保存两层的状态，状态大小会增加。如果倾斜严重、打散的份数很多（比如50份），状态量增加明显，Checkpoint时间会变长。如果Checkpoint时间超过了interval，反而会引发新的问题。建议打散份数不要超过20，同时适当调大Checkpoint间隔。

追问二：Flink 1.15以后有没有自动处理倾斜的机制？

Flink没有像Spark AQE那样的自动倾斜处理机制——Flink是流处理，运行时动态调整分区策略比批处理复杂得多，目前（到Flink 1.18）还没有自动化的倾斜处理能力。Flink社区有讨论过类似机制，但还没落地。所以Flink的倾斜目前必须手动处理。这也是为什么Flink比Spark更考验工程师的经验——很多问题没有AQE这样的自动兜底。

追问三：如果倾斜key是动态的（比如热点直播间随时间变化），热点分流怎么维护？

动态热点key的分流可以结合广播流（Broadcast State）来实现：把热点key列表发到一个广播流，主数据流接收广播流更新本地状态，处理时查本地广播状态判断是不是热点key。这样热点key列表可以动态更新，不需要重启作业。我在某元宇宙社交应用的实时数仓中，就用广播流维护了一个动态的「热点直播间列表」，每5分钟从Redis里拉取最新的Top 20直播间更新广播状态，主数据流做分流时读这个状态，实现了动态热点分流。

实战踩坑

我在某元宇宙社交应用从0搭建Flink实时数仓时，有一个统计直播间实时在线人数的作业。上线一周后，每到晚上黄金时段就触发背压报警，Checkpoint开始超时，最严重的时候作业直接OOM重启。

# 排查过程（按步骤）：
 
# 第一步：看Flink UI，确认是哪个算子在背压
# 发现：keyBy(live_id)下游的count算子，有2个SubTask背压率100%
# 其他28个SubTask背压率0%  → 确认是数据倾斜，不是资源问题
 
# 第二步：查这2个SubTask对应的live_id
# 临时加了一个侧输出流记录每个live_id的事件数
# 发现：某头部主播的live_id，每秒接收事件数是其他直播间的8000倍！
 
# 第三步：短期应急——把并行度从30提到100
# 倾斜key依然集中到少数SubTask，但其他SubTask空转，资源浪费，治标不治本
 
# 第四步：正式方案——两阶段聚合
# 第一阶段按(live_id, salt)分区做5秒窗口局部count
# 第二阶段按live_id汇总
# 上线后：背压消失，Checkpoint时间从平均45秒降到8秒
# 头部主播直播间的在线人数统计延迟从不稳定变为稳定的10秒以内

教训：实时作业上线前，一定要做压测，特别是模拟极端场景——最热的那个key能有多少数据？不要假设「正常情况下不会那么热」，因为热点随时可能出现。我们后来在压测时专门模拟了「头部主播开播」的场景，数据量是平时的1万倍，提前发现并解决了多个潜在的倾斜问题。

总结

Flink倾斜比Spark倾斜更危险，因为流处理倾斜会持续积累引发背压和崩溃。排查时去Flink UI看各SubTask的背压率和数据量，某SubTask独自高负载=数据倾斜。解决方案：两阶段聚合是最通用的，热点key分流适合已知热点场景，动态热点可以结合广播流维护。面试时能说出「Flink倾斜更危险因为持续积累」以及「两阶段聚合在Flink里怎么实现」，就能展示出实际的流处理工程经验。

下一篇聊Spark里最经典的调优场景：MapJoin（Broadcast Join）——小表多大才能广播？广播了真的一定快吗？有哪些反直觉的情况？

你们Flink生产环境遇到过最严重的倾斜是什么场景？最后用了哪个方案解决的？评论区分享～

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