企业官网建设流程全解析

凌晨一点，手机连震几下——调度平台告警：某个稳定跑了半年的标签聚合任务，在 Shuffle 阶段同时挂了三个 Executor，自动重试都没救回来。这是典型的Shuffle 阶段 OOM，很多人的第一反应是加内存，但直接加内存往往只会让问题更严重。这篇文章以一次真实排查为主线，把 Shuffle 阶段内存去向、OOM 类型判定、参数组合调优、代码级优化思路、以及调优后怎么验证全部串起来。适合正在维护 Spark 离线批处理任务、经常被 OOM 折磨的开发和数据平台同学，特别是那种每天跑批、一遇到大促数据翻倍就提心吊胆的人。

1. Shuffle阶段为什么是OOM重灾区——先弄清内存去哪了

1.1 Shuffle 过程到底在做什么，为什么这么吃内存

Shuffle 本质上就是按照某个 key 把数据重新分区。上游每个 task 算完自己的结果后，属于同一个下游 task 的数据要被拉到一起。这个“拉”的过程里，数据至少要在三个地方消耗内存。

第一是写侧缓冲区。每个 task 在写 Shuffle 数据时，不是一条条直接落盘，而是先写进一个内存 buffer，满了再溢写到磁盘。这个 buffer 默认是spark.shuffle.file.buffer，32KB。单个 buffer 不大，但是一个 Executor 上几十个 task 同时写，累计起来能吃掉几百 MB。这还不是主因，真正的重头戏在下面两个。

第二是读侧拉取窗口。下游 reducer 从上游拉数据时，有一个“同时最多在途数据量”的限制，也就是spark.reducer.maxSizeInFlight，默认 48MB。这个参数的意义是把“同时 hold 在内存里的 Shuffle 数据”限制在一个窗口内。很多人以为把它调大就能拉得更快，实际操作中它会直接放大 Executor 内部的内存峰值——尤其是并行度一高，几个 task 同时各拉 48MB，叠加起来不是线性翻倍，而是在聚合还没开始前就已经堆了一大片原始数据。

第三是聚合结构，这是 Shuffle 阶段 OOM 的绝对主战场。下游收到数据之后，如果要做 reduceByKey、groupByKey、distinct、join，会先把数据放进一个可以动态增长的聚合结构里。问题在于：这个结构只有在数据量超过内存预算之后才开始溢写，但溢写过程本身还需要额外内存，而且已经被聚合结构占住的空间不会立刻释放。如果某个 key 的条数极端多，这个结构永远降不下来。

打个比方：这就好比银行大厅，客户先在大厅排队等着叫号，超过一定人数才会被请到院子里。但如果某一个号被叫了几百万次，整个大厅就被这一个号占了，其他客户全得堵在外面。

1.2 两类 OOM 的快速分辨：报错信息比你想的有用

实际遇到的 OOM 并不总是一种，处理手段完全不一样。我从报错日志上把它们分成三类：

这里面最容易误判的是第二种。很多人看到 “Container killed” 就只去调大spark.yarn.executor.memoryOverhead，但如果堆内已经到极限，光加堆外没有任何意义。真正要做的，是先判断瓶颈是堆内还是堆外。

判断方法其实不复杂：看同一个 Executor 上的堆内使用率。如果 YARN 杀掉 Container 时，Spark UI 里显示的 Storage Memory 和 Execution Memory 都已经标红接近上限，说明堆内首先是满的，堆外只是压垮骆驼的最后一根稻草；如果堆内还剩下很多，但物理内存超了，那才是堆外的问题。

2. 一次真实的OOM告警：从日志到根因的排查链路

2.1 报错现场：同一个 Stage 反复失败，两种错误交替出现

那次任务是我们自研离线调度平台上的“用户行为标签聚合”，每天处理约 30 亿条行为日志，按用户 ID 聚合后输出品类偏好标签。任务运行到第 10 分钟左右时，某一个 Stage 开始大面积报错，而且报错还不止一种：

• 一部分 Executor 报的是java.lang.OutOfMemoryError: Java heap space，堆栈指向UnsafeExternalSorter.growPointerArrayIfNecessary。这个位置是外部排序器在给指针数组扩容时无法在堆内分配到内存，说明当时堆内确实挤爆了。
• 另一部分 Executor 报的是Container killed by YARN for exceeding memory limits. 10.5 GB of 10 GB physical memory used。说明 Container 的物理内存上限是 10GB（当时 Executor 是 8G，overhead 已经调到 2G），实际用到了 10.5GB，还是超了。

