“数据倾斜”和“数据膨胀”这两个词凡是在大数据这条线上真正跑过批任务、调过 Spark 或 MR 作业的人几乎都被它们各扇过一巴掌。我印象最深的一次一个本该二十分钟跑完的日增量任务硬是从下午两点卡到晚上九点监控面板上 2000 个 task 里有 1992 个在第九分钟全部 SUCCEEDED剩下 8 个慢悠悠地爬了四个多小时中间还 OOM 挂了一次重跑之后又挂在同一批分区上。那会儿我刚接手这条链路第一反应是加内存、加 executor结果内存翻了一倍任务确实不挂了但耗时只从四小时缩到三小时四十分——问题原封不动地待在那里。后来我才明白数据倾斜不是资源问题是数据分布问题而数据膨胀更隐蔽它不一定会让任务失败却会像人身上的脂肪一样一点一点把整条链路的响应速度、存储成本、甚至下游的取数体验全部拖垮。这两件事经常同时出现一个负责制造极端值一个负责放大绝对量叠加起来就是“某个分区本来就大还被 join 放大了十倍”。这篇内容我把这几年踩过的坑、验证过的处理手法、以及那些文档里不会写的判断标准整理出来偏向实战适合已经能写 SQL、但一遇到长尾任务就不知道从哪儿下手的同学也适合负责链路治理、需要给团队定规范的人参考。1. 先把两个概念掰开它们到底在说什么1.1 数据倾斜的真实画像不是整体慢是“卡在一个点上”很多人对数据倾斜的第一印象是“任务慢”这个描述太笼统会让你在排查时抓错方向。数据倾斜的本质是分布式计算里各个并行单元分到的数据量严重不均表现出来就是长尾绝大多数 task 几分钟内结束极少数 task 拖几个小时。判断它有一个非常直观的信号就是看任务耗时的分布——如果最大值和 P50中位数的比值超过 3 倍就已经值得翻一翻执行计划了超过 10 倍基本可以确诊。用生活里的场景类比超市开了十个收银台理论上人流均匀分流每个台子排五个人。但实际情况是有个台子前面排了二百人,另外九个台子空着。这时候你再加开五个收银台没用因为那二百人还挤在同一个台子前——他们不是按“随机分配到空闲台子”走的而是按“我买的东西属于哪一类”走的。分布式计算里的 shuffle 就是这样hash(key) % partitions决定了每一行数据去哪key 分布均匀就万事大吉某个 key 占了三成数据那个 key 所在的分区就会被压垮。真正痛的地方不只是慢。慢只是表象背后是三个连锁反应第一资源被长期占用一个卡住的任务会让整批队列堵着影响同一队列里其他人的作业第二极端分区会把 executor 的内存撑爆Spill 到磁盘之后性能断崖式下跌再严重就是 OOM 直接失败第三失败重跑还大概率失败在同一个地方因为它不是偶发问题是数据分布决定的必然结果。所以遇到倾斜靠“多跑几次碰运气”是没用的必须动数据或者动计算逻辑。1.2 数据膨胀的另一重痛你以为在算数据其实在造数据数据膨胀这个词在不同团队里有不同叫法有的叫“数据放大”有的叫“行数膨胀率超标”落到具体场景里它有两层意思都得管。第一层是计算过程中的行数放大一行输入经过 join、explode、行列转换之后变成十行甚至一百行输出中间结果落盘体积暴涨下游的每一个环节都要为这个放大买单。第二层是存储层面的物理膨胀小文件堆积、副本冗余、没有生命周期的快照表、压缩比极差的存储格式导致同样的逻辑数据量占用了数倍的物理空间。我经手过一条链路源表大概是两百多 GB中间做了一张明细宽表做完之后变成了 2.4TB。当时没人觉得异常因为“宽表嘛字段多”。直到下游的报表任务从二十分钟变成三个小时才有人回头去看。查下来是两个原因叠加一是关联订单明细的时候商品维表里有重复行同一个商品 ID 因为多次变更留下了多条快照记录没有取最新导致一对多放大二是没有做列裁剪和分区裁剪上游全量扫中间结果全量落。两个问题改完那张表降到 300GB 出头下游任务回到了二十五分钟。这里有个容易被忽略的点数据膨胀往往比倾斜更危险因为倾斜至少会让任务明显变慢你能感知到膨胀很可能是“任务照样按时跑完只是成本翻了几倍”等你发现的时候存储账单和下游抱怨已经堆成山了。所以治理膨胀靠的不是救火是常态化的监控指标。1.3 为什么这两件事总是一起出现数据倾斜和数据膨胀不是两个独立的问题它们在链路上是互相喂养的关系。倾斜会导致某几个分区的数据量远超平均而膨胀会把这个“绝对值”进一步放大——本来最大分区就有 30GB再被 join 放大三倍就是 90GB 压在一个 task 上任何内存配置都扛不住。反过来膨胀也会制造出新的倾斜。举个很典型的场景一张表本身分布还算均匀但它关联了一张有明显热点 key 的维表比如某个默认值、某个测试账号、某个占了大头的渠道 IDjoin 完之后这些 key 的行数被成倍放大原本不算热点的 key 变成了新热点。所以你在排查的时候不能只看源表的分布得看经过每一步转换之后的分布。这也是为什么我一直建议在关键节点落中间表并统计行数行数变化是排查膨胀最直接的证据分布变化是排查倾斜最直接的证据。2. 从现象到根因一次典型事故的完整还原2.1 现象观测先看三个信号别急着改代码我现在的习惯是遇到长尾任务先在监控面板上读三个信号这三条基本能定位问题类型。第一个信号是task 耗时的分布直方图看 P99 和 P50 的比值比值大说明是长尾典型的倾斜特征如果所有 task 都慢那是资源不足或者数据量整体上涨跟倾斜没关系别改错方向。第二个信号是shuffle read size 的分区分布也就是每个 reduce task 读了多少数据。Spark UI 的 Stage 页面里能看到每个 task 的 Shuffle Read Size最大值和平均值差 5 倍以上基本就是倾斜。这个指标比耗时更准确因为耗时会被 GC、磁盘抖动等因素干扰而数据量是实打实的。第三个信号是Spill磁盘溢写和 GC 时间。Spill 大量出现说明单个 task 的内存撑不住它的数据量这是倾斜的直接后果GC 时间占比超过 10% 也说明内存压力大。有这两个信号加持你几乎可以确定问题出在某个分区的数据量上而不是在算子上。提示不要只看任务总耗时这一个数字。总耗时是结果不是原因。原因藏在分布里。2.2 根因定位执行计划、数据分布、日志三条线交叉验证定位倾斜我一般走三条线交叉验证。第一条线是执行计划explain extended或者 Spark UI 的 DAG 图重点看 shuffle 发生在哪个阶段是 group by 的 hash aggregate是 join 的 sort-merge还是 window 函数的 partition by。不同的阶段对应不同的处理手法搞错阶段就是白费劲。第二条线是数据分布采样直接在源表上按 key 做一次 count 排序看 top20 的 key 占了多大比例。这一步有个细节采样的时候要带上实际的分区条件别拿全表跑一个是慢另一个是全表分布和当天增量的分布可能差很远——比如某个大促日某个渠道的订单量暴涨全表看是均匀的当天看是极端倾斜的。第三条线是日志和异常栈。OOM 的堆栈信息里通常能看到是哪个集合撑爆的是UnsafeExternalSortershuffle 排序还是HashMap聚合容器这能反推是聚合阶段还是 join 阶段。还有一种特别隐蔽的情况笛卡尔积。join 条件写漏了、或者关联键的类型不一致导致隐式转换失败、或者用了!、or这种非等值条件优化器会退化成 BroadcastNestedLoopJoin数据量一大直接原地爆炸。这种问题在数据量小的时候完全看不出来上线之后才暴露所以执行计划里看到CartesianProduct或者BroadcastNestedLoopJoin一定要警觉。2.3 为什么“加资源”这个第一反应通常是错的我刚入行时的第一反应也是加资源加内存、加 executor 数量、调大并行度。这个操作有它的合理性——它确实能把 OOM 变成不 OOM让任务先跑通。但它解决的是症状不是病因副作用是三个成本上去了、耗时没明显改善、下次数据量再涨一点又挂。更麻烦的是“调大并行度”这个操作经常是无效的。spark.sql.shuffle.partitions从 200 调到 2000听起来每个分区变小了但如果你的聚合 key 只有几十个比如按省份聚合全国就三十几个 key那 hash 分区数根本不是瓶颈瓶颈在 key 的粒度上——三十几个 key 分到 2000 个分区里还是只有三十几个分区有数据剩下的全空着。这时候调并行度纯属自我安慰。所以我现在的顺序是先确认是不是倾斜再确认倾斜发生在哪个算子再选对应的处理手法最后才是调参数兜底。参数是最后一层保险不是第一层手段。3. 数据分布诊断动手之前先把数据看清楚3.1 三行 SQL 看清 key 的分布比任何猜测都靠谱诊断倾斜最有效的手段就是把分布打出来不需要什么工具几行 SQL 就够。第一行看 top 值select shop_id, count(1) as cnt from dwd_order_detail where dt ${bizdate} group by shop_id order by cnt desc limit 20;第二行看总量和 top 值占比这一步是为了量化“到底多严重”select max(cnt) as max_cnt, avg(cnt) as avg_cnt, sum(cnt) as total_cnt, max(cnt) / sum(cnt) as top1_ratio from ( select shop_id, count(1) as cnt from dwd_order_detail where dt ${bizdate} group by shop_id ) t;第三行看分布形态用分位数判断是“少数几个超大头”还是“整体长尾”select percentile_approx(cnt, 0.5) as p50, percentile_approx(cnt, 0.9) as p90, percentile_approx(cnt, 0.99) as p99, max(cnt) as max_cnt from ( select shop_id, count(1) as cnt from dwd_order_detail where dt ${bizdate} group by shop_id ) t;这三行跑完你心里就有数了。如果是 max 远大于 p99说明是极少数热点 key 造成的处理方法就是热点剥离或者加盐打散如果是 p90 就已经很高、分布整体右偏说明粒度选得不对得从业务模型上换聚合维度。注意统计的时候记得把 null 和默认值单独看一下。很多倾斜的真正元凶是null、-1、unknown、空字符串这类“无效 key”它们匹配不上任何维表却在 group by 里稳稳占据一个巨大分区。3.2 什么程度算倾斜什么程度可以不管这个判断标准很关键因为不是所有不均都需要处理。处理倾斜有成本加盐会让代码变复杂广播会占内存热点剥离要多一次 union如果倾斜程度轻微处理收益还不如维护成本。我给自己定的经验阈值是这样的倾斜程度判断标准处理建议可忽略max / avg 3且最大分区小于 200MB不动观察即可轻度max / avg 在 3 到 5 之间优先调参数和 AQE 自适应暂不改逻辑中度max / avg 在 5 到 20 之间或 top1 占比超过 10%针对性处理加盐或广播重度max / avg 20或 top1 占比超过 30%必须改逻辑热点剥离 加盐组合致命单个分区超过 10GB 或稳定 OOM停机排查通常是笛卡尔积或维表重复这套阈值不是拍脑袋背后是资源账一个 task 处理 200MB 数据用默认 1GB 左右的执行内存完全够多花十分钟你也感觉不到一旦单分区超过 2GBSpill 几乎必然发生性能下降就不是线性的了。3.3 用执行计划定位倾斜发生的阶段分布看完了接下来要知道它是在哪一步炸的。原表分布均匀跑完 join 之后炸了那问题在 join原表就炸那问题在读取或者上游写入。看执行计划的时候重点盯三个位置Exchange hashpartitioningshuffle 节点看它的分区键是什么这个键就是倾斜的载体。SortMergeJoin等值 join 走这个算子倾斜通常来自关联键的分布。HashAggregate聚合算子倾斜来自 group by 的键。Window窗口函数里partition by的字段分布不均同样会倾斜。有一种情况值得单独说多个 shuffle 阶段串联的时候倾斜会被放大。比如第一步 group by 已经倾斜了第二步再 join 一个表那个倾斜的分区会带着更大的数据量进入下一个 shuffle雪球越滚越大。所以排查要从最早出现长尾的那个 stage 开始别盯着最后一个。4. 数据倾斜的处理手法与选型逻辑4.1 空值倾斜最便宜的一刀先砍了再说在所有倾斜类型里null 或者无效默认值造成的倾斜是最好处理的因为这类数据往往在业务上没有意义。处理方式有两种取决于业务要不要保留这些行。如果业务上不需要直接过滤掉一行where shop_id is not null and shop_id ! -1就能解决大半问题。如果业务上需要保留比如要统计“未归属渠道”的订单量那就给这些 null 打上随机后缀让它们分散到不同分区select case when shop_id is null or shop_id -1 then concat(unknown_, cast(floor(rand() * 100) as string)) else shop_id end as shop_id_key, order_id, amount from dwd_order_detail where dt ${bizdate};这样处理后聚合结果会分散成 100 份最后再按业务口径把它们合并回“未知”这一类。这个手法的成本极低几乎不影响性能是排查倾斜时应该第一个尝试的动作。提示我见过不少团队在关联维表之前不过滤 null理由是“null 关联不上会丢失数据”。但你要想清楚关联不上的数据本来在结果里就是 null与其让它集中在一个分区里空转不如提前处理掉。4.2 两阶段聚合加盐打散的标准做法两阶段聚合是处理 group by 倾斜最通用的手法思路很朴素既然一个 key 全砸在一个分区里那就人为地把它拆成 N 份先在局部聚合再去掉盐做全局合并。整个过程分三步走。第一步打盐。给每一行加一个 0 到 N-1 的随机数作为盐值drop table if exists tmp_order_salted; create table tmp_order_salted as select shop_id, cast(floor(rand() * 100) as int) as salt, order_id, amount from dwd_order_detail where dt ${bizdate};第二步按(shop_id, salt)做局部聚合这一步的 key 数量变成原来的 100 倍但每个 key 都很小天然均匀drop table if exists tmp_order_part_agg; create table tmp_order_part_agg as select shop_id, salt, count(1) as order_cnt, sum(amount) as total_amt from tmp_order_salted group by shop_id, salt;第三步去掉盐做全局合并这一步的数据量已经小了 100 倍即使还有倾斜也扛得住insert overwrite table dws_shop_order_1d partition(dt ${bizdate}) select shop_id, sum(order_cnt) as order_cnt, sum(total_amt) as total_amt from tmp_order_part_agg group by shop_id;这里有个非常容易踩的坑如果聚合逻辑里包含count(distinct xxx)直接套用上面的三步会算错。原因是同一行数据只带一个盐但同一个order_id可能出现在不同店铺、不同行的记录里被分到不同的盐分组各自计数之后再求和就重复了。正确处理是先按(shop_id, order_id)去重再去打盐计数drop table if exists tmp_order_dedup; create table tmp_order_dedup as select distinct shop_id, order_id from dwd_order_detail where dt ${bizdate}; -- 之后的打盐、局部计数、全局求和三步照常盐值的数量怎么定我的经验是按“热点 key 的数据量 / 单个 task 可承受的数据量”来估。比如最大 key 有 8GB单个 task 舒服的量级是 200MB那就要拆成 40 份以上取 100 比较稳妥。盐值太少起不到打散效果太多会让第二阶段的 key 数量暴涨反而增加调度开销。100 到 500 这个区间是实践中比较常见的。4.3 Map Join 与广播小表关联的最优解如果倾斜发生在 join 阶段而其中一张表足够小比如几十 MB 的维表最干脆的做法是把它广播到每个 executor让 join 变成 map 端的本地操作彻底消除 shuffle。Spark 里可以显式加 hintselect /* BROADCAST(dim_shop) */ a.order_id, a.amount, b.shop_name, b.city_name from dwd_order_detail a left join dim_shop b on a.shop_id b.shop_id where a.dt ${bizdate};Hive 里的写法是/* mapjoin(b) */。这个方案的好处是彻底、干净没有额外的代码复杂度。但它有三个必须知道的限制。第一广播有阈值超了会静默退化。Spark 的spark.sql.autoBroadcastJoinThreshold默认是 10MB即使你加了 hint如果表大小超过阈值优化器还是可能退化成 SortMergeJoin而且不会有任何报错提示你只会发现任务还是那么慢。所以加完 hint 之后一定要看执行计划里是不是真的出现了BroadcastHashJoin。第二广播表会被复制到每个 executor如果表本身有几百 MB加上 driver 分发和 executor 缓存的开销很容易把内存顶满甚至引发 driver OOM。所以广播只适合真正的小表几十 MB 以内比较安全。第三广播解决的是小表问题解决不了大表之间的倾斜。两个大表关联关联键本身分布不均广播没用还是得回到加盐或者热点剥离。4.4 热点 key 单独剥离把最重的那个拎出来有些场景下热点 key 的数量极少可能就两三个商家、几个测试账号但它们占了全表三成以上的数据。这种情况下与其给全表打盐会波及所有本来正常的 key不如把这几个热点单独拎出来处理剩下的走正常流程最后 union 回去。-- 冷数据正常关联 drop table if exists tmp_result_cold; create table tmp_result_cold as select a.order_id, a.shop_id, b.shop_name, a.amount from dwd_order_detail a left join dim_shop b on a.shop_id b.shop_id where a.dt ${bizdate} and a.shop_id not in (shop_hot_001, shop_hot_002); -- 热数据单独把维表拉成 map在代码侧完成关联 -- 或者给热点数据打盐后关联维表同步复制多份 drop table if exists tmp_result_hot; create table tmp_result_hot as select a.order_id, a.shop_id, b.shop_name, a.amount from ( select order_id, shop_id, amount, cast(floor(rand() * 50) as int) as salt from dwd_order_detail where dt ${bizdate} and shop_id in (shop_hot_001, shop_hot_002) ) a left join ( select shop_id, shop_name, salt from dim_shop lateral view explode(array(0,1,2,3,4,5,6,7,8,9)) t as salt where shop_id in (shop_hot_001, shop_hot_002) -- 实际使用时这里要展开到 0 到 49 ) b on a.shop_id b.shop_id and a.salt b.salt; -- 合并 insert overwrite table dws_order_detail partition(dt ${bizdate}) select * from tmp_result_cold union all select * from tmp_result_hot;这个方案代码量偏大但收益也很直接既解决了热点又不影响正常数据的执行路径。用的时候有个关键前提——热点 key 得是稳定且少量的。如果热点是动态变化的今天这三个明天那五个那维护成本会很高还是老老实实全表打盐更省心。所以上线之前我会先观察一周的 top key 列表确认热点是否固定。另外要提醒一句剥离之后要复查。因为剥离本身会改变数据分布可能把原来的第二梯队 key 顶上来变成新热点union 完之后再跑一次分布统计别嫌麻烦。4.5 参数与引擎层面的兜底AQE 是必开的但不能迷信现代的计算引擎在自适应优化上已经做了不少事情Spark 3.x 的 AQEAdaptive Query Execution就是典型代表。它能在运行时根据实际的 shuffle 统计信息动态合并小分区、拆分大分区、甚至自动处理倾斜 join。这几个参数我基本是必开的spark.sql.adaptive.enabledtrue spark.sql.adaptive.coalescePartitions.enabledtrue spark.sql.adaptive.coalescePartitions.initialPartitionNum2000 spark.sql.adaptive.advisoryPartitionSizeInBytes256MB spark.sql.adaptive.skewJoin.enabledtrue spark.sql.adaptive.skewJoin.skewedPartitionFactor5 spark.sql.adaptive.skewJoin.skewedPartitionThresholdInBytes256MB这几个参数的逻辑是先把初始分区数设大一点让 AQE 有足够的余地做合并advisoryPartitionSizeInBytes控制每个分区期望的大小skewedPartitionFactor是判定倾斜的倍数阈值超过这个倍数的分区会被拆分。但 AQE 不是万能的它有两个明确的边界。第一它只能处理 join 阶段的倾斜group by 阶段的倾斜它管不了第二拆分本身有开销如果倾斜是那种“一个 key 占了 50% 数据”的极端情况拆完还是大只是从一个大块变成几个中等块效果有限。所以我的定位是AQE 作为默认配置常开能兜住大部分轻度倾斜重度的还是得改逻辑。另外并行度这个参数要理解清楚它的作用边界。spark.sql.shuffle.partitions调到几千只在 key 的基数足够大的时候有效。如果你的 group by 维度只有几十个值调多大都没用因为数据只能分到那几十个桶里去。这一点我在前面提过但值得再强调一遍因为这是最常见的无效操作。4.6 从业务模型上根治换粒度、预聚合、改口径前面说的都是计算侧的技巧但真正治本的方案往往在业务侧。我处理过一个特别顽固的倾斜按省份 城市聚合某个城市的订单量是其他城市的几百倍。加盐试过了能缓解但治不了因为那个城市本身就占了全量的四成打散 100 份之后每份还是很大。最后的解法是换粒度——把聚合维度从“城市”下沉到“城市 商圈 小时”粒度细了数据天然分散再往上汇总就很快了。这个思路可以推广成一句话当你在粗粒度上聚合遇到倾斜时先看看能不能在更细的粒度上先聚合一轮。还有一个思路是预聚合。日粒度的任务可以改成小时级增量跑每个小时的数据量小倾斜的绝对影响就小或者在上游就把常用的聚合结果算好存成汇总表下游直接查汇总表绕开大表关联。最后是改口径。有些指标本身的计算方式就注定会倾斜比如“全站去重用户数”这种全局去重的需求如果硬要用一个 job 算天然就是单点。可以考虑用 HyperLogLog 这类近似算法或者用位图Bitmap做交并差牺牲一点精度换取性能。这个决策需要业务方参与不能技术侧单方面拍板。4.7 手法选型对照表别拿锤子找钉子不同倾斜场景适配不同的手法我把常用的组合整理成一张表遇到问题可以直接对号入座场景特征首选手法次选手法不推荐null / 默认值占比高过滤或打随机后缀单独拎出来算加盐没必要小表 join 大表广播 Map Join小表扩容打盐调并行度单 key 占比 10%~30%两阶段加盐聚合热点剥离只调参数单 key 占比超过 30%热点剥离 细粒度预聚合换聚合维度单纯加盐多阶段串联放大从最早的 stage 开始治理中间结果落表只看最后一个 stage全局去重类需求近似算法 / 位图分桶预聚合硬算 count distinct笛卡尔积修 join 条件补等值关联键加内存5. 数据膨胀的防控给链路做“体脂管理”5.1 膨胀的四个主要来源先认清楚再动手数据膨胀看着玄乎其实来源就那么几类认清楚之后治理路径很清晰。第一类是 join 导致的行数放大这也是最常见的一类。根因通常是关联键在维表里不唯一——维表里有重复记录、拉链表没取最新快照、维度表做了 SCD 多版本但关联时没加时间条件。一行事实数据关联上三行维表输出就是三行关联上十行就是十倍。这类问题有个很典型的特征join 前后的行数对不上而且差异是整数倍关系。第二类是 explode 和行列转换。一行里存了一个数组或者一个 JSON展开成多行如果数组长度很大比如一个用户的标签列表有上千个放大倍数非常夸张。这类膨胀有时候是业务必需的但通常可以在展开之前先过滤、先去重、先截断。第三类是多表关联的顺序问题。三张大表 join先关联两个大的得到一个巨型中间结果再关联第三个中间结果的落盘和 shuffle 成本都被浪费了。正确的顺序应该是先做过滤和预聚合把每个表都缩小之后再关联。第四类是写入策略问题。全量覆盖写、快照表不清理、分区粒度过细导致小文件堆积这些属于存储层面的膨胀。我见过一个表按小时分区一天 24 个分区一年 8760 个分区每个分区里十几个小文件光元数据操作就要花好几分钟。5.2 维表去重一个 row_number 解决八成放大问题维表重复导致的行数放大处理起来其实很简单关联之前先用窗口函数取最新一条drop table if exists tmp_dim_shop_latest; create table tmp_dim_shop_latest as select shop_id, shop_name, city_code, shop_level, update_time from ( select shop_id, shop_name, city_code, shop_level, update_time, row_number() over ( partition by shop_id order by update_time desc ) as rn from dim_shop_full ) t where rn 1;这个写法几乎适用于所有“维表有多版本”的场景。有两个细节要注意一是排序字段必须是能唯一确定最新版本的如果update_time有重复要再加一个自增 ID 做次级排序否则结果不稳定二是如果维表本身数据量很大取最新这一步本身也是个大 job最好在维表的生产环节就做好别每次关联前都算一遍。对于拉链表除了去重还要加上时间条件只关联在有效期内start_date biz_date end_date的那条记录这一点漏了同样会放大。5.3 分区裁剪、列裁剪与生命周期三件最基础也最容易被忽略的事这三件事听起来像常识但我见过的膨胀问题里至少有一半跟它们有关。分区裁剪指的是where条件里必须带分区字段而且不能对分区字段做函数运算。where dt 20240101能裁剪where substr(dt, 1, 4) 2024就不能会导致全表扫描。这个坑每年都有人踩。列裁剪指的是只 select 需要的列别图省事写select *。宽表动辄几百个字段全选一遍IO 和内存都要多花几倍。尤其是中间表每一层都select *几层叠下来就是灾难。生命周期指的是分区数据的保留策略。ODS 层留多久、DWD 层留多久、中间临时表留多久得有明确规范并且用自动化的清理任务去执行。我一般建议临时表只保留 7 天中间层 90 天到 180 天明细层按业务要求冷数据归档到低频存储。没有生命周期管理的数仓膨胀是必然的只是时间问题。5.4 小文件治理与存储格式选择小文件是存储膨胀里最隐蔽的一类。它的问题不在于总容量大而在于文件数量多导致元数据操作和任务调度开销暴涨。一个分区里有一千个几 KB 的文件读的时候要开一千次写的时候要准备一千个 buffer性能比读一个 100MB 的文件差一个数量级。治理手段有两层。第一层是写入时控制Spark 里开spark.sql.adaptive.coalescePartitions.enabled和spark.sql.adaptive.advisoryPartitionSizeInBytesHive 里调hive.merge.mapfiles、hive.merge.smallfiles.avgsize、hive.merge.size.per.task让引擎在写入前自动合并。第二层是定期做合并任务把历史分区的小文件重写一遍。存储格式上优先选列式存储加压缩Parquet 或 ORC 都行压缩算法用 snappy 兼顾速度用 zstd 压缩率更高但 CPU 开销大一些。行式存储比如 TextFile、JSON在分析场景下几乎是自找麻烦压缩比差、列裁剪失效能换就换。治理动作影响维度典型收益落地难度维表 row_number 去重计算膨胀行数下降 30%~80%低加分区条件扫描量减少 90% 以上的无效读低去掉 select starIO 和内存减少 30%~60% 的读数据量低中间表设 TTL存储成本存储费用下降明显中小文件合并调度开销任务启动时间缩短数倍中换列式存储全链路压缩比提升 3~5 倍高6. 能直接抄作业的配置与实操清单6.1 关键参数清单与调优逻辑参数永远服务于场景脱离场景的“最佳配置”是不存在的。我把常用的参数和它的作用边界整理如下用的时候对照自己的情况取用参数作用常见取值什么时候别用spark.sql.shuffle.partitionsshuffle 分区数数据量 GB 数 × 2~3key 基数小于分区数时无效spark.sql.adaptive.enabled开启动态优化true基本没有理由关spark.sql.adaptive.skewJoin.enabled自动拆分倾斜分区true极端热点 key 效果有限spark.sql.autoBroadcastJoinThreshold广播表大小阈值10MB~100MB调太大容易 driver OOMspark.sql.adaptive.advisoryPartitionSizeInBytes目标分区大小128MB~256MB分区太小会增加调度开销hive.exec.reducers.bytes.per.reducer每个 reducer 处理量256MB~1GB调太小会产生大量小文件hive.merge.smallfiles.avgsize小文件合并阈值128MB对已存在的历史分区无效有个容易忽略的点这些参数的生效顺序是有先后的。AQE 的合并和拆分发生在运行时它依赖初始的分区数如果你把初始分区数设得太小比如 200AQE 连拆的余地都没有。所以正确做法是初始分区数适当放大让 AQE 有操作空间而不是指望一个参数解决所有问题。6.2 一套完整的加盐聚合模板把前面的步骤串起来给一份可以直接改字段名就用的模板。假设场景是订单明细表按shop_id聚合存在严重倾斜。-- step 0: 分布探测只在排查阶段跑别写进日常任务 select shop_id, count(1) as cnt from dwd_order_detail where dt ${bizdate} group by shop_id order by cnt desc limit 20; -- step 1: 打盐盐值数量按热点数据量估算这里取 100 drop table if exists tmp_order_salted; create table tmp_order_salted as select shop_id, cast(floor(rand() * 100) as int) as salt, order_id, amount from dwd_order_detail where dt ${bizdate} and shop_id is not null and shop_id not in (-1, unknown); -- step 2: 局部聚合 drop table if exists tmp_order_part_agg; create table tmp_order_part_agg as select shop_id, salt, count(1) as order_cnt, sum(amount) as total_amt from tmp_order_salted group by shop_id, salt; -- step 3: 全局合并 insert overwrite table dws_shop_order_1d partition(dt ${bizdate}) select shop_id, sum(order_cnt) as order_cnt, sum(total_amt) as total_amt from tmp_order_part_agg group by shop_id; -- step 4: 清理临时表 drop table if exists tmp_order_salted; drop table if exists tmp_order_part_agg;这套模板在大多数批处理引擎上都能跑字段名和表名换一下就行。要注意的是 step 2 和 step 3 之间最好落一次盘虽然多了一次 IO但换来的是可观测性——你能看到中间结果的行数和分布出问题时能快速定位是哪一步炸的。6.3 上线前的验证动作别跳过改完逻辑直接上线这是最容易翻车的地方。我给自己定的上线检查有四项。第一结果正确性比对。用新旧两套逻辑跑同一天的数据把结果全字段比对差异行数必须为 0。注意这里要特别关注 null 值和边界值加盐处理最容易在这些地方出问题。第二性能对比。不只是看总耗时要看 task 耗时的分布。如果总耗时降了但长尾还在说明只是数据量变小了倾斜没根治下次数据量涨回来还会犯。第三中间表体积对比。改完之后中间结果的行数和体积应该明显下降如果没降反升说明某个环节的处理方式不对。第四回归下游。倾斜和膨胀的治理经常涉及口径微调比如 null 的处理方式变了、去重逻辑变了这些变化会传导到下游指标上必须提前跟下游对齐别等报表出数了才发现对不上。7. 常见问题速查与踩坑心得7.1 典型问题速查表现象最可能的原因排查动作处理方式少数 task 跑几小时shuffle 分区数据不均看 shuffle read size 分布加盐或热点剥离广播 hint 加了没效果表超过阈值静默退化看执行计划是否为 BroadcastHashJoin调阈值或改用其他手法加盐之后结果变多了count distinct 被打散重复计数比对去重前后的行数先 distinct 再打盐任务不慢但存储涨得快全量覆盖写或分区过多看分区数和文件数设 TTL、合并小文件join 后行数暴涨十倍维表重复或拉链未取最新检查维表主键唯一性row_number 取最新调大并行度毫无改善key 基数太小看 group by 字段基数换方案别调参数OOM 稳定复现在同一 stage单分区数据量超内存看 spill 和 GC 指标两层处理先降水再抗压早上跑得快下午跑得慢队列资源竞争看资源池排队情况错峰调度或独立队列7.2 几条从坑里爬出来的经验第一条先测分布再动手别凭感觉改。我早期最大的毛病就是看到任务慢就直接加盐结果加了半天发现倾斜根本不在聚合阶段而是在前面的 join 上白忙一场。现在我改任何倾斜逻辑之前一定要先跑分布统计和执行计划确认清楚了再动手这个习惯省下来的时间远超测分布花的那几分钟。第二条临时表不要省。很多人为了“少一次 IO”把中间步骤写成嵌套子查询觉得性能更好。实测下来多写一次临时表反而更容易排查而且因为中间结果有统计信息优化器的执行计划往往更准。尤其是治理期可观测性比那点 IO 开销重要得多。第三条治理要留痕。每次调整都把“原始分布、处理手法、调整后分布、耗时变化”记下来形成一个表格。我维护了这样一份记录一年下来积累了几十条后来遇到新问题基本能直接在里面找到相似案例比翻文档快得多。第四条别一次性改太多变量。优化的时候容易贪心一次改五处结果效果好不知道是哪一处起的作用效果差也不知道是哪一处搞坏的。我现在的做法是一次只动一个点跑完对比一次确认有收益再动下一个。第五条对“看起来正常”的分布保持警惕。有一些倾斜是业务特性决定的比如某个大商家、某个默认渠道它们的量大是合理的不该无脑去掉。处理之前先跟业务确认这些大头是不是应该存在别把正常业务数据当异常清理了。最后分享一个我一直在用的小习惯每个核心任务旁边挂一个“体检 SQL”每天跑一次输出三个数——最大分区占比、中间表膨胀率、单分区最大体积。这三个数超过阈值就告警。这套机制帮我提前发现了不少问题比等任务挂了再查要从容得多。 
长河君 · 2026/10/26 22:01:13 · 阅读 1.2k 数据倾斜与数据膨胀治理：从分布诊断到Spark/MR实战优化
“数据倾斜”和“数据膨胀”这两个词凡是在大数据这条线上真正跑过批任务、调过 Spark 或 MR 作业的人几乎都被它们各扇过一巴掌。我印象最深的一次一个本该二十分钟跑完的日增量任务硬是从下午两点卡到晚上九点监控面板上 2000 个 task 里有 1992 个在第九分钟全部 SUCCEEDED剩下 8 个慢悠悠地爬了四个多小时中间还 OOM 挂了一次重跑之后又挂在同一批分区上。那会儿我刚接手这条链路第一反应是加内存、加 executor结果内存翻了一倍任务确实不挂了但耗时只从四小时缩到三小时四十分——问题原封不动地待在那里。后来我才明白数据倾斜不是资源问题是数据分布问题而数据膨胀更隐蔽它不一定会让任务失败却会像人身上的脂肪一样一点一点把整条链路的响应速度、存储成本、甚至下游的取数体验全部拖垮。这两件事经常同时出现一个负责制造极端值一个负责放大绝对量叠加起来就是“某个分区本来就大还被 join 放大了十倍”。这篇内容我把这几年踩过的坑、验证过的处理手法、以及那些文档里不会写的判断标准整理出来偏向实战适合已经能写 SQL、但一遇到长尾任务就不知道从哪儿下手的同学也适合负责链路治理、需要给团队定规范的人参考。1. 先把两个概念掰开它们到底在说什么1.1 数据倾斜的真实画像不是整体慢是“卡在一个点上”很多人对数据倾斜的第一印象是“任务慢”这个描述太笼统会让你在排查时抓错方向。数据倾斜的本质是分布式计算里各个并行单元分到的数据量严重不均表现出来就是长尾绝大多数 task 几分钟内结束极少数 task 拖几个小时。判断它有一个非常直观的信号就是看任务耗时的分布——如果最大值和 P50中位数的比值超过 3 倍就已经值得翻一翻执行计划了超过 10 倍基本可以确诊。用生活里的场景类比超市开了十个收银台理论上人流均匀分流每个台子排五个人。但实际情况是有个台子前面排了二百人,另外九个台子空着。这时候你再加开五个收银台没用因为那二百人还挤在同一个台子前——他们不是按“随机分配到空闲台子”走的而是按“我买的东西属于哪一类”走的。分布式计算里的 shuffle 就是这样hash(key) % partitions决定了每一行数据去哪key 分布均匀就万事大吉某个 key 占了三成数据那个 key 所在的分区就会被压垮。真正痛的地方不只是慢。慢只是表象背后是三个连锁反应第一资源被长期占用一个卡住的任务会让整批队列堵着影响同一队列里其他人的作业第二极端分区会把 executor 的内存撑爆Spill 到磁盘之后性能断崖式下跌再严重就是 OOM 直接失败第三失败重跑还大概率失败在同一个地方因为它不是偶发问题是数据分布决定的必然结果。所以遇到倾斜靠“多跑几次碰运气”是没用的必须动数据或者动计算逻辑。1.2 数据膨胀的另一重痛你以为在算数据其实在造数据数据膨胀这个词在不同团队里有不同叫法有的叫“数据放大”有的叫“行数膨胀率超标”落到具体场景里它有两层意思都得管。第一层是计算过程中的行数放大一行输入经过 join、explode、行列转换之后变成十行甚至一百行输出中间结果落盘体积暴涨下游的每一个环节都要为这个放大买单。第二层是存储层面的物理膨胀小文件堆积、副本冗余、没有生命周期的快照表、压缩比极差的存储格式导致同样的逻辑数据量占用了数倍的物理空间。我经手过一条链路源表大概是两百多 GB中间做了一张明细宽表做完之后变成了 2.4TB。当时没人觉得异常因为“宽表嘛字段多”。直到下游的报表任务从二十分钟变成三个小时才有人回头去看。查下来是两个原因叠加一是关联订单明细的时候商品维表里有重复行同一个商品 ID 因为多次变更留下了多条快照记录没有取最新导致一对多放大二是没有做列裁剪和分区裁剪上游全量扫中间结果全量落。两个问题改完那张表降到 300GB 出头下游任务回到了二十五分钟。这里有个容易被忽略的点数据膨胀往往比倾斜更危险因为倾斜至少会让任务明显变慢你能感知到膨胀很可能是“任务照样按时跑完只是成本翻了几倍”等你发现的时候存储账单和下游抱怨已经堆成山了。所以治理膨胀靠的不是救火是常态化的监控指标。1.3 为什么这两件事总是一起出现数据倾斜和数据膨胀不是两个独立的问题它们在链路上是互相喂养的关系。倾斜会导致某几个分区的数据量远超平均而膨胀会把这个“绝对值”进一步放大——本来最大分区就有 30GB再被 join 放大三倍就是 90GB 压在一个 task 上任何内存配置都扛不住。反过来膨胀也会制造出新的倾斜。举个很典型的场景一张表本身分布还算均匀但它关联了一张有明显热点 key 的维表比如某个默认值、某个测试账号、某个占了大头的渠道 IDjoin 完之后这些 key 的行数被成倍放大原本不算热点的 key 变成了新热点。所以你在排查的时候不能只看源表的分布得看经过每一步转换之后的分布。这也是为什么我一直建议在关键节点落中间表并统计行数行数变化是排查膨胀最直接的证据分布变化是排查倾斜最直接的证据。2. 从现象到根因一次典型事故的完整还原2.1 现象观测先看三个信号别急着改代码我现在的习惯是遇到长尾任务先在监控面板上读三个信号这三条基本能定位问题类型。第一个信号是task 耗时的分布直方图看 P99 和 P50 的比值比值大说明是长尾典型的倾斜特征如果所有 task 都慢那是资源不足或者数据量整体上涨跟倾斜没关系别改错方向。第二个信号是shuffle read size 的分区分布也就是每个 reduce task 读了多少数据。Spark UI 的 Stage 页面里能看到每个 task 的 Shuffle Read Size最大值和平均值差 5 倍以上基本就是倾斜。这个指标比耗时更准确因为耗时会被 GC、磁盘抖动等因素干扰而数据量是实打实的。第三个信号是Spill磁盘溢写和 GC 时间。Spill 大量出现说明单个 task 的内存撑不住它的数据量这是倾斜的直接后果GC 时间占比超过 10% 也说明内存压力大。有这两个信号加持你几乎可以确定问题出在某个分区的数据量上而不是在算子上。提示不要只看任务总耗时这一个数字。总耗时是结果不是原因。原因藏在分布里。2.2 根因定位执行计划、数据分布、日志三条线交叉验证定位倾斜我一般走三条线交叉验证。第一条线是执行计划explain extended或者 Spark UI 的 DAG 图重点看 shuffle 发生在哪个阶段是 group by 的 hash aggregate是 join 的 sort-merge还是 window 函数的 partition by。不同的阶段对应不同的处理手法搞错阶段就是白费劲。第二条线是数据分布采样直接在源表上按 key 做一次 count 排序看 top20 的 key 占了多大比例。这一步有个细节采样的时候要带上实际的分区条件别拿全表跑一个是慢另一个是全表分布和当天增量的分布可能差很远——比如某个大促日某个渠道的订单量暴涨全表看是均匀的当天看是极端倾斜的。第三条线是日志和异常栈。OOM 的堆栈信息里通常能看到是哪个集合撑爆的是UnsafeExternalSortershuffle 排序还是HashMap聚合容器这能反推是聚合阶段还是 join 阶段。还有一种特别隐蔽的情况笛卡尔积。join 条件写漏了、或者关联键的类型不一致导致隐式转换失败、或者用了!、or这种非等值条件优化器会退化成 BroadcastNestedLoopJoin数据量一大直接原地爆炸。这种问题在数据量小的时候完全看不出来上线之后才暴露所以执行计划里看到CartesianProduct或者BroadcastNestedLoopJoin一定要警觉。2.3 为什么“加资源”这个第一反应通常是错的我刚入行时的第一反应也是加资源加内存、加 executor 数量、调大并行度。这个操作有它的合理性——它确实能把 OOM 变成不 OOM让任务先跑通。但它解决的是症状不是病因副作用是三个成本上去了、耗时没明显改善、下次数据量再涨一点又挂。更麻烦的是“调大并行度”这个操作经常是无效的。spark.sql.shuffle.partitions从 200 调到 2000听起来每个分区变小了但如果你的聚合 key 只有几十个比如按省份聚合全国就三十几个 key那 hash 分区数根本不是瓶颈瓶颈在 key 的粒度上——三十几个 key 分到 2000 个分区里还是只有三十几个分区有数据剩下的全空着。这时候调并行度纯属自我安慰。所以我现在的顺序是先确认是不是倾斜再确认倾斜发生在哪个算子再选对应的处理手法最后才是调参数兜底。参数是最后一层保险不是第一层手段。3. 数据分布诊断动手之前先把数据看清楚3.1 三行 SQL 看清 key 的分布比任何猜测都靠谱诊断倾斜最有效的手段就是把分布打出来不需要什么工具几行 SQL 就够。第一行看 top 值select shop_id, count(1) as cnt from dwd_order_detail where dt ${bizdate} group by shop_id order by cnt desc limit 20;第二行看总量和 top 值占比这一步是为了量化“到底多严重”select max(cnt) as max_cnt, avg(cnt) as avg_cnt, sum(cnt) as total_cnt, max(cnt) / sum(cnt) as top1_ratio from ( select shop_id, count(1) as cnt from dwd_order_detail where dt ${bizdate} group by shop_id ) t;第三行看分布形态用分位数判断是“少数几个超大头”还是“整体长尾”select percentile_approx(cnt, 0.5) as p50, percentile_approx(cnt, 0.9) as p90, percentile_approx(cnt, 0.99) as p99, max(cnt) as max_cnt from ( select shop_id, count(1) as cnt from dwd_order_detail where dt ${bizdate} group by shop_id ) t;这三行跑完你心里就有数了。如果是 max 远大于 p99说明是极少数热点 key 造成的处理方法就是热点剥离或者加盐打散如果是 p90 就已经很高、分布整体右偏说明粒度选得不对得从业务模型上换聚合维度。注意统计的时候记得把 null 和默认值单独看一下。很多倾斜的真正元凶是null、-1、unknown、空字符串这类“无效 key”它们匹配不上任何维表却在 group by 里稳稳占据一个巨大分区。3.2 什么程度算倾斜什么程度可以不管这个判断标准很关键因为不是所有不均都需要处理。处理倾斜有成本加盐会让代码变复杂广播会占内存热点剥离要多一次 union如果倾斜程度轻微处理收益还不如维护成本。我给自己定的经验阈值是这样的倾斜程度判断标准处理建议可忽略max / avg 3且最大分区小于 200MB不动观察即可轻度max / avg 在 3 到 5 之间优先调参数和 AQE 自适应暂不改逻辑中度max / avg 在 5 到 20 之间或 top1 占比超过 10%针对性处理加盐或广播重度max / avg 20或 top1 占比超过 30%必须改逻辑热点剥离 加盐组合致命单个分区超过 10GB 或稳定 OOM停机排查通常是笛卡尔积或维表重复这套阈值不是拍脑袋背后是资源账一个 task 处理 200MB 数据用默认 1GB 左右的执行内存完全够多花十分钟你也感觉不到一旦单分区超过 2GBSpill 几乎必然发生性能下降就不是线性的了。3.3 用执行计划定位倾斜发生的阶段分布看完了接下来要知道它是在哪一步炸的。原表分布均匀跑完 join 之后炸了那问题在 join原表就炸那问题在读取或者上游写入。看执行计划的时候重点盯三个位置Exchange hashpartitioningshuffle 节点看它的分区键是什么这个键就是倾斜的载体。SortMergeJoin等值 join 走这个算子倾斜通常来自关联键的分布。HashAggregate聚合算子倾斜来自 group by 的键。Window窗口函数里partition by的字段分布不均同样会倾斜。有一种情况值得单独说多个 shuffle 阶段串联的时候倾斜会被放大。比如第一步 group by 已经倾斜了第二步再 join 一个表那个倾斜的分区会带着更大的数据量进入下一个 shuffle雪球越滚越大。所以排查要从最早出现长尾的那个 stage 开始别盯着最后一个。4. 数据倾斜的处理手法与选型逻辑4.1 空值倾斜最便宜的一刀先砍了再说在所有倾斜类型里null 或者无效默认值造成的倾斜是最好处理的因为这类数据往往在业务上没有意义。处理方式有两种取决于业务要不要保留这些行。如果业务上不需要直接过滤掉一行where shop_id is not null and shop_id ! -1就能解决大半问题。如果业务上需要保留比如要统计“未归属渠道”的订单量那就给这些 null 打上随机后缀让它们分散到不同分区select case when shop_id is null or shop_id -1 then concat(unknown_, cast(floor(rand() * 100) as string)) else shop_id end as shop_id_key, order_id, amount from dwd_order_detail where dt ${bizdate};这样处理后聚合结果会分散成 100 份最后再按业务口径把它们合并回“未知”这一类。这个手法的成本极低几乎不影响性能是排查倾斜时应该第一个尝试的动作。提示我见过不少团队在关联维表之前不过滤 null理由是“null 关联不上会丢失数据”。但你要想清楚关联不上的数据本来在结果里就是 null与其让它集中在一个分区里空转不如提前处理掉。4.2 两阶段聚合加盐打散的标准做法两阶段聚合是处理 group by 倾斜最通用的手法思路很朴素既然一个 key 全砸在一个分区里那就人为地把它拆成 N 份先在局部聚合再去掉盐做全局合并。整个过程分三步走。第一步打盐。给每一行加一个 0 到 N-1 的随机数作为盐值drop table if exists tmp_order_salted; create table tmp_order_salted as select shop_id, cast(floor(rand() * 100) as int) as salt, order_id, amount from dwd_order_detail where dt ${bizdate};第二步按(shop_id, salt)做局部聚合这一步的 key 数量变成原来的 100 倍但每个 key 都很小天然均匀drop table if exists tmp_order_part_agg; create table tmp_order_part_agg as select shop_id, salt, count(1) as order_cnt, sum(amount) as total_amt from tmp_order_salted group by shop_id, salt;第三步去掉盐做全局合并这一步的数据量已经小了 100 倍即使还有倾斜也扛得住insert overwrite table dws_shop_order_1d partition(dt ${bizdate}) select shop_id, sum(order_cnt) as order_cnt, sum(total_amt) as total_amt from tmp_order_part_agg group by shop_id;这里有个非常容易踩的坑如果聚合逻辑里包含count(distinct xxx)直接套用上面的三步会算错。原因是同一行数据只带一个盐但同一个order_id可能出现在不同店铺、不同行的记录里被分到不同的盐分组各自计数之后再求和就重复了。正确处理是先按(shop_id, order_id)去重再去打盐计数drop table if exists tmp_order_dedup; create table tmp_order_dedup as select distinct shop_id, order_id from dwd_order_detail where dt ${bizdate}; -- 之后的打盐、局部计数、全局求和三步照常盐值的数量怎么定我的经验是按“热点 key 的数据量 / 单个 task 可承受的数据量”来估。比如最大 key 有 8GB单个 task 舒服的量级是 200MB那就要拆成 40 份以上取 100 比较稳妥。盐值太少起不到打散效果太多会让第二阶段的 key 数量暴涨反而增加调度开销。100 到 500 这个区间是实践中比较常见的。4.3 Map Join 与广播小表关联的最优解如果倾斜发生在 join 阶段而其中一张表足够小比如几十 MB 的维表最干脆的做法是把它广播到每个 executor让 join 变成 map 端的本地操作彻底消除 shuffle。Spark 里可以显式加 hintselect /* BROADCAST(dim_shop) */ a.order_id, a.amount, b.shop_name, b.city_name from dwd_order_detail a left join dim_shop b on a.shop_id b.shop_id where a.dt ${bizdate};Hive 里的写法是/* mapjoin(b) */。这个方案的好处是彻底、干净没有额外的代码复杂度。但它有三个必须知道的限制。第一广播有阈值超了会静默退化。Spark 的spark.sql.autoBroadcastJoinThreshold默认是 10MB即使你加了 hint如果表大小超过阈值优化器还是可能退化成 SortMergeJoin而且不会有任何报错提示你只会发现任务还是那么慢。所以加完 hint 之后一定要看执行计划里是不是真的出现了BroadcastHashJoin。第二广播表会被复制到每个 executor如果表本身有几百 MB加上 driver 分发和 executor 缓存的开销很容易把内存顶满甚至引发 driver OOM。所以广播只适合真正的小表几十 MB 以内比较安全。第三广播解决的是小表问题解决不了大表之间的倾斜。两个大表关联关联键本身分布不均广播没用还是得回到加盐或者热点剥离。4.4 热点 key 单独剥离把最重的那个拎出来有些场景下热点 key 的数量极少可能就两三个商家、几个测试账号但它们占了全表三成以上的数据。这种情况下与其给全表打盐会波及所有本来正常的 key不如把这几个热点单独拎出来处理剩下的走正常流程最后 union 回去。-- 冷数据正常关联 drop table if exists tmp_result_cold; create table tmp_result_cold as select a.order_id, a.shop_id, b.shop_name, a.amount from dwd_order_detail a left join dim_shop b on a.shop_id b.shop_id where a.dt ${bizdate} and a.shop_id not in (shop_hot_001, shop_hot_002); -- 热数据单独把维表拉成 map在代码侧完成关联 -- 或者给热点数据打盐后关联维表同步复制多份 drop table if exists tmp_result_hot; create table tmp_result_hot as select a.order_id, a.shop_id, b.shop_name, a.amount from ( select order_id, shop_id, amount, cast(floor(rand() * 50) as int) as salt from dwd_order_detail where dt ${bizdate} and shop_id in (shop_hot_001, shop_hot_002) ) a left join ( select shop_id, shop_name, salt from dim_shop lateral view explode(array(0,1,2,3,4,5,6,7,8,9)) t as salt where shop_id in (shop_hot_001, shop_hot_002) -- 实际使用时这里要展开到 0 到 49 ) b on a.shop_id b.shop_id and a.salt b.salt; -- 合并 insert overwrite table dws_order_detail partition(dt ${bizdate}) select * from tmp_result_cold union all select * from tmp_result_hot;这个方案代码量偏大但收益也很直接既解决了热点又不影响正常数据的执行路径。用的时候有个关键前提——热点 key 得是稳定且少量的。如果热点是动态变化的今天这三个明天那五个那维护成本会很高还是老老实实全表打盐更省心。所以上线之前我会先观察一周的 top key 列表确认热点是否固定。另外要提醒一句剥离之后要复查。因为剥离本身会改变数据分布可能把原来的第二梯队 key 顶上来变成新热点union 完之后再跑一次分布统计别嫌麻烦。4.5 参数与引擎层面的兜底AQE 是必开的但不能迷信现代的计算引擎在自适应优化上已经做了不少事情Spark 3.x 的 AQEAdaptive Query Execution就是典型代表。它能在运行时根据实际的 shuffle 统计信息动态合并小分区、拆分大分区、甚至自动处理倾斜 join。这几个参数我基本是必开的spark.sql.adaptive.enabledtrue spark.sql.adaptive.coalescePartitions.enabledtrue spark.sql.adaptive.coalescePartitions.initialPartitionNum2000 spark.sql.adaptive.advisoryPartitionSizeInBytes256MB spark.sql.adaptive.skewJoin.enabledtrue spark.sql.adaptive.skewJoin.skewedPartitionFactor5 spark.sql.adaptive.skewJoin.skewedPartitionThresholdInBytes256MB这几个参数的逻辑是先把初始分区数设大一点让 AQE 有足够的余地做合并advisoryPartitionSizeInBytes控制每个分区期望的大小skewedPartitionFactor是判定倾斜的倍数阈值超过这个倍数的分区会被拆分。但 AQE 不是万能的它有两个明确的边界。第一它只能处理 join 阶段的倾斜group by 阶段的倾斜它管不了第二拆分本身有开销如果倾斜是那种“一个 key 占了 50% 数据”的极端情况拆完还是大只是从一个大块变成几个中等块效果有限。所以我的定位是AQE 作为默认配置常开能兜住大部分轻度倾斜重度的还是得改逻辑。另外并行度这个参数要理解清楚它的作用边界。spark.sql.shuffle.partitions调到几千只在 key 的基数足够大的时候有效。如果你的 group by 维度只有几十个值调多大都没用因为数据只能分到那几十个桶里去。这一点我在前面提过但值得再强调一遍因为这是最常见的无效操作。4.6 从业务模型上根治换粒度、预聚合、改口径前面说的都是计算侧的技巧但真正治本的方案往往在业务侧。我处理过一个特别顽固的倾斜按省份 城市聚合某个城市的订单量是其他城市的几百倍。加盐试过了能缓解但治不了因为那个城市本身就占了全量的四成打散 100 份之后每份还是很大。最后的解法是换粒度——把聚合维度从“城市”下沉到“城市 商圈 小时”粒度细了数据天然分散再往上汇总就很快了。这个思路可以推广成一句话当你在粗粒度上聚合遇到倾斜时先看看能不能在更细的粒度上先聚合一轮。还有一个思路是预聚合。日粒度的任务可以改成小时级增量跑每个小时的数据量小倾斜的绝对影响就小或者在上游就把常用的聚合结果算好存成汇总表下游直接查汇总表绕开大表关联。最后是改口径。有些指标本身的计算方式就注定会倾斜比如“全站去重用户数”这种全局去重的需求如果硬要用一个 job 算天然就是单点。可以考虑用 HyperLogLog 这类近似算法或者用位图Bitmap做交并差牺牲一点精度换取性能。这个决策需要业务方参与不能技术侧单方面拍板。4.7 手法选型对照表别拿锤子找钉子不同倾斜场景适配不同的手法我把常用的组合整理成一张表遇到问题可以直接对号入座场景特征首选手法次选手法不推荐null / 默认值占比高过滤或打随机后缀单独拎出来算加盐没必要小表 join 大表广播 Map Join小表扩容打盐调并行度单 key 占比 10%~30%两阶段加盐聚合热点剥离只调参数单 key 占比超过 30%热点剥离 细粒度预聚合换聚合维度单纯加盐多阶段串联放大从最早的 stage 开始治理中间结果落表只看最后一个 stage全局去重类需求近似算法 / 位图分桶预聚合硬算 count distinct笛卡尔积修 join 条件补等值关联键加内存5. 数据膨胀的防控给链路做“体脂管理”5.1 膨胀的四个主要来源先认清楚再动手数据膨胀看着玄乎其实来源就那么几类认清楚之后治理路径很清晰。第一类是 join 导致的行数放大这也是最常见的一类。根因通常是关联键在维表里不唯一——维表里有重复记录、拉链表没取最新快照、维度表做了 SCD 多版本但关联时没加时间条件。一行事实数据关联上三行维表输出就是三行关联上十行就是十倍。这类问题有个很典型的特征join 前后的行数对不上而且差异是整数倍关系。第二类是 explode 和行列转换。一行里存了一个数组或者一个 JSON展开成多行如果数组长度很大比如一个用户的标签列表有上千个放大倍数非常夸张。这类膨胀有时候是业务必需的但通常可以在展开之前先过滤、先去重、先截断。第三类是多表关联的顺序问题。三张大表 join先关联两个大的得到一个巨型中间结果再关联第三个中间结果的落盘和 shuffle 成本都被浪费了。正确的顺序应该是先做过滤和预聚合把每个表都缩小之后再关联。第四类是写入策略问题。全量覆盖写、快照表不清理、分区粒度过细导致小文件堆积这些属于存储层面的膨胀。我见过一个表按小时分区一天 24 个分区一年 8760 个分区每个分区里十几个小文件光元数据操作就要花好几分钟。5.2 维表去重一个 row_number 解决八成放大问题维表重复导致的行数放大处理起来其实很简单关联之前先用窗口函数取最新一条drop table if exists tmp_dim_shop_latest; create table tmp_dim_shop_latest as select shop_id, shop_name, city_code, shop_level, update_time from ( select shop_id, shop_name, city_code, shop_level, update_time, row_number() over ( partition by shop_id order by update_time desc ) as rn from dim_shop_full ) t where rn 1;这个写法几乎适用于所有“维表有多版本”的场景。有两个细节要注意一是排序字段必须是能唯一确定最新版本的如果update_time有重复要再加一个自增 ID 做次级排序否则结果不稳定二是如果维表本身数据量很大取最新这一步本身也是个大 job最好在维表的生产环节就做好别每次关联前都算一遍。对于拉链表除了去重还要加上时间条件只关联在有效期内start_date biz_date end_date的那条记录这一点漏了同样会放大。5.3 分区裁剪、列裁剪与生命周期三件最基础也最容易被忽略的事这三件事听起来像常识但我见过的膨胀问题里至少有一半跟它们有关。分区裁剪指的是where条件里必须带分区字段而且不能对分区字段做函数运算。where dt 20240101能裁剪where substr(dt, 1, 4) 2024就不能会导致全表扫描。这个坑每年都有人踩。列裁剪指的是只 select 需要的列别图省事写select *。宽表动辄几百个字段全选一遍IO 和内存都要多花几倍。尤其是中间表每一层都select *几层叠下来就是灾难。生命周期指的是分区数据的保留策略。ODS 层留多久、DWD 层留多久、中间临时表留多久得有明确规范并且用自动化的清理任务去执行。我一般建议临时表只保留 7 天中间层 90 天到 180 天明细层按业务要求冷数据归档到低频存储。没有生命周期管理的数仓膨胀是必然的只是时间问题。5.4 小文件治理与存储格式选择小文件是存储膨胀里最隐蔽的一类。它的问题不在于总容量大而在于文件数量多导致元数据操作和任务调度开销暴涨。一个分区里有一千个几 KB 的文件读的时候要开一千次写的时候要准备一千个 buffer性能比读一个 100MB 的文件差一个数量级。治理手段有两层。第一层是写入时控制Spark 里开spark.sql.adaptive.coalescePartitions.enabled和spark.sql.adaptive.advisoryPartitionSizeInBytesHive 里调hive.merge.mapfiles、hive.merge.smallfiles.avgsize、hive.merge.size.per.task让引擎在写入前自动合并。第二层是定期做合并任务把历史分区的小文件重写一遍。存储格式上优先选列式存储加压缩Parquet 或 ORC 都行压缩算法用 snappy 兼顾速度用 zstd 压缩率更高但 CPU 开销大一些。行式存储比如 TextFile、JSON在分析场景下几乎是自找麻烦压缩比差、列裁剪失效能换就换。治理动作影响维度典型收益落地难度维表 row_number 去重计算膨胀行数下降 30%~80%低加分区条件扫描量减少 90% 以上的无效读低去掉 select starIO 和内存减少 30%~60% 的读数据量低中间表设 TTL存储成本存储费用下降明显中小文件合并调度开销任务启动时间缩短数倍中换列式存储全链路压缩比提升 3~5 倍高6. 能直接抄作业的配置与实操清单6.1 关键参数清单与调优逻辑参数永远服务于场景脱离场景的“最佳配置”是不存在的。我把常用的参数和它的作用边界整理如下用的时候对照自己的情况取用参数作用常见取值什么时候别用spark.sql.shuffle.partitionsshuffle 分区数数据量 GB 数 × 2~3key 基数小于分区数时无效spark.sql.adaptive.enabled开启动态优化true基本没有理由关spark.sql.adaptive.skewJoin.enabled自动拆分倾斜分区true极端热点 key 效果有限spark.sql.autoBroadcastJoinThreshold广播表大小阈值10MB~100MB调太大容易 driver OOMspark.sql.adaptive.advisoryPartitionSizeInBytes目标分区大小128MB~256MB分区太小会增加调度开销hive.exec.reducers.bytes.per.reducer每个 reducer 处理量256MB~1GB调太小会产生大量小文件hive.merge.smallfiles.avgsize小文件合并阈值128MB对已存在的历史分区无效有个容易忽略的点这些参数的生效顺序是有先后的。AQE 的合并和拆分发生在运行时它依赖初始的分区数如果你把初始分区数设得太小比如 200AQE 连拆的余地都没有。所以正确做法是初始分区数适当放大让 AQE 有操作空间而不是指望一个参数解决所有问题。6.2 一套完整的加盐聚合模板把前面的步骤串起来给一份可以直接改字段名就用的模板。假设场景是订单明细表按shop_id聚合存在严重倾斜。-- step 0: 分布探测只在排查阶段跑别写进日常任务 select shop_id, count(1) as cnt from dwd_order_detail where dt ${bizdate} group by shop_id order by cnt desc limit 20; -- step 1: 打盐盐值数量按热点数据量估算这里取 100 drop table if exists tmp_order_salted; create table tmp_order_salted as select shop_id, cast(floor(rand() * 100) as int) as salt, order_id, amount from dwd_order_detail where dt ${bizdate} and shop_id is not null and shop_id not in (-1, unknown); -- step 2: 局部聚合 drop table if exists tmp_order_part_agg; create table tmp_order_part_agg as select shop_id, salt, count(1) as order_cnt, sum(amount) as total_amt from tmp_order_salted group by shop_id, salt; -- step 3: 全局合并 insert overwrite table dws_shop_order_1d partition(dt ${bizdate}) select shop_id, sum(order_cnt) as order_cnt, sum(total_amt) as total_amt from tmp_order_part_agg group by shop_id; -- step 4: 清理临时表 drop table if exists tmp_order_salted; drop table if exists tmp_order_part_agg;这套模板在大多数批处理引擎上都能跑字段名和表名换一下就行。要注意的是 step 2 和 step 3 之间最好落一次盘虽然多了一次 IO但换来的是可观测性——你能看到中间结果的行数和分布出问题时能快速定位是哪一步炸的。6.3 上线前的验证动作别跳过改完逻辑直接上线这是最容易翻车的地方。我给自己定的上线检查有四项。第一结果正确性比对。用新旧两套逻辑跑同一天的数据把结果全字段比对差异行数必须为 0。注意这里要特别关注 null 值和边界值加盐处理最容易在这些地方出问题。第二性能对比。不只是看总耗时要看 task 耗时的分布。如果总耗时降了但长尾还在说明只是数据量变小了倾斜没根治下次数据量涨回来还会犯。第三中间表体积对比。改完之后中间结果的行数和体积应该明显下降如果没降反升说明某个环节的处理方式不对。第四回归下游。倾斜和膨胀的治理经常涉及口径微调比如 null 的处理方式变了、去重逻辑变了这些变化会传导到下游指标上必须提前跟下游对齐别等报表出数了才发现对不上。7. 常见问题速查与踩坑心得7.1 典型问题速查表现象最可能的原因排查动作处理方式少数 task 跑几小时shuffle 分区数据不均看 shuffle read size 分布加盐或热点剥离广播 hint 加了没效果表超过阈值静默退化看执行计划是否为 BroadcastHashJoin调阈值或改用其他手法加盐之后结果变多了count distinct 被打散重复计数比对去重前后的行数先 distinct 再打盐任务不慢但存储涨得快全量覆盖写或分区过多看分区数和文件数设 TTL、合并小文件join 后行数暴涨十倍维表重复或拉链未取最新检查维表主键唯一性row_number 取最新调大并行度毫无改善key 基数太小看 group by 字段基数换方案别调参数OOM 稳定复现在同一 stage单分区数据量超内存看 spill 和 GC 指标两层处理先降水再抗压早上跑得快下午跑得慢队列资源竞争看资源池排队情况错峰调度或独立队列7.2 几条从坑里爬出来的经验第一条先测分布再动手别凭感觉改。我早期最大的毛病就是看到任务慢就直接加盐结果加了半天发现倾斜根本不在聚合阶段而是在前面的 join 上白忙一场。现在我改任何倾斜逻辑之前一定要先跑分布统计和执行计划确认清楚了再动手这个习惯省下来的时间远超测分布花的那几分钟。第二条临时表不要省。很多人为了“少一次 IO”把中间步骤写成嵌套子查询觉得性能更好。实测下来多写一次临时表反而更容易排查而且因为中间结果有统计信息优化器的执行计划往往更准。尤其是治理期可观测性比那点 IO 开销重要得多。第三条治理要留痕。每次调整都把“原始分布、处理手法、调整后分布、耗时变化”记下来形成一个表格。我维护了这样一份记录一年下来积累了几十条后来遇到新问题基本能直接在里面找到相似案例比翻文档快得多。第四条别一次性改太多变量。优化的时候容易贪心一次改五处结果效果好不知道是哪一处起的作用效果差也不知道是哪一处搞坏的。我现在的做法是一次只动一个点跑完对比一次确认有收益再动下一个。第五条对“看起来正常”的分布保持警惕。有一些倾斜是业务特性决定的比如某个大商家、某个默认渠道它们的量大是合理的不该无脑去掉。处理之前先跟业务确认这些大头是不是应该存在别把正常业务数据当异常清理了。最后分享一个我一直在用的小习惯每个核心任务旁边挂一个“体检 SQL”每天跑一次输出三个数——最大分区占比、中间表膨胀率、单分区最大体积。这三个数超过阈值就告警。这套机制帮我提前发现了不少问题比等任务挂了再查要从容得多。