数据倾斜与数据膨胀：从长尾任务到成本治理的定位与优化

数据倾斜和数据膨胀这两个词，在很多团队里是被混着用的，但只要真正上手处理过一次，你就会发现它们其实是两种病：一个让你任务卡在 99% 转圈到天亮，一个让你磁盘账单和计算成本悄悄翻几倍，而报表看起来一切正常。我做过几个数据量在 PB 级别的离线链路和实时链路，倾斜和膨胀几乎每隔一段时间就会以新的面貌出现一次，有的是业务埋的雷，有的是自己写 SQL 时留下的坑。这篇就把我对这两类问题的判断方法、定位链路、常见打法和踩过的坑完整讲一遍，适合已经能写 SQL、跑过离线任务，但遇到长尾任务只能靠“加内存”“调并行度”硬顶的同學参考，也适合做数据平台、负责成本治理的同行对照着看。

1. 先分清“倾斜”和“膨胀”：两件常被混为一谈的事

1.1 一个跑得慢，一个跑得贵

数据倾斜的本质是计算资源分配不均。同一批数据被切成了 N 个分片，理想情况下每个分片大小差不多，任务并行跑完；现实是某几个分片分到了几千万条记录，其余分片只有几万条，于是 N-1 个 Task 秒级结束，剩下一两个 Task 跑了四十分钟。表现出来就是 Stage 卡在 99%、GC 时间飙高、大量 spill、最后要么超时失败，要么勉强跑完但拖垮整个调度窗口。

数据膨胀的本质是数据量被非预期地放大。最典型的是 Join 时把一对多关系当一对一用，本来 1 亿行的主表关联上一张一对多明细表，输出变成 8 亿行；再往上叠一层关联，变成 40 亿行。中间结果膨胀之后，shuffle 数据量、落盘数据量、后续所有算子的输入量全部跟着涨，成本是乘法级的。

两者会互相喂养：膨胀出来的中间结果如果 Key 分布还不好，立刻演变成倾斜；倾斜任务为了扛住数据量去加内存加核数，成本继续往上堆。所以治理的时候不能只盯着一个，得同时看。

1.2 三条判断线：怎么快速确认自己遇上的是哪一种

我一般用三条线来判断，不需要看代码，看监控就够：

这里最关键的一条是shuffle read 分布。倾斜是“不均匀”，膨胀是“整体偏大”。我见过不少同学一看到任务慢就去调spark.sql.shuffle.partitions，从 200 调到 2000，结果膨胀型任务反而更慢——分区变多，小文件变多，调度开销上去了，单个 Task 的数据量没降下来。所以动手之前先花五分钟看分布，能省掉半天瞎试。

还有一条经验：倾斜往往伴随数据正确性隐患。当一个 Key 占了全表 30% 的时候，通常意味着这个 Key 是脏数据或者默认值，比如空字符串、-1、unknown、1970-01-01。这类 Key 本身就不该参与业务关联，处理倾斜的同时顺手把数据质量问题修掉，一举两得。

2. 定位倾斜：从任务监控到文件分布的排查链路

2.1 看 Stage 与 Task 的长尾分布

我的习惯是先看 UI 上的 Stage 列表，找那个 duration 明显异常的 Stage，然后点进去看 Task 的 Summary 表。重点看三列：Duration 的 Max 与 75 分位比值、Shuffle Read Size 的 Max 与中位数比值、Spill 的大小。

如果 Max/中位数超过 5 倍，基本可以确认是倾斜；超过 20 倍属于严重倾斜，靠调参数基本救不回来，必须改逻辑。Spill 这一列特别有用：当某个 Task 的 spill 达到几个 G，说明它不光数据多，还放不下，会触发大量磁盘读写，耗时是线性放大的。

再往下看这个 Stage 属于哪种操作。如果是shuffle write之后的聚合类 Stage，说明是 GroupBy 倾斜；如果是 Join 前的 shuffle，说明是关联键倾斜。这一步决定了后面用哪套打法。

