Flink SQL 中的数据倾斜是指数据在分布式计算节点间分布不均，导致部分 Task（Subtask）负载过重，而其他 Task 空闲。这会引发反压、延迟飙升、频繁 GC 甚至作业崩溃（OOM）。

以下是排查与解决 Flink SQL 数据倾斜的系统性方案：

一、 如何排查数据倾斜

在优化之前，首先需要确认是否存在倾斜以及定位倾斜的 Key。

1. 通过 Flink Web UI 监控

• ‌观察指标‌：进入 Job Manager UI，查看具体算子（Operator）的 ‌Subtasks‌。
• ‌判断标准‌：对比不同 Subtask 的 Records In/Out（输入/输出记录数）或 Bytes In/Out。如果某些 Subtask 的数据量远高于平均值（例如相差几倍甚至几十倍），则存在数据倾斜。
• ‌其他迹象‌：
  • 部分 Subtask 出现严重的 ‌Backpressure‌（反压）。
  • 部分 TaskManager JVM 内存占用极高，频繁触发 Full GC。
  • Checkpoint 耗时过长或失败。

2. 通过 SQL 预分析热点 Key

如果怀疑是某个字段（如 user_id, city_code）导致倾斜，可以在源头数据上运行一个简单的聚合查询来找出“热点 Key”。

sql

-- 示例：找出数据量最大的前10个用户ID SELECT user_id, COUNT(*) AS cnt FROM kafka_source_table GROUP BY user_id ORDER BY cnt DESC LIMIT 10;

如果某几个 Key 的数量级远超其他 Key，这些就是导致倾斜的热点 Key。

二、 常见原因与解决方案

数据倾斜通常发生在 GROUP BY、JOIN 或 DISTINCT 操作中。以下是针对不同场景的优化策略：

1. 开启 Flink 内置优化参数（首选方案）

对于大多数聚合场景，Flink 提供了内置的两阶段聚合优化，无需修改 SQL 逻辑即可生效。

• ‌开启 MiniBatch（微批处理）‌：减少状态访问频率，提升吞吐。
• ‌开启 Local-Global Aggregation（局部全局聚合）‌：
  • ‌原理‌：先在本地进行预聚合（Local Agg），减少 Shuffle 阶段传输的数据量，再在全局进行最终聚合（Global Agg）。这能有效缓解热点 Key 带来的网络压力和单点压力。
  • ‌配置‌：

    sql

    SET 'table.exec.mini-batch.enabled' = 'true'; SET 'table.optimizer.agg-phase-strategy' = 'two_phase'; -- 强制使用两阶段聚合

• ‌开启 Distinct Split Optimization（去重拆分）‌：
  • 针对 COUNT(DISTINCT col) 这种极易倾斜的操作，Flink 会自动添加随机前缀进行打散聚合。
  • ‌配置‌：

    sql

    SET 'table.optimizer.distinct-agg.split.enabled' = 'true';

2. Group By / Aggregate 倾斜优化

如果内置参数效果不佳，或者需要更精细的控制，可以采用 ‌手动两阶段聚合（加盐法）‌。

• ‌原理‌给热点 Key 加上随机后缀（Salt），将原本集中在一个 Task 的数据打散到多个 Task 进行局部聚合，然后再去掉后缀进行全局聚合。
• ‌SQL 实现示例‌：

  sql

  -- 第一阶段：局部聚合，通过 RAND() 打散数据 SELECT date_str, category_id, SUM(partial_gmv) AS gmv FROM ( SELECT date_str, category_id, SUM(price) AS partial_gmv FROM source_table GROUP BY date_str, category_id, MOD(HASH_CODE(RAND()), 100) -- 将数据打散成100份 ) GROUP BY date_str, category_id;

3. Join 倾斜优化

Join 倾斜通常发生在一张大表关联一张小表，或者两张大表存在热点 Key 时。

• ‌场景 A：大表 Join 小表（维表关联）‌

  • ‌解决方案‌：使用 ‌Broadcast Join‌。将小表广播到所有 TaskManager，避免大表数据 Shuffle。
  • ‌SQL 提示‌：

    sql

    SELECT * FROM large_table JOIN small_dim_table WITH BROADCAST ON large_table.key = small_dim_table.key;

  • ‌注意‌：确保小表数据量能放入内存。

• ‌场景 B：大表 Join 大表（存在热点 Key）‌

  • ‌解决方案 1：热点 Key 单独处理（分流法）‌
    将热点 Key 和非热点 Key 分开处理。热点 Key 采用特殊的 Join 策略（如增加并行度或加盐），非热点 Key 正常 Join，最后 Union 结果。

    sql

    -- 1. 标记热点 CREATE VIEW split_view AS SELECT *, CASE WHEN key IN ('hot_key_1', 'hot_key_2') THEN 'hot' ELSE 'normal' END AS key_type FROM source_table; -- 2. 分别处理并 Union INSERT INTO result_table SELECT ... FROM normal_data JOIN dim_table ON ... UNION ALL SELECT ... FROM hot_data JOIN dim_table ON ...; -- 可对 hot_data 单独调优

  • ‌解决方案 2：Key 加盐（Salting）‌
    类似 Group By 的加盐法，给 Join Key 加上随机后缀，将大表数据打散，同时将小表数据膨胀（复制多份带不同后缀的记录）进行关联。

4. 数据源与并行度优化

• ‌Kafka Partition 倾斜‌：
  • 如果 Kafka 某些 Partition 数据量极大，会导致对应的 Flink Source Subtask 负载高。
  • ‌解决‌：调整 Kafka Producer 的分区策略，或在 Flink Source 后使用 .rebalance() 或 .rescale() 重新均匀分布数据。
• ‌并行度不匹配‌：
  • 确保 Flink 算子的并行度与上游数据源分区数合理匹配。如果下游并行度过小，会导致多个上游分区数据汇聚到少数下游 Task，引发倾斜。适当增加下游算子的并行度。

5. 空值（Null）或默认值处理

• ‌问题‌：大量数据的 Join Key 或 Group By Key 为 NULL 或空字符串，这些值会被分发到同一个 Task，造成严重倾斜。
• ‌解决‌：在 SQL 中过滤掉无效 Key，或将其转换为随机值以打散分布。

  sql

  -- 过滤 Null 值 WHERE key IS NOT NULL -- 或将 Null 转换为随机值打散 GROUP BY CASE WHEN key IS NULL THEN CONCAT('null_', RAND()) ELSE key END

三、 总结与建议流程

1. ‌监控定位‌：通过 Web UI 确认是否存在倾斜，识别是哪个算子（Source, Join, Agg）出问题。
2. ‌参数调优‌：优先开启 mini-batch 和 two_phase_agg，这是成本最低且效果显著的手段。
3. ‌SQL 改写‌：
   • 若是 COUNT(DISTINCT)，确保开启 distinct split。
   • 若是普通 GROUP BY 倾斜，尝试手动加盐两阶段聚合。
   • 若是 JOIN 倾斜，判断是否可用 Broadcast Join，否则采用分流或加盐策略。
4. ‌资源调整‌：检查并行度设置，确保数据源分区与算子并行度匹配，必要时增加并行度。
5. ‌源头治理‌：如果可能，在数据生产端（如 Kafka Producer）优化分区策略，从根源上避免数据分布不均