两种错误交替出现，这通常不是单一参数的问题，而是“执行内存预算 + 堆外开销 + 单 task 数据量”三个因素叠加。后面排查下来也确实印证了这个判断。

2.2 第一步：打开 Spark UI 定位异常 Task，而不是先改配置

遇到 OOM 先别急着调参，先去 Spark UI 的对应 Stage 页面看两个指标：Shuffle Read Size / Records和任务运行时长分布。

当时看到的画面非常典型：同一个 Stage 里，绝大多数 task 的 Shuffle Read 只有 50MB 到 200MB，但有一小撮 task 的 Shuffle Read 达到了 4GB 到 6GB，运行时长相差 20 倍以上。所有失败的任务，全部集中在处理最大 key 的那几个 task 上。这就是数据倾斜的直接证据。

注意，这里要强调一个很容易误判的点：数据倾斜不是配置错了，是上游数据本身在某几个 key 上严重聚集。我还记得当时查了上游日志，那个异常时间点正好赶上某个渠道的活动流量爆发，一个公共渠道 ID 在当天占了整批数据的 30% 以上。这种倾斜是任何单点参数都无法解决的——就算把 Executor 加到 32G，那一个 key 的聚合结果也放不进单个 task 的内存。

2.3 第二步：逐层排除，确认不是参数“恰好不够”而是结构问题

参数排查有一个固定顺序，这一步很多人做得不够仔细。我的排查链路是这样的：

1. 先看 Executor 内存与核数的配比。当时集群给的是 4 核配 8G 内存。一个 Executor 上同时跑 4 个 task，每个 task 都在做 Shuffle Read 和大 key 聚合，内存池很容易被冲垮。核数越多，同时抢内存的任务越多，OOM 概率越大。所以面临 OOM 时，不能只盯内存本身，还得看并发 task 数量。

2. 再看统一内存管理器的两个核心比例。Spark 2.x 之后用的是统一内存管理器，spark.memory.fraction默认 0.6，意思是堆内最多 60% 分给执行和存储共用；其中spark.memory.storageFraction默认 0.5，意思是这 60% 里面再留一半给 Storage 缓存。那个任务基本没有 RDD 缓存，Storage 部分完全闲置，执行内存实际只有不到 3G，这才是堆内爆掉的根本预算问题。

3. 算一笔账。8G Executor，减去 JVM 保留开销，可用堆约 7G 左右，执行+存储共约 4.2G，其中执行约 2.1G。而一个倾斜 task 的 Shuffle Read 数据是 4-6G，还要加上聚合结构，怎么可能不爆。

4. 再看堆外。YARN Container killed 说明堆外也有开销。RPC 通信、NIO、Executor 内部线程栈、Direct ByteBuffer，都会吃spark.yarn.executor.memoryOverhead。默认值是 Executor 内存的 10%，也就是 0.8G，明显偏紧。但我们调到 2G 后还是被杀了，说明堆外只是被拖累，核心问题仍然是堆内聚合塞满后，JVM 为了完成 GC 或其他操作占用了额外物理内存。

把这几条串起来，根因就很清楚了：数据倾斜导致少数 task 要独自扛起几 GB 的 Shuffle Read，堆内执行预算又只有 2G 左右，聚合结构一膨胀就直接堆内爆掉；与此同时，堆外 overhead 也不够，另一个 Executor 在堆内接近满员的状态下被物理内存上限杀掉。

2.4 定位根因时容易被忽略的三个细节

排查这种问题，有三个经验值得单独列出来，因为它们都不是一眼能看出来的。

第一，别只看“Executor OOM”就调大 executor memory。调大内存是兜底手段，不是首要手段。如果数据结构大、GC 频繁，加大内存反而让 Full GC 时间更长，任务更慢。我用同样的数据在测试环境把 Executor 内存从 8G 加到 16G，Shuffle 阶段是没报了，但整体运行时间反而多了 12%。

