Kafka→Flink→Doris 整条实时链路，哪个环节倾斜最难搞？端到端排查思路

# 检查Kafka各Partition的消息量和Lag
# 方法1：用kafka-consumer-groups.sh
kafka-consumer-groups.sh \
  --bootstrap-server kafka:9092 \
  --describe \
  --group flink-consumer-group
# 关注：LAG列（积压消息数），各Partition的LAG差异
# 如果某个Partition的LAG是其他的10倍，说明这个Partition数据量大
# 对应消费这个Partition的Flink SubTask负载高
 
# 方法2：查看各Partition的Offset增长速度
# 在Kafka Manager或Kafka UI里看各Partition的消息增长曲线
# 某个Partition的曲线斜率远大于其他 = 该Partition数据写入量大
 
# 方法3：分析生产者的partition key分布
# 如果生产者按user_id分区，查user_id的消息量分布
# 发现热点user_id → Kafka分区倾斜的根源

// Kafka分区倾斜的处理：修改生产者的分区策略
 
// 问题：生产者按user_id哈希分区，热点user_id消息量暴增
// 原始（倾斜）：
// ProducerRecord<String, String> record =
//     new ProducerRecord<>(topic, userId, message);
// userId作为key，Kafka按key的哈希值决定分区
 
// 方案1：改用随机分区（彻底均匀，但失去顺序性）
ProducerRecord<String, String> record =
    new ProducerRecord<>(topic, null, message);  // key=null → 随机分区
 
// 方案2：加盐分区（保留部分局部性，同时均匀分布）
String saltedKey = userId + "_" + (System.currentTimeMillis() % 10);
// 同一个userId被分到10个不同分区，每次随机一个
ProducerRecord<String, String> record =
    new ProducerRecord<>(topic, saltedKey, message);
 
// 注意：修改分区策略后，如果下游依赖消息的顺序性（如按用户事件时序），
// 需要在Flink端重新按user_id排序，代价是引入了状态和延迟

# 区分「Kafka传导的倾斜」和「Flink自身的倾斜」：
 
# 看Flink UI的Source算子：
# 如果Source算子各SubTask的numRecordsIn差异大 → Kafka分区倾斜传导
# 如果Source算子各SubTask的numRecordsIn接近均匀
#   但下游keyBy算子差异大 → Flink自身keyBy倾斜
 
# 对于Kafka传导的倾斜：
# Source算子的并行度和Kafka Partition数通常是1:1
# （Flink从每个Partition拉数据的SubTask固定，不能重分配）
# 解决：在Source之后立刻做rebalance()或rescale()，打乱数据后再keyBy
 
# Flink中添加rebalance（代价：增加一次网络传输）
DataStream<Event> balanced = source
    .rebalance()   // 轮询方式分发到下游，打散Kafka分区带来的不均匀
    .keyBy(e -> e.getUserId())
    .process(new MyProcessor());

// Flink的rebalance() vs rescale() vs shuffle()
 
// rebalance()：轮询分发，全局均匀，但跨TaskManager传输
// 适合：Kafka分区倾斜传导，需要全局均匀
stream.rebalance();
 
// rescale()：在本地TaskManager内轮询，不跨TaskManager
// 适合：上下游并行度不同（如从4个SubTask分发到8个SubTask），减少网络开销
stream.rescale();
 
// shuffle()：随机分发，接近均匀但不保证轮询
// 适合：快速打散，对均匀性要求不高
stream.shuffle();
 
// 实际工程中：Kafka分区不均匀用rebalance()，
// keyBy本身的倾斜用两阶段聚合（第6篇的方案）

第三段：Doris写入倾斜的排查和处理

Flink写入Doris时，如果数据在Flink端已经有倾斜（某个SubTask数据量大），写入Doris时该SubTask对应的BE节点会承受更大的写入压力，产生写入倾斜：

-- 排查Doris写入倾斜
 