我还会顺手看一眼输入数据的分区文件大小分布。有些倾斜根本不是 SQL 的问题，而是上游写文件的时候按某个字段分区，结果某个分区目录下有几千个大文件，另外一个分区只有几个小文件。Task 划分是按文件块来的，文件大小不均直接导致 Task 不均。这种情况改 SQL 是白费力气，得回到写入端治理。

2.2 顺着 Key 分布往回查：SQL 与数据的双向验证

确认了是哪个 Stage、哪类操作之后，下一步是把热点 Key 找出来。最直接的办法是把这个 Stage 的输入表单独拉出来跑一个分布统计：

-- 找出关联键上的 Top 热点 SELECT join_key, COUNT(1) AS cnt FROM dwd_order_detail WHERE dt = '2024-06-01' GROUP BY join_key ORDER BY cnt DESC LIMIT 50;

跑完之后看两件事：Top 1 的占比和Top 50 的累计占比。如果 Top 1 超过 5%，或者 Top 50 累计超过 30%，这个键就有问题。我遇到过一次，Top 1 是个空字符串，占了全表 42%，原因是上游埋点失败时把用户 ID 写成了空串，而关联逻辑没做过滤，全部挤到一个 Reducer 上。

反过来也要验证：拿 SQL 里的关联条件去对一遍，看是不是漏了某个维度条件导致关联范围被放大。有些倾斜是“逻辑写错”伪装出来的，比如本该按天关联，结果条件里没带日期，变成了全历史关联，某个 Key 的历史数据全部涌进来，自然就倾斜了。

2.3 我的排查清单（按顺序走，别乱试参数）

踩过几次坑之后我固化了一张清单，遇到长尾任务按顺序走，基本二十分钟内能定位：

1. 确认是不是倾斜：看 Task 的 Duration 和 Shuffle Read 分布，Max/中位数 > 5 才算。
2. 确认发生在哪个算子：GroupBy、Join、Distinct、窗口函数，四类的解法完全不同。
3. 看输入文件分布：文件大小是否均匀，有没有不可拆分格式（比如某些压缩格式单文件就是一个 Task）。
4. 统计热点 Key：Top N 占比，判断是脏数据还是真实业务热点。
5. 检查 SQL 逻辑：关联条件是否漏了分区过滤、是否多表串联放大了中间结果。
6. 最后才动参数：并行度、广播阈值、AQE 开关。

提示：第 5 步最容易被跳过。我见过一个案例，前四步都做完了，热点 Key 也找到了，打散方案都写好了，最后发现是 where 条件里少写了一个分区过滤，加上之后数据量掉了 90%，倾斜自然消失。改一行 SQL 比改半天代码划算得多。

3. 四类典型倾斜的成因与对应打法

3.1 GroupBy 热点 Key：两阶段聚合与局部预聚合

GROUP BY倾斜是最常见的一种。SELECT user_id, COUNT(1) FROM log GROUP BY user_id这种写法，shuffle 阶段按 user_id 哈希分区，某个大 V 或者某个机器人账号有几千万条日志，全部落到一个分区。

最省事的解法是两阶段聚合。思路很简单：先在 Map 端做一次带随机前缀的局部聚合，把大 Key 拆成 N 份，再在 Reduce 端去掉前缀做最终聚合。

-- 第一阶段：加随机前缀做局部聚合 SELECT concat_ws('_', cast(floor(rand() * 10) as string), user_id) AS salted_key, COUNT(1) AS partial_cnt FROM ods_user_log WHERE dt = '2024-06-01' GROUP BY concat_ws('_', cast(floor(rand() * 10) as string), user_id); -- 第二阶段：去掉前缀做最终聚合 SELECT split(salted_key, '_')[1] AS user_id, SUM(partial_cnt) AS total_cnt FROM tmp_stage1 GROUP BY split(salted_key, '_')[1];