第二，一定要看 Executor 页面的 GC Time。一个 Executor 的 GC 时间如果超过运行时间的 20%，说明堆内碎片严重，大量对象一直在被拷贝和晋升。这是聚合结构长期不释放的信号，比看报错更能说明问题。

第三，查上游数据的时间窗口。标签聚合这类任务有个特点：数据量按天波动，有大促活动时瞬间翻倍。只靠参数是防不住这类“偶发性倾斜”的，必须从数据侧验证当天是否有特殊流量。后来我们确实在上游日志里看到了那个公共渠道 ID 的异常峰值。

3. 关键参数调优：真正改过的那些参数和组合

3.1 先说结论：这组参数改完之后任务稳定运行

当时最终确定的参数组合，写在 Executor 8G、4 核的配置上：

spark-submit \ --conf spark.sql.shuffle.partitions=800 \ --conf spark.memory.storageFraction=0.3 \ --conf spark.yarn.executor.memoryOverhead=1g \ --conf spark.shuffle.file.buffer=64k \ --conf spark.reducer.maxSizeInFlight=24m \ --conf spark.shuffle.sort.bypassMergeThreshold=400

直接看可能觉得奇怪：内存没加，反而把storageFraction降了、把maxSizeInFlight减了。但实际效果就是稳定了。原因是这几个改动一起把“单 task 数据峰值”和“执行内存预算”匹配了起来。数据倾斜依然存在，但被新的分区数和窗口控住了。

3.2 每个参数的作用原理与失效场景

spark.sql.shuffle.partitions：从 200 到 800。

这是 SQL 和 DataFrame 操作的 shuffle 分区数。默认 200 的问题在于，如果你处理的是几十亿条数据，每个分区平均要处理 1500 万条以上，单 task 的内存压力巨大。改成 800 后，单 task 数据量变成原来的四分之一，聚合结构更容易落进执行内存预算，溢写概率大幅下降。

但注意，分区数不能无限调大。分区越多，shuffle write 产生的文件块越多，网络连接数越多，调度开销越大。我见过有人把分区数直接调到 2000，结果 Shuffle 阶段建立连接的时间比计算时间还长。核心思路是让“单分区数据量”和“单 task 内存预算”匹配，不是越大越好。

spark.memory.storageFraction：从 0.5 到 0.3。

这个参数很多人理解不到位。统一内存管理器下，memory.fraction决定执行+存储总共能用多少堆内，storageFraction决定在这部分内部存储优先占多大比例。当任务不缓存 RDD 时，存储部分完全闲置，但闲置部分不会自动全部让给执行——它只在执行内存吃紧时才被借用，而借用的代价是可能需要淘汰存储数据。把storageFraction降到 0.3，等于直接给执行池多留了约 20% 的执行内存。这个改动对纯 Shuffle 型任务收益非常明显。

spark.yarn.executor.memoryOverhead：比例改绝对值，1G。

在 YARN 模式下，这个参数既影响 Container 物理内存上限的判定，也决定堆外能分配多少。默认值是 Executor 内存的 10%，8G 就是 0.8G，对 NIO 开销略大的任务来说很紧张。我建议小内存 Executor 直接用绝对值，逻辑更可预期。但如果你 Executor 本身是 32G，百分比可能更合适，这个要按集群规格来。

spark.shuffle.file.buffer：从 32K 到 64K。

这个参数影响写侧缓冲。调大之后，每个 task 在写 shuffle 数据时溢写小文件的数量会减少，磁盘 IO 次数降低。但注意，它只影响写侧，不解决下游聚合内存问题。很多教程喜欢把这个问题吹得无所不能，实际对于多数的读侧 OOM 完全无效。只有当你的任务有大量 spill 时，这个参数才有意义。

spark.reducer.maxSizeInFlight：从 48M 到 24M。

这个参数代表每个 reducer 一次最多拉取多少数据。它本质是流量窗口，控制的是“同时 hold 在内存里的数据上限”，不是总拉取量。分区数改成 800 之后，如果窗口还保持 48M，那么每个 task 依然可能在高峰期拉满 48M，在聚合结构还没释放时又叠加 48M 的原始数据，风险反而增加。降到 24M 后，分区数多了，单个 task 的拉取任务变轻，整体吞吐没下降，内存峰值却小了很多。

spark.shuffle.sort.bypassMergeThreshold：从 200 到 400。