-- 方法1：查看各BE节点的写入速度和Compaction状态
SHOW PROC '/backends';
-- 关注：DataUsedCapacity（各BE数据量），差异大 = BE级写入倾斜
 
-- 方法2：查看各Tablet的数据量
SELECT
  BackendId,
  COUNT(*) AS tablet_cnt,
  SUM(RowCount) AS total_rows,
  MAX(RowCount) AS max_tablet_rows,
  MIN(RowCount) AS min_tablet_rows
FROM information_schema.tablets
WHERE TABLE_NAME = 'your_table'
GROUP BY BackendId;
-- 某个BackendId的total_rows远大于其他 = BE级倾斜
 
-- 方法3：查看Compaction状态
SHOW PROC '/statistic';
-- 如果某个BE的compaction_score持续偏高，说明写入压力大

// Flink写入Doris时控制写入分布
// 使用Flink Doris Connector时，可以设置写入并行度和sink的分桶策略
 
// 方案1：在写入前做数据重分布，让写入均匀
dataStream
    .keyBy(e -> e.getAdId())  // 业务逻辑keyBy
    .process(new AggregationFunction())
    .rebalance()  // 聚合完后rebalance，让写入数据均匀分布
    .addSink(dorisSink);  // 均匀写入各个Doris Tablet
 
// 方案2：增加Doris分桶数，减少每个Bucket的数据量
// 原来32个Bucket，写入集中时某些Bucket过大
// 改为128个Bucket，分散写入压力
// （需要重建表）
 
// 方案3：开启Doris的Stream Load批量写入并控制批次大小
// 避免单次大批量写入打爆某个BE
DorisSinkOptions options = DorisSinkOptions.builder()
    .setStreamLoadProp("max_filter_ratio", "0.01")
    .setBatchSize(50000)  // 每批5万行，控制单次写入量
    .setBatchIntervalMs(5000L)  // 每5秒flush一次
    .build();

端到端排查流程图

检查点	检查方法	倾斜症状	处理方案
Kafka层	kafka-consumer-groups查各Partition LAG	某Partition LAG>>其他	改生产者分区策略或加盐
Flink Source层	Flink UI看Source SubTask的numRecordsIn	Source SubTask数据量不均	Source后加rebalance()
Flink算子层	Flink UI看keyBy下游SubTask的numRecordsIn和背压率	某SubTask背压100%	两阶段聚合或热点分流
Doris写入层	SHOW PROC '/backends'查各BE数据量	某BE数据量>>其他	写入前rebalance，或增加分桶数
Doris查询层	SHOW TABLET查Tablet数据分布	某Tablet数据量>>其他	重建表选对分桶key

面试高频追问

追问一：Flink的rebalance()会破坏消息的顺序性吗？

会。rebalance()是轮询分发，同一个Kafka Partition的消息会被分发到不同的下游SubTask，破坏了Partition内的消息顺序。如果下游逻辑依赖消息顺序（比如需要按用户的事件发生顺序处理），在rebalance()之后keyBy(user_id)又会把同一个用户的消息路由到同一个SubTask，恢复了用户维度的顺序性；但不同Partition来的消息混合后，跨Partition的全局顺序就丢了。实际上大多数流处理场景只关心「同一个用户的事件顺序」，不关心跨用户的全局顺序，所以rebalance()→keyBy()的组合是安全的。如果真的需要全局顺序，只能让Kafka只有一个Partition，但那样就无法并行了。

追问二：Flink写Doris时，如何保证Exactly-Once？倾斜处理会不会影响Exactly-Once？