这个方案的成本是多了一次落表和一次扫描。所以我不建议无脑用，只有当热点 Key 明确、且聚合逻辑可以拆分（sum、count、max 可拆，count distinct 和某些百分位统计不可拆）时才用。像collect_list、精确去重这类不可拆的聚合，打散之后还需要额外合并逻辑，写起来麻烦很多。

如果引擎支持，优先让自适应执行框架自动处理。Spark 3.x 的 AQE 开启倾斜处理之后，会在运行时检测到分区过大自动拆分：

spark.sql.adaptive.enabled=true spark.sql.adaptive.skewJoin.enabled=true spark.sql.adaptive.skewJoin.skewedPartitionFactor=5 spark.sql.adaptive.skewJoin.skewedPartitionThresholdInBytes=256MB

我实测下来这套配置能覆盖大部分中等程度的倾斜，省掉不少手写打散的活。但它有边界：对于极端热点（单 Key 占 40% 以上），拆出来的子分区依然很大，还是得回到业务层处理。

3.2 Join 倾斜：小表广播、大表打散、Salted Key

Join 倾斜的处理要先判断一件事：参与关联的两张表，是不是有一张明显小。

如果小表能放进内存（一般经验值 200MB 以内，具体看集群配置），最干净的做法是广播，彻底消灭 shuffle：

SELECT /*+ BROADCAST(dim_shop) */ f.order_id, f.amount, d.shop_name FROM dwd_order f LEFT JOIN dim_shop d ON f.shop_id = d.shop_id;

广播的代价是把小表复制到每个 Executor，如果小表实际有 2G 而你以为只有 50M，就会直接打爆内存。所以我一般会先跑一句SELECT COUNT(1) FROM dim_shop确认行数，再看一眼文件大小。经验上，广播阈值设在 100MB 到 200MB 之间比较稳，超过这个量级就考虑别的方案。

大表对大表、且热点 Key 无法消除时，用打散 + 扩容：给大表的关联键加随机后缀，小表（或热点部分）做 N 倍扩容，然后关联。

-- 大表侧打散 SELECT concat_ws('_', order_id, cast(floor(rand() * 8) as string)) AS join_key, amount FROM dwd_order; -- 维表侧扩容 8 份 SELECT concat_ws('_', shop_id, cast(n as string)) AS join_key, shop_name FROM dim_shop LATERAL VIEW explode(array(0,1,2,3,4,5,6,7)) t AS n;

这个方案的成本是维表膨胀 8 倍、shuffle 数据量增加，属于用空间换均匀。所以 N 的选择要克制，一般 4 到 16 之间，根据热点程度定。我见过有人直接写 100，结果 shuffle 数据量翻了百倍，得不偿失。

3.3 空值与默认值：最容易被忽略的“隐形大Key”

这一类倾斜严格来说不是技术问题，是数据质量问题，但它出现的频率高到值得单独拎出来讲。

关联键为 NULL 或者空字符串时，哈希函数会把它们全部映射到同一个分区。如果主表有 30% 的行关联不上，这 30% 全挤在一起。处理方式分两种，取决于业务语义：

• 这些行本来就不需要：直接过滤掉，加WHERE join_key IS NOT NULL AND join_key <> ''。
• 这些行需要保留但不需要真关联：给空值加随机后缀，让它们均匀散开，反正关联不上维表，结果是 NULL，不影响正确性。

SELECT CASE WHEN join_key IS NULL OR join_key = '' THEN concat('__null__', cast(floor(rand() * 100) as string)) ELSE join_key END AS join_key_safe, ... FROM dwd_order;

注意：给空值加随机后缀这招，只能用在 LEFT JOIN 的场景。如果是 INNER JOIN，直接过滤更干净；如果是需要统计“关联不上的比例”，记得在打散之前先做一次标记字段，别把统计口径搞乱了。

我个人的偏好是先在上游 ETL 层把空值统一转成明确的哨兵值（比如-1），并在数据字典里写清楚，这样下游所有关联都按同一个约定处理，不用每张表都写一遍 CASE WHEN。