这个参数解决的问题是：当 shuffle 分区数少于这个阈值时，shuffle write 会走一个不排序的合并路径，避免做无谓的排序。分区数从 200 调到 800 之后，如果阈值还停留在默认 200，那所有超过 200 的分区都要走排序路径，白白浪费 CPU。调到 400 之后，至少 400 个分区以内的 shuffle 都可以跳过排序。这是最容易被忽略的“隐性开销”之一。

3.3 参数调优不能一次性全改：记录与验证方法论

参数调优有一个容易踩坑的习惯：一次性把所有参数全改掉，跑一轮成功就以为大功告成。这种做法的问题是，如果出了问题，你根本不知道是哪个参数导致的；而且很多参数之间有交互作用，单看一个参数无法判断收益。

我的做法是每次只动一个维度，跑一组冒烟数据做对照。比如先固定 Executor 内存和 core，只动shuffle.partitions，记录 Shuffle Read 均值和 P99 是否下降；再动memory.storageFraction，看 GC Time 是否下降；最后动maxSizeInFlight，看单个 task 的内存峰值。每一步都从 Spark UI 记录两个数据：平均运行时长和GC 时长。只有两项都朝好的方向走，才保留改动。

4. 代码层面的优化：比调参更彻底的方案

4.1 尽量规避 Shuffle：从源头减少数据搬运

调参只是让同样结构的问题不再发生，真正长期有效的方案是在代码上做减法。那次之后我把整个标签聚合代码重写了一遍，效果比参数调整更明显。

第一，能用reduceByKey/aggregateByKey就绝不用groupByKey。reduceByKey会在每个分区内先合并一遍再 shuffle，shuffle 过去的数据量可能只有原来的五分之一甚至十分之一。这是成本最低、收益最直接的改动。

// 不推荐：先全量 group 再 sum，所有数据全部 shuffle rdd.groupByKey().mapValues(_.sum) // 推荐：map 端预聚合，shuffle 数据量大幅下降 rdd.reduceByKey(_ + _)

第二，用广播变量代替 join。如果有一边是小表（几百 MB 以内），直接做成广播变量，完全跳过 shuffle join。那次任务里有一个品类字典表只有 30MB，最初用的是普通 join，每个 task 都要拉全量，后来改成 broadcast join，整体耗时缩短了约 30%。

第三，提前过滤无意义的 key。有些 key 的数据根本不需要输出，用 filter 在聚合前处理掉，而不是聚合后再丢。这个改动在数据量大的时候效果很明显，省掉的是整个 shuffle 过程中的无用数据传输。

4.2 倾斜 key 的加盐与两阶段聚合

数据倾斜是 Shuffle 阶段 OOM 的高频诱因，而且经常是“偶发性倾斜”——平时没感觉，一到某些时间窗口就爆发。彻底方案是“加盐 + 两阶段聚合”：

1. 先用 SQL 或 DataFrame 把 key 分组计数，找出超过某个阈值（比如总行数的 1%）的热点 key。
2. 对热点 key 加一个随机前缀（比如 0-99 的随机数），让它们分散到 100 个分区里。
3. 第一阶段先按“加了前缀的 key”做聚合，输出中间结果。
4. 第二阶段去掉前缀，再按真实 key 聚合一次。

from pyspark.sql import functions as F # 1. 找出热点 key：row_number / count 统计 # 2. 对热点 key 加盐（0-99 随机数） df_salted = df.withColumn( "salted_key", F.when(F.col("is_hot"), F.concat(F.col("key"), F.lit("_"), F.rand() * 100)) .otherwise(F.col("key")) ) # 3. 第一阶段按 salted_key 聚合 stage1 = df_salted.groupBy("salted_key").agg(F.sum("value").alias("sum_value")) # 4. 去掉盐，第二阶段按真实 key 聚合 stage2 = stage1.withColumn("key", F.split("salted_key", "_")[0]) \ .groupBy("key").agg(F.sum("sum_value").alias("sum_value"))

这样做之后，单个热 key 的聚合压力从“一个分区打满”变成“100 个分区共同承担”。那次 12 亿条数据的渠道 key，加盐后分成 100 份，每份 1200 万条，再结合 800 个分区，单 task 最多扛 1200 万条聚合，内存完全够用。

