数据倾斜和数据膨胀这两个词我在过去几年里几乎每隔一两个月就要跟它们打一次照面。业务方跑过来一句话——“这个任务昨天跑了四十分钟今天跑了四个小时”剩下的活全在我这儿。刚上手那会儿我也迷信参数把并行度从 200 调到 2000、把内存从 4G 加到 16G结果除了让集群更难受跑出来的效果基本没变。后来踩的坑多了才明白数据倾斜和数据膨胀是两件不同性质的事前者是分布问题某几个 Key 扛走了大半的数据量后者是体量问题一个 Join 或者一个窗口函数把中间结果放大了十几倍。它们经常同时出现在一个任务里互相掩护让你看日志看不出来、看指标看不明白。这篇东西写给正在被这俩问题反复折腾的数据开发和数仓同学从怎么判断、怎么定位到几套能直接抄的改法和真实的调参账本我都会摊开讲。1. 先把两个概念掰开倾斜看分布膨胀看体量1.1 数据倾斜的本质是Key分布不均引发的水桶效应分布式计算的大部分时间都花在 Shuffle 上而 Shuffle 的基本原则是“同一个 Key 必须进同一个分区”。这条规则本身没问题问题是现实数据里的 Key 从来不是均匀分布的。一个用户表头部几个账号可能占了 30% 的行为数据一个订单表某几个商户的订单量是普通商户的几百倍一张埋点表空字符串、unknown、-1这类占位值能占到千万级。这些 Key 落到同一个 Task 上任务耗时就变成了正态分布里那根突兀的长尾——99% 的 Task 三分钟跑完剩下几个跑了两个小时。这个场景特别像超市结账。你开了 200 个收银台理论上队伍应该是均匀的但只要有一个收银台前面排了 500 个人的大团购整个超市的“平均结账时间”就失去意义了所有人都在等那一队。加收银台调并行度解决不了问题因为你不能把一个大团购拆成 500 个单人——同一个订单必须一次性结算完。真正有效的办法是要么把这个大团购拆成多个可独立结算的小批次加盐打散要么单独给它开一个专用通道热点 Key 分离要么干脆在进场前把它的信息提前广播给每个收银台Map 端 Join。判断倾斜有一个很朴素的指标同一 Stage 内 Task 的 Shuffle Read Size 最大值与中位数之比。我自己的经验阈值是 3 倍开始留意10 倍基本确诊50 倍以上就得立刻动手了。另外还有一个信号也很好用——任务整体耗时远大于各 Task 耗时的中位数乘以并行度说明大量资源在等少数几个 Task这是典型的尾部拖累。1.2 数据膨胀是中间结果被放大的过程不是源数据变多数据膨胀容易被误解成“数据量变大了”其实它指的是中间结果相比输入数据被成倍放大。最典型的是 Join 放大左表 1000 万行去关联一张右表如果右表的关联键不唯一一个左表行匹配上 5 个右表行结果就是 5000 万行。更狠的是多对多关联两边都有重复键实际输出行数是两个分布相乘的结果几十倍的放大很常见。我见过一个最夸张的案例两张百万级的表关联之后中间结果超过了 20 亿行磁盘直接写满任务反复重试整个队列被它堵了半天。膨胀的另一个来源是统计口径。count(distinct user_id)在分布式执行时会把数据按 user_id 打散再聚合如果同时还有 group by中间结果会先按“分组键 去重键”展开等于把数据按去重键的基数放大了一遍。窗口函数也一样row_number() over(partition by user_id order by ts)在按 user_id 重分区之后如果某个用户的记录有上千万条那这个分区的排序内存直接爆掉。膨胀的隐蔽性在于它不一定报错。很多时候任务能跑完只是慢了、磁盘写多了、下游表大了一圈你不在执行计划里翻一遍根本发现不了。所以我一直强调优化之前必须先看执行计划里的行数估算和实际输出行数尤其是每个 Join 节点的输出行数对比输入行数这个比值就是膨胀系数。1.3 两个问题为什么会手拉手出现有意思的是倾斜和膨胀经常是同一个 SQL 的两个症状。举个例子一张行为表按 user_id 关联一张用户维表维表里因为历史原因同一个 user_id 有 3 条记录不同来源关联之后数据膨胀了 3 倍。膨胀之后本来就不均匀的 user_id 分布被进一步放大——那个占 30% 数据量的头部账号膨胀之后可能吃掉了一半以上的数据。这时候你只处理倾斜不处理膨胀等于给一个漏水的桶换了个更大的桶。反过来也一样。为了缓解膨胀有人会把 Join 拆成多步、加上各种中间表结果每一步的 Shuffle 都可能引入新的倾斜点。我在实际项目里形成的顺序习惯是先治膨胀再治倾斜。因为膨胀会改变数据分布先把不该有的行去掉倾斜的严重程度往往就下降一个量级后面用的手段也能轻一些。2. 定位先拿证据再动手别上来就调参2.1 三个信号快速判断自己是不是真的遇到了倾斜不是所有慢任务都是倾斜。我给团队定的判断顺序是这样的先看输入数据量有没有异常增长再看 Stage 的 Task 耗时分布最后看单个 Task 的 Spill 情况。如果输入数据量正常、Task 耗时分布正常、没有大量 Spill那大概率是资源竞争或者上游依赖问题跟倾斜没关系。真正需要怀疑倾斜的信号有三个。第一某个 Stage 卡在 99% 很久不动界面上的 Task 数显示还剩几个第二这个 Stage 的 Shuffle Read Size 分布极度不均最大值是中位数的几十倍第三少数 Task 的 Spill磁盘溢写数据量远超其他 Task说明它们在内存里装不下被迫落盘。这三个信号同时出现基本可以锁定。还有一个容易被忽略的角度看输入源的分布。如果上游表的分区大小差异很大比如按天分区某一天的数据量是平时的 20 倍大促、活动、补数那倾斜是必然的不用查执行计划都能猜到。我现在的习惯是每天跑任务前先扫一眼上游各分区的大小超过均值 5 倍的分区单独标记出来跑的时候心里有数。2.2 从执行计划里找到膨胀具体发生在哪一层Spark 3.x 之后EXPLAIN EXTENDED的输出已经很好读了重点看三样东西每个 Join 的类型BroadcastHashJoin 还是 SortMergeJoin、每个节点的估行数、以及是否有 Filter 被下推。膨胀问题基本都能在 Join 节点上看出来——如果某个 Join 的输出行数是左表行数的 5 倍以上而业务上这个关联本该是一对一那就说明右表有关联键重复。对于 Hive 侧的任务看执行计划要更细一点。我会把每个阶段的输出行数抄下来做成一个简单的对照表逐层看哪一层的放大最猛。这个动作听起来笨但真的有效我靠这个办法在一个四层嵌套的宽表任务里定位到问题出在第二层的维表关联上——那张维表因为一次补数操作产生了重复主键导致后面所有环节都在为这 3 倍膨胀买单。另外提醒一句估行数不可全信。统计信息过期是常态尤其是经过多层子查询和 UDF 之后优化器给出的行数可能差好几个数量级。所以看到可疑的节点最好单独把那一层的输入输出抽出来跑一次采样用真实行数验证别拿估算值当判决书。2.3 我常用的分层观察清单观察层级具体指标正常区间异常含义数据源层分区大小 / 表行数与历史均值偏差 2 倍上游补数或重复写入Shuffle 层Task Shuffle Read 最大值 / 中位数 3 倍超过 10 倍即倾斜Join 层输出行数 / 左表行数接近 1一对一大于 3 说明关联键重复聚合层中间结果行数 / 输入行数小于 1聚合收敛大于 1 说明维度爆炸落盘层Spill 数据量 / 单 Task 处理量 30%超过 50% 说明内存严重不足这张表我基本是贴在显示器旁边的每次排查先按行扫一遍能过滤掉一半以上的误判。3. 数据倾斜的六套解法与选型逻辑3.1 方案一两阶段聚合把一次聚合拆成局部加全局这是最通用、成本最低的方案适用于group by场景的倾斜。核心思路是把一次聚合拆成两次第一阶段给 Key 加上随机前缀俗称加盐让原本挤在一个分区的数据被打散到 N 个分区做局部聚合第二阶段去掉前缀再做一次全局聚合。因为第一阶段已经把数据量压下来了第二阶段的输入规模通常只有原来的百分之几即使还有倾斜也扛得住。用 SQL 表达大概是这个意思-- 第一阶段加盐局部聚合 SELECT concat(user_id, _, cast(floor(rand() * 16) as string)) AS salted_key, user_id, count(1) AS cnt, sum(amount) AS amt FROM behavior_log WHERE dt 2024-05-20 GROUP BY concat(user_id, _, cast(floor(rand() * 16) as string)), user_id; -- 第二阶段去盐全局聚合 SELECT user_id, sum(cnt) AS cnt, sum(amt) AS amt FROM stage1_result GROUP BY user_id;加盐的倍数 N 怎么定我的经验公式是N ceil(热点 Key 数据量 / 单分区可承受数据量)。假设单个分区舒服处理 200MB热点 Key 有 3.2GB那 N 至少取 16。取 16 是我比较常用的默认值因为它是 2 的幂分区数量对齐比较自然实测下来对绝大多数场景够用。取太大反而不好因为盐值过多会让第一阶段产生大量小分区调度开销反而上升。注意加盐方案只对聚合类操作有效对 Join 无效。而且它要求聚合函数是可分解的sum、count、min、max 可以collect_list、percentile 这类不行。3.2 方案二Map 端 Join 与广播把小表塞进内存如果倾斜的原因是“大表关联小表”那最省事的办法就是让小表不进 Shuffle直接广播到每个 Executor 的内存里大表在 Map 阶段就地完成关联Shuffle 直接消失倾斜也就无从谈起。这是我认为性价比最高的一招改动量小效果立竿见影。Spark 里的控制参数是spark.sql.autoBroadcastJoinThreshold默认 10MB。很多团队的实践是调到 50100MB但这里有个容易被忽视的账广播变量是每个 Executor 一份副本。假设表 80MB任务开了 50 个 Executor每个 Executor 4 个并发 Task那广播数据在集群里的实际内存占用是 50 × 80MB 4GB而且这部分内存是在堆内跟 Task 抢的。所以调大阈值的同时必须确认 Executor 的可用内存够否则就会出现“Join 是快了但开始频繁 GC 甚至 OOM”的新问题。Hive 侧对应的是 MapJoin通过hive.auto.convert.jointrue配合hive.mapjoin.smalltable.filesize控制。我一般把文件大小阈值卡在 25MB 左右超过这个体量的小表宁可走普通 Join 加其他优化手段也不硬塞。还有一种情况是“大表关联中表”中表比如 500MB广播肯定不行但又不想全量 Shuffle。这时候可以做分桶 JoinBucket Join两张表都按关联键分桶且桶数成倍数关系就能在 Map 端按桶对齐完成关联。这个方案对表结构有强要求落地成本高只有在稳定复用的核心链路上才值得做。3.3 方案三热点 Key 拆出来单独处理再合并回去对于那种“就那两三个 Key 有问题其他都正常”的场景最直接的做法是把热点 Key 过滤出来单独算剩下的正常算最后 union 回去。这个方案逻辑最简单也最容易验证效果但前提是你得事先知道哪些是热点 Key。找热点 Key 的方式是抽样统计SELECT user_id, count(1) AS c FROM behavior_log WHERE dt 2024-05-20 GROUP BY user_id ORDER BY c DESC LIMIT 20;跑一次拿到 Top 20如果第一名是中位数的几百倍那就锁定它。然后把 SQL 拆成两段WHERE user_id xxx的部分单独跑这部分数据量大但 Key 单一可以配合加盐或者直接给它更大的资源WHERE user_id xxx的部分正常跑最后 union all。这个方案的代价是维护成本。热点 Key 是会变的今天是大促商户明天可能是另一个。所以我一般会把它做成配置化把热点 Key 列表放在配置表里每天跑之前先更新一次SQL 从配置表里读。这套东西搭起来有点麻烦但对于每天都要跑的核心任务值得。3.4 方案四给倾斜 Key 加随机前缀再做扩容关联这个方案是专门治 Join 倾斜的思路是把大表里的热点 Key 打上随机前缀同时把对应的小表数据按同样规则复制 N 份让原本一个 Key 的关联被拆成 N 份并行做做完再去掉前缀。它本质上是“用空间换并行度”。salt 16 # 大表热点key加随机前缀 big_df big_df.withColumn( salted_key, when(col(user_id) hot_key, concat(col(user_id), lit(_), (rand() * salt).cast(int))) .otherwise(col(user_id)) ) # 小表热点key复制N份 small_hot small_df.filter(col(user_id) hot_key) small_hot small_hot.withColumn(salted_key, explode(array([lit(f{hot_key}_{i}) for i in range(salt)]))) small_other small_df.filter(col(user_id) ! hot_key).withColumn(salted_key, col(user_id)) result big_df.join(small_hot.unionAll(small_other), onsalted_key, howleft)注意这里有个坑扩容之后小表的数据量也会跟着涨 N 倍如果小表本身不小扩容后的体积可能超过广播阈值导致执行计划从 BroadcastJoin 退化成 SortMergeJoin反而更慢。所以我一般会在扩容前后各看一眼小表的实际行数超过 100MB 就不走这个方案了。3.5 方案五参数与分区器层面的调整属于兜底手段前面四招都是改逻辑这一招是改配置。顺序上我一直把参数放在最后因为参数只能缓解不能解决。但有些场景确实参数就够了比如倾斜不严重、只是分区数设置得不合理。常用的几个参数# 自适应执行让Spark自己合并小分区、拆分大分区 spark.sql.adaptive.enabledtrue spark.sql.adaptive.coalescePartitions.enabledtrue spark.sql.adaptive.advisoryPartitionSizeInBytes128MB # 倾斜Join自动处理 spark.sql.adaptive.skewJoin.enabledtrue spark.sql.adaptive.skewJoin.skewedPartitionFactor5 spark.sql.adaptive.skewJoin.skewedPartitionThresholdInBytes256MB # Shuffle分区数 spark.sql.shuffle.partitions2048skewedPartitionFactor5的意思是“分区大小超过中位数的 5 倍就算倾斜”skewedPartitionThresholdInBytes256MB是“同时还得超过 256MB 才处理”。两个条件是且的关系这么设计是为了避免对小数据量的轻微不均做无谓的拆分。这套参数在 Spark 3.x 上确实能自动救回一部分倾斜但它的处理方式是“把大分区再拆成几个小分区”对于真正的超级热点 Key 治标不治本只是把两小时变成一小时。Hive 侧对应的开关是hive.groupby.skewindatatrue它做的事情就是自动帮你做两阶段聚合原理跟方案一一样。这个参数我一直建议在 ETL 任务里默认打开没什么副作用。3.6 六套方案的选型对照方案适用场景改动成本效果上限主要风险两阶段聚合group by 倾斜低高聚合函数必须可分解Map 端 Join大表关联小表低很高小表体积失控导致 OOM热点 Key 分离少数 Key 极端倾斜中很高热点会漂移需配置化加盐扩容 JoinJoin 侧倾斜中高扩容后小表变广播失败分桶 Join稳定复用的核心链路高很高对表结构有强约束参数与 AQE轻度倾斜、兜底极低中治标不治本我自己的选择顺序是先看能不能广播方案二不行再看聚合还是 Join方案一或方案四热点极其集中的直接走方案三最后才是参数兜底。4. 数据膨胀的治理从源头到落地的四道闸门4.1 第一道闸门过滤与列裁剪必须前置到最早的子查询膨胀治理最重要的一条原则是尽早砍掉不需要的数据。这一步听起来像废话但我见过的膨胀问题里至少三成是因为过滤条件写在了最外层导致内层全量参与 Join膨胀之后再过滤白算了一大半。举个常见的错误写法SELECT a.user_id, b.city, count(1) FROM (SELECT * FROM behavior_log) a LEFT JOIN user_dim b ON a.user_id b.user_id WHERE a.dt 2024-05-20 AND b.status 1 GROUP BY a.user_id, b.city;这里 dt 过滤写在最外层behavior_log 可能被扫了三年数据b.status 1写在 WHERE 里还会让 LEFT JOIN 退化成 INNER JOIN因为不匹配的行 b.status 是 NULL会被过滤掉语义都变了。正确写法是把 dt 推进子查询把 status 条件放到 ON 里或者用子查询先过滤维表。列裁剪同样重要。SELECT *在宽表场景下会把几十个不用的字段一路带到 Shuffle 阶段序列化和网络传输的开销全部翻倍。我现在写 SQL 的第一条规矩就是不许出现SELECT *连临时表也不行。还有一个进阶技巧是分区裁剪 分桶裁剪。如果下游只用到某几个分区明确写死分区值而不是依赖动态推断能让优化器做出更准确的计划。这条在数据量大、分区多比如按小时分区的表上效果非常明显。4.2 第二道闸门关联维表前先去重警惕一对多放大维表关联键重复是膨胀的头号杀手而且在业务上往往有正当理由——同一用户在不同渠道有多条记录、同一商品有多个供应商。这种表直接拿去 Join放大倍数等于重复度的平均值你还查不出来因为 SQL 没有任何语法错误。我的做法是在关联前强制加一层去重子查询用row_number()或者group by把维表收敛成关联键唯一WITH dim_dedup AS ( SELECT user_id, city, status FROM ( SELECT user_id, city, status, row_number() OVER (PARTITION BY user_id ORDER BY update_time DESC) AS rn FROM user_dim WHERE status 1 ) t WHERE rn 1 ) SELECT a.user_id, d.city, count(1) FROM behavior_log a LEFT JOIN dim_dedup d ON a.user_id d.user_id WHERE a.dt 2024-05-20 GROUP BY a.user_id, d.city;这段去重本身也要注意倾斜——PARTITION BY user_id如果有超级大用户那个分区照样会爆。所以我会在建维表的时候就做一层控制让维表的关联键天然唯一把这层去重做成落表的一部分而不是每次查询时现算。这叫把治理动作固化到数据模型里比在查询里打补丁靠谱得多。顺带说一句多对多的 Join 要格外小心。如果两张表都不唯一输出行数是两个分布的组合稍微有点重复度就是几十倍放大。我以前做过一个用户标签宽表两张表都各自有 1.5 倍左右的重复关联之后直接变成 2.25 倍加上原有的 8000 万行中间结果冲到 1.8 亿行跑一晚上没跑完。后来改成先按关联键预聚合才把规模压回来。4.3 第三道闸门中间结果落盘要带分区、带收敛有些膨胀是“算完了但没用上”典型场景是中间表全量落盘下游只用了很小一部分。这种问题在链路上看不出来但存储成本和时间成本都在那儿。我的做法是给所有中间表强制加两个东西分区字段和行数收敛检查。分区字段保证下游能裁剪行数收敛检查是在写入前跑一次 count如果发现输出行数明显超预期比如超过输入行数的 1.2 倍直接让任务失败告警而不是让它静默地产出一张超大的废表。-- 写入前先验一把超阈值就报错 INSERT OVERWRITE TABLE dws_user_summary PARTITION(dt2024-05-20) SELECT ... FROM ( SELECT ..., count(1) OVER() AS total_rows FROM tmp_result ) t WHERE total_rows 500000000;这个写法有点土但它救过我很多次。有一次上游数据源重复推送导致当天数据量翻了 4 倍就是靠这个阈值把任务拦下来的否则下游十几个任务全得跟着重跑。另外关于文件数量也要管。中间结果如果小文件太多每个文件几 KB读取时的元数据开销和任务调度开销会非常夸张。我一般会在写入时加一个DISTRIBUTE BY或者分区合并让单文件控制在 128MB 左右。这个动作看着琐碎但对整条链路的稳定性影响很大。4.4 第四道闸门窗口函数和去重统计里的隐藏膨胀这两个是膨胀的重灾区因为它们从语法上看不出任何问题。count(distinct)在分布式执行时会转成两层聚合先按“分组键 去重键”展开再 count。如果去重键基数很大中间结果的行数会非常夸张。我遇到过一个count(distinct session_id)的查询session_id 有上亿的基数中间结果达到了输入数据的 60%直接把磁盘写满了。改写方式是先做子查询去重再 count-- 改前 SELECT dt, count(distinct session_id) FROM log GROUP BY dt; -- 改后 SELECT dt, count(1) FROM (SELECT DISTINCT dt, session_id FROM log) t GROUP BY dt;逻辑上等价但后者在 Spark 3.x 上配合 AQE 的表现好很多中间结果可控。如果去重维度有两个以上用group by展开的写法会更稳。窗口函数的膨胀在于分区大小。partition by user_id如果某个用户有几千万条记录那个分区排序时内存必爆。处理办法要么是按时间先做局部窗口再合并要么是给窗口分区加一层盐打散再聚合本质上跟处理倾斜是一套路子。所以你看绕了一圈又回到开头那句话——膨胀和倾斜是同一个问题的两面。5. 一次完整优化实录从8小时到52分钟5.1 现场现象与基线这里说的是一次模拟项目里的宽表任务为了避免具体信息我把它抽象成“用户行为汇总表”。任务逻辑不复杂以行为明细表为主表关联三张维表用户、商品、地区做聚合后写入每日汇总。改造前的基线是这样的输入行为表 2.1TB约 80 亿行用户维表 300MB约 800 万行商品维表 120MB地区维表 3MB。任务跑在 Spark 上Executors 100 个每个 8 核 32G。实际耗时 8 小时 20 分中间出现过两次 Executor OOM重试 3 次。Spark UI 上的表现是有一个 Stage 卡在 199/200 长达两个多小时Shuffle Read Size 最大值 42GB中位数 380MB比值超过 110 倍。同时磁盘 Spill 总量达到了 6TB远超输入数据量。5.2 定位三步锁定元凶第一步我看的是 Join 层的输出行数。执行计划显示行为表关联用户维表之后的估行数是 250 亿行是输入的 3 倍多。这明显不正常因为 user_id 到用户维表应该是一对一。我把用户维表单独拉出来按 user_id 统计发现有 12% 的 user_id 存在 25 条记录属于典型的维表重复。第二步我看的是行为表本身的分布。Top 20 用户的数据量占了全表的 21%其中第一名的记录数是中位数的 600 多倍。这就是倾斜的根源。第三步我看的是聚合环节。SQL 里有一个count(distinct item_id)items 基数 3000 万这层去重把中间结果又放大了一遍。三个问题叠在一起维表重复导致 3 倍膨胀头部用户导致百倍倾斜去重统计再叠一层放大。难怪跑不动。5.3 改造代码与参数的具体调整改造分四步走。第一步维表去重落表。把用户维表按 user_id 收敛成唯一记录做成一张日更的维表同时在落表脚本里加了一个主键唯一性校验。INSERT OVERWRITE TABLE dim_user_uniq PARTITION(dt2024-05-20) SELECT user_id, city, level, register_ts FROM ( SELECT user_id, city, level, register_ts, row_number() OVER (PARTITION BY user_id ORDER BY update_time DESC, source_priority ASC) AS rn FROM dim_user_raw WHERE dt 2024-05-20 ) t WHERE rn 1;source_priority这个字段是我加的用来在多来源冲突时指定优先级比单纯按更新时间排序更符合业务预期。这一步做完关联的放大倍数从 3.1 倍降到了 1.0 倍光这一项就砍掉了 60% 的中间数据量。第二步商品维表改成广播。120MB 稍微超过了默认阈值我把spark.sql.autoBroadcastJoinThreshold调到 150MB同时把 Executor 内存从 32G 提到 40G给广播变量留出空间。第三步处理头部用户。抽样发现 Top 20 的用户贡献了 21% 的数据我把它们放进配置表跑的时候单独处理这 20 个用户的数据走加盐聚合盐值取 16其余数据正常跑最后 union。第四步改写count(distinct)。改成子查询去重加 group by同时对聚合阶段开了hive.groupby.skewindata的两阶段聚合。参数方面Shuffle 分区数从默认的 200 调到了 4096。5.4 资源账要算清楚别拍脑袋定参数分区数怎么算出来的当时改造后的 Shuffle Write 大约是 400GB我按每个分区 100MB 的目标来算400 × 1024 ÷ 100 ≈ 4096。这个 100MB 的取值不是随便定的它取决于 Executor 的核数和内存——每个 Task 处理 100MB 数据大约需要 300400MB 内存8 核的 Executor 并发跑 8 个 Task需要 3GB 左右的 Shuffle 内存在 40G 的 Executor 里留出这个空间是安全的。如果分区数定得更大比如 8192每个分区只有 50MBTask 数量翻倍调度开销反而会吃掉收益。广播变量的账也要算。150MB 的表100 个 Executor每个一份副本集群里实际占用 15GB 内存。这个数字听起来吓人但因为广播数据是只读的、可以被多个 Task 共享实际上比每个 Task 各持一份要省得多。真正的风险在于 Executor 内存不足时会触发驱逐所以我在调阈值的同时把 Executor 内存抬了 8G就是给这部分留余量。5.5 效果对比与遗留的三个小问题改造后的结果耗时从 8 小时 20 分降到 52 分钟Spill 总量从 6TB 降到 180GB最长 Stage 的 Task 耗时分布从 110 倍差距收敛到 4 倍以内没有再出现 OOM。中间结果行数从 250 亿降到 82 亿基本等于输入行数说明膨胀被彻底压住了。指标改造前改造后变化总耗时8h20m52m下降 89%中间结果行数250 亿82 亿下降 67%Shuffle Read 最大/中位数110 倍4 倍明显收敛Spill 总量6TB180GB下降 97%OOM 次数20消除遗留问题有三个。一是配置化的热点 Key 列表需要人工维护目前是每周更新一次偶尔会有新的大户漏掉。二是广播阈值调到 150MB 之后如果有新的维表超过这个体量需要重新评估。三是 union 回去的那部分逻辑让 SQL 变得比较长可读性下降我在代码里加了详细注释也把热点处理的逻辑封装成了公共宏避免每个任务重复写。6. 常见问题与排查技巧实录6.1 加盐之后还是慢问题往往出在这三处最常见的原因是盐值加错了位置。有些人给维表加盐、给事实表不加或者两边的盐值规则不一致导致关联不上最后数据全变成 NULL任务倒是跑快了结果是错的。我一般在改完之后一定跑一次总数比对改造前后的记录数差异必须能解释清楚否则不推进。第二个原因是盐值倍数不够。热点 Key 的数据量如果是平均值的 500 倍你只加 16 个盐那每个分区的数据量还是平均值的 31 倍依然倾斜。这种情况要么把盐值提到 64 或 128要么改用热点分离方案。我判断的标准是加盐后每个分区的数据量应该在平均分区大小的 2 倍以内超出就说明倍数不够。第三个原因是加盐破坏了原本可以利用的聚合下推。有些场景下加盐之后优化器没法做部分聚合导致中间结果反而变多。这种情况我建议先跑个 10% 的采样数据看看 Shuffle Write 的变化别直接上全量。6.2 广播 Join 报 OOM阈值不是越大越好广播 OOM 的报错信息通常比较隐晦表现为 Executor 的堆内存溢出堆栈里能看到BroadcastExchange相关的调用。遇到这种情况第一反应应该是把阈值调回去而不是加内存。我处理这类问题的顺序是先确认小表真实大小不是文件大小是反序列化后在内存里的大小通常是文件大小的 1.53 倍再看 Executor 的并发 Task 数和可用内存最后决定是调阈值还是换方案。如果一个表解压后有 400MB那广播基本没戏别硬撑老老实实走 SortMergeJoin 加两阶段聚合。还有一个折中方案是spark.sql.autoBroadcastJoinThreshold-1全局关掉自动广播然后对确定的表用 Hint 强制广播SELECT /* BROADCAST(t) */ ...。这样能精确控制哪些表走广播避免优化器误判。这个做法在表大小波动比较大的场景下很实用。6.3 磁盘被打满与任务重试的死循环膨胀最严重的后果不是慢是把磁盘写满。一旦磁盘满了Task 会失败重试重试又会产生新的中间文件越滚越大最后整个队列都被拖垮。我遇到过两次这种情况第二次之后我就在所有大任务的配置里加了两个保护spark.shuffle.spill.numElementsForceSpillThreshold设一个上限以及在提交脚本里做磁盘水位检查超过 85% 直接不提交。真遇到了也有救急办法先把任务停掉手动清理对应的 Shuffle 临时目录然后改成小批量跑按分区逐个处理跑完一个提交一个。这个办法笨但能保住当天的数据产出。事后一定要做根因分析别每次都靠分批跑糊过去。6.4 排查速查表现象最可能的原因优先排查动作卡在 99% 长时间不动Shuffle 倾斜看 Task 的 Shuffle Read Size 分布中间行数远超输入维表关联键重复按关联键统计维表重复度Executor 频繁 OOM广播表过大或窗口分区过大查广播表内存占用、看窗口分区大小磁盘快速写满Spill 失控或笛卡尔积查 Spill 量、核对 Join 条件任务重试次数很多数据倾斜导致的失败重试看失败 Task 是否集中在同一分区改造后结果对不上加盐或去重逻辑改变了数据改造前后做总数与明细抽样比对小文件数量爆炸分区数过多或写入未合并查输出文件平均大小加合并逻辑6.5 几条踩出来的经验第一条永远先做数据探查再改代码。我现在接到倾斜任务的第一步不是打开 SQL而是先跑几条分布统计把 Top Key、基数、中位数都摸清楚。这个动作花 10 分钟能省掉后面两小时的瞎试。第二条改完必须做前后比对。优化最容易出的事故不是没优化好而是优化完数据错了。我的习惯是拿一小部分数据比如某一天的某个分区跑改造前后两版比对总数、比对 Top 100 的明细确认一致才推全量。第三条把治理动作沉淀到模型层。维表去重、分区规范、字段裁剪这些事如果每次都靠人记早晚会漏。我现在会把主键唯一性校验、行数波动告警、单文件大小检查都做成落表任务的固定环节让问题在产出阶段就被拦住而不是等到下游任务跑挂了才发现。第四条别迷信一个参数解决所有问题。AQE 的自适应倾斜处理确实好用但它本质上是把大分区再切小对于真正的超级热点只是缓解。该改逻辑的时候还得改逻辑参数是辅助不是主力。最后分享一个小习惯我会给每个跑得慢的任务建一个简单的记录写上当时的基线、定位过程、改了哪几处、最后的效果。这个记录一开始只是为了自己不忘事后来发现它在遇到相似问题时价值极大——很多所谓的“新问题”翻一翻记录就能找到现成的解法省下的时间远超记录的成本。数据量和业务都在变但倾斜和膨胀的套路其实就那么几种摸透了之后剩下的都是熟练度的问题。