3.4 文件级倾斜：不可拆分格式与小文件堆积

有一类倾斜跟 SQL 一点关系都没有，纯粹是存储层的问题。最典型的是用了不可拆分的压缩格式，一个大文件就是一个 Task，如果某个目录下有一个 2G 的不可拆分文件，这个 Task 就要单线程读完 2G。

另一种是小文件堆积：某个分区下有 5000 个几 KB 的文件，Task 启动和调度开销比读数据本身还大。这种情况在流式写入或者频繁小批量写入的场景里特别常见，一天下来目录里能攒几万个文件。

治理手段是合并。离线场景可以在写入前做一次重分区：

# 控制输出文件数量，按目标文件大小估算分区数 spark.sql.shuffle.partitions=400 spark.sql.files.maxPartitionBytes=134217728

maxPartitionBytes设成 128MB 是常见起点，含义是“每个 Task 最多读 128MB”。分区数则按总数据量 / 单分区目标大小反推，比如 200GB 数据、目标单分区 512MB，就是 400 个分区。不要拍脑袋写一个固定值，数据量涨上去之后原来合适的分区数就不合适了。

4. 数据膨胀：Join 放大与重复计算带来的隐性成本

4.1 一对多关联被当成一对一用

膨胀最常见的成因，是写 SQL 的人默认了“关联键在维表里唯一”，实际上并不唯一。主表 1 亿行，维表某个键有 6 条重复记录，关联之后主表这 1 亿行里有对应键的部分全部变成 6 倍。

这种问题的隐蔽性在于，结果看起来是对的。你跑一个COUNT(1)，数字大了一点，但如果没有基准值对比，根本发现不了。等到下游做汇总，指标莫名其妙涨了 20%，才开始往回查。

我的做法是在写关联之前先验证唯一性：

-- 验证关联键是否唯一，结果应该等于 0 SELECT COUNT(1) FROM ( SELECT shop_id, COUNT(1) AS c FROM dim_shop GROUP BY shop_id HAVING c > 1 ) t;

这句 SQL 花不到十秒，但能省掉后面几小时的排查。我在团队里把它做成了规范：所有新增的关联，先把这句验证 SQL 跑一遍，跑完贴在代码评审里。听着笨，但确实管用。

如果维表确实是一对多，而业务只需要一条，那就得明确“取哪一条”：按更新时间取最新、按优先级取最高，用窗口函数去重之后再关联。这一步偷懒，膨胀一定找上门。

4.2 中间结果的膨胀：多表串联时的乘法效应

单次关联膨胀 3 倍还能忍，怕的是串联。A 关联 B 膨胀 3 倍，再关联 C 膨胀 2 倍，再关联 D 膨胀 1.5 倍，最终就是 9 倍。原始 1 亿行变成 9 亿行，中间结果落盘几十个 G，后面每个算子都在为这 9 倍买单。

我治理这类问题的思路是在链条上设检查点。每做完一次关联，落一个临时表或者做一次聚合，把行数压回去，再做下一次关联。多写几步，但每一步的数据量都是可控的。

-- 先在明细层做聚合，把行数压下来 CREATE TABLE tmp_order_agg AS SELECT order_id, shop_id, SUM(amount) AS total_amount, COUNT(1) AS item_cnt FROM dwd_order_item GROUP BY order_id, shop_id; -- 再用聚合后的结果关联维表，避免明细层直接被放大 SELECT a.order_id, a.total_amount, s.shop_name, c.city_name FROM tmp_order_agg a LEFT JOIN dim_shop s ON a.shop_id = s.shop_id LEFT JOIN dim_city c ON s.city_id = c.city_id;

关键是先聚合再关联，而不是先关联再聚合。前者把膨胀挡在源头，后者是在放大了 N 倍的数据上做聚合，成本差一个数量级。这个原则写起来简单，但在实际代码里被违反的概率极高，因为大家写 SQL 的习惯是从明细往汇总一层层 JOIN 上去，很少有人回头把链条重新组织。