注意加盐的粒度：前缀数太小等于没分，太大则每个分区都要读一遍全量数据，第二阶段耗时上升。一般取 50-200 之间，根据热点 key 的条数来算，条数越多前缀值相对越大。

4.3 数据结构与序列化优化：堆内的隐形杀手

有时候 OOM 不是数据多，而是数据结构太废内存。几个典型的坑：

• 使用默认 Java 序列化存储了大量对象。Java 序列化的膨胀率动辄 3-5 倍，对象在内存里还带对象头。换 Kryo 序列化之后，同一个数据量级的内存占用能降 30%-50%。
• 在聚合逻辑里创建大量临时对象。比如在 map 阶段对每条记录new一个包装对象，几亿条记录就是几亿个对象，GC 压力巨大。改成复用对象，或者用 DataFrame/Dataset 的表达式完成计算，避免显式创建中间对象。
• 大量 String 类型的 key。如果 key 是长字符串，可以考虑用哈希或字典映射成整数 ID，聚合阶段在整数上做，内存占用大幅下降。

4.4 避免重复 Shuffle 的链路重排

还有一类问题是同一个任务里出现了多次 Shuffle，而其中一部分完全可以合并。比如有些代码会在 join 之后做 groupByKey，又在聚合结果上做另一个 join——每次都触发一次 Shuffle。如果能通过调整算子顺序，让同 key 的重分布只发生一次，内存压力和总耗时都会显著下降。

具体做法是：先通过 broadcast 把字典表送过去，再在 map 阶段做关联，最后只做一次聚合 Shuffle。那次改造之后，Shuffle 次数从 3 次降到 1 次，总运行时长从接近 50 分钟降到 22 分钟。这是所有优化里收益最大的一步。

5. 验证与监控：调优之后怎么确认问题真正解决

5.1 从 Spark UI 读什么指标才算真稳定

调优完了不能只看任务成功。我在任务跑完第一轮之后做了三件事，这三件事建议你也加进检查清单。

第一，看 Shuffle Spill 指标。在 Spark UI 的 Stage 详情里，看 Spill (Memory) 和 Spill (Disk) 两列。如果磁盘 spill 的数据量很大，说明执行内存还是不够，聚合结构一直在往磁盘吐数据。理想状态是 spill 不出现，或者只出现在少数 task 上。那次调整后，磁盘 spill 从几 GB 降到了几十 MB。

第二，看 GC Time。进入 Executor 页面，观察 Executor 的 GC 时长。调整后第一轮任务的 GC 时间占比从原来的 33% 降到 8% 左右。GC 时间是个很好的辅助指标，比运行时长更能反映内存问题是否解决。

第三，看 Task 运行时长分布。同一个 Stage 里的 max/median 比值。之前经常超过 20，之后基本在 2-3 之间，说明倾斜 key 的影响已经被摊平。

5.2 给任务加上与 Shuffle 相关的告警

数据倾斜这类问题不会每次都报 OOM，更多时候是“变慢、长时间不完成、偶尔成功偶尔失败”，所以告警指标不能只盯着 OOM。我在监控面板上加了这几个指标：

• Shuffle Read Bytes 超阈值告警（按任务历史均值动态计算）
• Stage 耗时 P95 超过 30 分钟告警
• Executor 物理内存使用率超过 85% 且持续 5 分钟告警
• 任务重试次数超过 2 次告警

另外加了一组“结果正确性”校验：对聚合后的输出行数做断言，如果输出行数与预期相差过大，立即告警。这能快速暴露那些“进程没死但数据不对”的隐形问题。

5.3 调优后遗留的注意事项

有一点必须坦白：这次调整不是一劳永逸的。数据分布会变，倾斜 key 会换，活动规模也会变。最好的策略是把“每次跑批的 Shuffle Read / 运行时长 / 最大 task 耗时”三个指标形成基线，每次跑完对一下，超过基线的 20% 就主动排查。

参数调优的目标，不是让任务永远不出问题，而是让它在问题发生前有足够余量、发生后能快速定位根因。这次调整的所有参数和当时的运行版本我都整理成了配置单，建议你也这么做。很多问题是三个月后换一批人又重新踩一遍的，一份配置单加上监控基线，能省掉大量排查时间。