Flink写Doris的Exactly-Once依赖两阶段提交（2PC）：Flink Checkpoint时，先把数据写入Doris的临时事务（Pre-commit），Checkpoint成功后再Commit，失败则Abort。rebalance()这类数据重分布操作不影响2PC的正确性，因为2PC的边界是以Checkpoint为基准的，不是以Task为基准。加盐或两阶段聚合操作改变了数据的逻辑分组，但只要整个算子链都在同一个Checkpoint的作用域内，最终写入的数据还是一致的。需要注意的是：两阶段聚合引入了额外的窗口状态，这些状态也需要被Checkpoint保存，Checkpoint的数据量和时间都会增加，要适当调大Checkpoint的间隔和超时时间。

追问三：实时链路的倾斜监控，和批处理的监控有什么不同？

批处理的倾斜监控是「事后分析」——作业跑完后看Spark UI的Task分布；实时链路必须是「实时监控」，因为倾斜一旦发生会持续积累，不能等作业跑完再看。实时监控的核心指标：Kafka的Consumer Lag（监控上游数据积压趋势）、Flink的numRecordsIn per SubTask（实时看各SubTask负载）、Flink的Background Pressure Rate（背压率，接近100%就是严重倾斜）、Doris的Compaction Score（写入压力指标）。这些指标都可以接入Prometheus+Grafana建立实时大盘，设置告警阈值（比如Consumer Lag增速超过1万/分钟就告警），做到倾斜发生前就预警，而不是等反压报警了才来救火。

实战踩坑

我在某元宇宙社交应用从0搭建Flink+Doris实时数仓时，有一次线上出现了一个非常诡异的问题：整条链路看起来正常，Kafka Lag稳定、Flink没有明显背压，但Doris里查某个直播间的实时数据，发现数据有几分钟的延迟，而其他直播间的数据实时性很好。

# 排查过程：
 
# 第一步：确认Kafka Lag
# 结果：各Partition的Lag都很稳定，没有积压
# 排除：Kafka分区倾斜
 
# 第二步：看Flink UI各SubTask指标
# 结果：所有SubTask的numRecordsIn和背压率都正常
# 排除：Flink算子倾斜
 
# 第三步：查Doris各BE状态
SHOW PROC '/backends';
# 发现：某个BE的Compaction Score持续偏高（正常<10，这个BE达到了80+）
# 这个BE的写入队列很长，新写入的数据要等待Compaction完成才能被查询到
 
# 根因：该直播间对应的Tablet恰好大部分在这个BE上
# 写入该直播间数据的Flink SubTask负载虽然正常，
# 但写入目标Tablet集中在一个BE，该BE Compaction压力大
# 新写入数据虽然落盘了，但查询时看不到（被Compaction阻塞）
 
# 解决方案：
# 短期：手动触发该BE的Compaction：
ADMIN COMPACT TABLE your_table;
 
# 长期：重建表，增加分桶数（从32增到128），
# 让该直播间的数据分散到更多Tablet和BE

教训：实时链路的倾斜不总是表现为「慢」，有时候表现为「数据看不到」（数据写进去了但查询不到，实时性变差）。排查时不要只看吞吐量指标，还要看存储层的健康状态（Compaction Score、Tablet数据分布）。这次踩坑让我对实时链路的监控体系做了升级：加入了Doris的Compaction Score监控，Score超过50就告警。

总结

实时链路的倾斜排查要端到端：从Kafka分区LAG → Flink Source SubTask数据量 → Flink算子背压率 → Doris BE写入压力，逐层检查，找到最上游的倾斜点优先解决。Kafka分区倾斜改生产者策略；Flink Source不均用rebalance()；Flink算子倾斜用两阶段聚合；Doris写入倾斜重建表加分桶。实时链路的监控必须是实时的，Prometheus+Grafana接入各层指标，设置分级告警，做到防患于未然。

下一篇聊一个容易被忽视的细节：数据类型和编码对倾斜的影响——String类型的key和Int类型的key，哈希分布有什么不同？编码错误会不会引起倾斜？

你们的实时链路有没有出现过「数据延迟但吞吐量正常」这种诡异情况？最后发现是哪个环节的问题？评论区分享～

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