4.3 用 EXPLAIN 和行数采样给中间结果“称重”

治理膨胀的前提是能看见它。我常用的两个手段：

第一个是看执行计划里的估算行数。EXPLAIN输出的每个算子后面会带统计信息，虽然估算经常不准，但两个相邻算子的行数比例很有参考价值。如果关联前是 100 万，关联后估算 3000 万，即使估算有 50% 误差，也说明有问题。

第二个是抽样验证。取千分之一的数据，把整条链路跑一遍，对比输入和输出的行数比：

-- 抽样跑一遍，看膨胀倍数 WITH sample AS ( SELECT * FROM dwd_order WHERE dt = '2024-06-01' AND rand() < 0.001 ) SELECT COUNT(1) AS out_cnt FROM sample a LEFT JOIN dim_shop b ON a.shop_id = b.shop_id;

如果抽样输出行数明显大于采样输入行数，膨胀就实锤了。这招的好处是快，几秒钟出结果，比跑全量任务然后等失败强太多。

5. 引擎参数与执行计划层面的调优杠杆

5.1 并行度、分区大小与 shuffle 文件数

并行度是最常被乱调的参数，也是最容易调错的。核心原则是：分区数由数据量决定，不由习惯决定。

一个粗略的估算公式：

目标分区数 = 单次 shuffle 数据量 / 单分区目标大小（建议 128MB ~ 512MB）

比如一次 shuffle 要处理 500GB，目标单分区 256MB，那分区数大约是 2000。如果只设 200，每个分区 2.5GB，一定会 spill；如果设 20000，每个分区 25MB，文件数爆炸，调度开销吃掉收益。

这里有个容易被忽略的副作用：分区数直接影响输出文件数。一个 Stage 2000 个分区，如果直接写出去，就是 2000 个文件。一天跑几十个任务，小文件问题就来了。所以我一般会在最后一步做一次合并，用repartition或者写入端的自适应合并开关，把文件数压到合理范围。

5.2 AQE 相关开关怎么配

自适应查询执行是这几年最有价值的引擎能力，值得单独说。它能做三件事：动态合并 shuffle 分区、动态切换 Join 策略、动态处理倾斜。

我的常规配置大致是这样：

需要强调的是最后两个参数是与的关系，不是或。也就是说分区必须既超过中位数 5 倍、又超过 256MB，才会被拆分。这个设计是为了避免把一堆小分区误判成倾斜拆得稀碎。如果你的场景数据量普遍偏小，可以把阈值往下调，但要谨慎。

AQE 不是万能的。它解决的是“运行时能检测到的倾斜”，对于 Key 分布极端不均、单个 Key 就超过一个分区的场景，拆完之后还是有超大子分区。这时候还是得回到业务层做打散。

5.3 内存与溢写：别把倾斜当内存问题治

我看到过最常见的误操作是：任务 OOM，第一反应是加executor-memory。加完之后任务能跑完了，但耗时没降，资源成本翻倍。

原因很简单：倾斜是分布问题，不是容量问题。你给每个 Executor 都加了内存，但只有那一两个 Task 需要，其余 99% 的 Executor 内存空着。正确的做法是把那个热点拆开，让每个 Task 的数据量接近，然后现有内存就够了。

真正需要加内存的场景，是单个 Task 的数据量确实超过可用内存且无法拆分，比如大表的宽行聚合、超大维度表的广播。这时候加内存是合理的。判断标准也很简单：看 GC 时间和 spill 大小占比，如果 spill 占了 shuffle read 的一半以上，说明确实放不下。

还有一个参数值得关注：spark.memory.fraction和spark.memory.storageFraction。前者控制执行内存和存储内存的总占比，后者控制两者之间的分配。默认值对大多数场景够用，只有在缓存了大量数据又频繁 shuffle 的场景下才需要调。

6. 存储侧的膨胀治理：小文件、分区与压缩

6.1 小文件合并的时机与代价

小文件是膨胀在存储层的表现。一个分区下几万个文件，元数据压力大、读取时打开文件的次数多、Task 调度次数多，三重开销叠加。

合并的时机有两个选择：

写入时合并：在写入前用repartition控制分区数，让输出文件数落在一个合理区间。优点是源头治理，缺点是每次写入都要多一次 shuffle。

离线合并：定时跑一个合并任务，把历史分区的小文件重写一遍。优点是灵活，缺点是需要额外的调度和维护。

我一般两者结合：新数据写入时控制文件数，历史数据按周或按月跑一次合并。合并的时候要注意控制单次处理的数据量，别一次把整个表重写，按分区粒度推进，出问题可以回滚单个分区。

6.2 分区粒度与压缩格式的取舍

分区粒度是个需要权衡的事。分区太粗（比如按月），单分区数据量大，查询时扫描范围大；分区太细（比如按小时），分区数爆炸，元数据压力大，小文件多。

经验值是单分区数据量控制在 1GB 到 10GB 之间，低于 1GB 考虑往上合，高于 10GB 考虑往下拆。当然这要看查询模式，如果业务永远是按天查，那按天分区最合适，哪怕单分区有 20GB，也比分到小时再扫 24 个分区强。

压缩格式方面，优先选可拆分的格式。不可拆分的格式会导致单个大文件只能由一个 Task 处理，天然产生倾斜。列式存储配合可拆分的压缩算法，是离线场景的常规组合。压缩率不是越高越好，解压开销也是成本，尤其在对延迟敏感的链路里。

7. 我在实际项目里踩过的几个坑与固化下来的做法

7.1 打散 Key 引入的额外代价

第一次处理严重倾斜时，我把热点 Key 打散了 100 份，任务从两小时降到二十分钟，很爽。但下个月做成本复盘时发现，这个任务的 shuffle 数据量涨了 30 倍，整个集群的资源消耗反而上升了。

问题出在打散的倍数没有和热点程度匹配。热点 Key 占 20%，打散 10 份就够均匀了，打散 100 份纯属浪费。后来我固化了一个估算方式：打散倍数 ≈ 热点 Key 占比 / 平均占比，再向上取到 2 的幂。比如 Top 1 占 20%，平均占比 0.01%，倍数就是 20，取 16 或 32。

还有一个细节：只打散热点，别打散全量。把 Top 10 的 Key 单独拆出来走一条支路处理，其余数据走正常路径，最后 union 起来。这样绝大多数数据不用承担打散的开销。

7.2 监控与准入：把问题挡在上线前

靠人工发现倾斜和膨胀，永远是被动的。等报表出问题、等任务超时告警，损失已经产生了。所以我在现在的工作里推动了两件事：

第一件是任务级监控。对每个核心任务采集三个指标：Task 耗时的 Max/中位数比值、shuffle read 的 Max/中位数比值、输出与输入行数比。前两个超过 5 告警，第三个超过阈值（按业务定，一般是 1.2）告警。这套指标不用很复杂，但要常态化跑，才能发现渐进式的劣化。

第二件是上线准入。所有新上的关联逻辑，必须提供关联键唯一性验证结果和抽样膨胀倍数。这两项成本很低，但能拦住大部分问题。刚开始有同事觉得繁琐，跑了两周之后，大家发现被拦下来的问题里有一半会导致线上指标异常，就没人抱怨了。

提示：监控指标不要只看单次任务，要看趋势。数据量每天涨 5%，一周之后分区大小就翻倍了，原来合适的配置就不合适了。我一般会在监控里加一条“分区大小周环比”的曲线，涨得快的表提前介入。

最后分享一个我自己常用的习惯：每次处理完一次倾斜或膨胀，我都会在任务注释里写清楚“什么时候、什么数据特征下、用了什么方案、效果如何”。听起来像记流水账，但半年后数据特征变了、老问题重新冒出来的时候，这段注释能省掉重新定位的一大半时间。数据链路是会生长的，今天治好的病，明天换个形态还会回来。