在跑离线任务的时候，最让人头疼的不是计算慢，而是明明前一天还稳定的作业，第二天突然在Shuffle阶段大面积OOM。一个接一个的executor被打爆，任务反复重试，监控面板红色告警不断。尤其在大促或月末跑数这种时间窗口紧张的时候，遇到这种问题，连排查都是抢时间的。
我这几年处理过不少类似的故障，从七八台小集群到几十台的中型集群都遇到过。Shuffle阶段的OOM，表象就那几个，但真正的原因往往是数据倾斜、参数不匹配、代码低效三类问题混在一起，单靠“把executor内存调大”解决不了，甚至会把问题拖得更久。这篇文章不写大而全的参数手册，只讲我在实际排查和调优中验证过的思路、顺序和细节，希望能让遇到同类问题的人少走弯路。
1. Shuffle阶段为什么总是“内存杀手”
1.1 Shuffle过程拆解：Map端与Reduce端的内存占用
Shuffle说白了就是把数据从上游的Map任务“理顺”，再按key分发到下游的Reduce任务去。这个“理顺+分发”的过程，天然比普通的map/filter操作更吃内存。
Map端在做ShuffleWrite的时候，每个task会把输出的key-value先写进一个内存缓冲区。基于排序的shuffle会在这个缓冲区里做排序，排完如果放不下了，就spill（溢写）到磁盘。这里的问题在于，如果某个分区命中的数据特别多，这个缓冲区就会反复被写满、spill、排序，既有大量磁盘IO，又占内存。
Reduce端更难控制。每个Reduce任务要从上游所有Map任务拉取属于自己的数据，拉下来的数据先落在内存里的一个聚合缓冲中，再做combine或者aggregation。如果你用的是groupByKey这类操作，同一个key的所有value必须全部留在内存里等着聚合，这几乎就是往内存里塞一整份分布式的数据。
所以Shuffle阶段的OOM，绝大多数不是“哪一行代码写得明显错了”，而是“某个环节的数据量远超预期，内存缓冲被直接打穿”。这也是为什么排查难度高——你光看代码往往什么都看不出来。
1.2 看懂OOM报错的三种现场
我在排查OOM时，第一件事是看报错的类型。不同报错对应的修复方向完全不同，如果用错药方，越改越乱。
最常见的三种现场：
java.lang.OutOfMemoryError: Java heap space。这个最直白，说明JVM堆内空间被对象全占满了。一般出现在Reduce端的聚合阶段，或者Map端spill之前的缓冲区膨胀。此时优先怀疑shuffle分区数太少、单个task数据过大、或者GroupByKey/HashAggregation把数据堆积在了内存里。
java.lang.OutOfMemoryError: Direct buffer memory。堆外内存爆了，也就是direct buffer或各种native层缓冲区占满了。这种一般和spark.executor.memoryOverhead设置不足有关系，尤其是K8s或Yarn环境下，container里面除了JVM堆还有一堆元空间和网络缓冲。这种现场最容易被人误判成“堆不够”，然后盲目加executor.memory，结果越加越崩。
ExecutorLostFailure / FetchFailedException。严格说这不算OOM，但实际场景里它经常和OOM伴随发生。一个executor崩了之后，其他executor去拉它的数据块拉不到，就会报FetchFailed。很多人被这个误导，以为要调节流参数或者网络参数，实际上根因还是某个executor先内存爆了，controller在这里需要先把ExecutorLostFailure当作“后续现象”处理，而不是只盯着FetchFailed调fetch参数。
1.3 自我诊断：先分清“真OOM”还是“被杀”
这里有个特别普遍的坑，就是你看着Yarn或者K8s的事件里写的是“Container killed by OOM killer”，但实际上JVM日志里根本没有OutOfMemoryError。
我遇到过好几次这类情况：Yarn上报executor被物理机层面的oom killer杀了，作业日志里却干干净净，完全没有OOM堆栈。新手一般就没方向了，老手会先去查spark.executor.memoryOverhead。
原因不复杂。Spark里的executor内存配置其实只是JVM堆大小，但真正的容器内存等于堆内存 + overhead内存。overhead主要覆盖线程栈、元空间、JNI、网络缓冲区、Direct Memory等。
很多人设了executor.memory=8g，却没设置memoryOverhead，默认值是“max(384MB, 8g*10%)”，也就是约819MB。如果8g堆本身真用到了6~7GB，JVM里虽然还没报OOM，但进程整体占用已经逼近甚至超过了container限额，内核一看超限直接杀。这种情况你去调大executor.memory，往往是让限制变大，但同时物理机内存压力也变大，很容易引发雪崩。
所以遇到executor被杀，第一反应不是调大堆，而是先看overhead和物理机总内存规划。
2. 排查思路：从UI指标到根因定位
2.1 第一步不是调参，而是看监控
我见过太多人一上线就改参数，这里加两个G，那里调大并发，结果问题原地踏步。调优这事，第一步永远是观察。
打开Spark Web UI，重点看三个页面：
Stage页面：能看到每个Stage的Shuffle Read/Write数据量。我一般先看失败的那个Stage，把失败task的Shuffle Read字节数和成功task比一比。如果出现“几个task拉到几百MB，其他task只有几十MB”这种悬殊差距，基本可以断定是数据倾斜。
Executors页面：这里能看到每个executor的Storage Memory、Shuffle Spill (Memory)、Shuffle Spill (Disk)等指标。如果Shuffle Spill (Memory)的峰值特别高，说明缓冲数据接近内存极限，已经在持续溢写了；如果Shuffle Spill (Disk)也大，说明内存已经兜不住了，溢写压力巨大，磁盘IO也会拖慢整个作业。
Event Timeline：配合看executor是被驱逐、被kill，还是任务超时重试。是否伴随“lost executor”或“node blacklisted”等事件。
看的时候记得把时间范围和失败时间对上，别拿整个作业的聚合数据去判断，那样会把极端的task数据稀释掉。我习惯把失败Stage的上万个task按Shuffle Read大小拉出来排序，只看头部几十个，倾斜情况一目了然。
2.2 内存模型前置知识：堆内划分与堆外开销
要调Shuffle的内存参数，必须先理解Spark 2.x之后的内存管理模型。很多人盯着spark.executor.memory这个值，却不知道JVM堆里其实分了好几块。
简化来讲，executor堆内分三块：
Reserved Memory：固定300MB左右，保留内存，存Spark内部对象用，不参与其他分配。
User Memory：占比为(1 - spark.memory.fraction)。默认memory.fraction是0.6，所以User Memory占堆的40%。这里存用户代码里创建的对象、数据结构、以及通过mapPartitions这类算子处理时产生的临时对象。用户代码如果大量生成Java对象却不释放，挤的就是这里。
Unified Memory（统一内存）：由spark.memory.fraction决定，默认占堆的60%，又被分为Storage内存和Execution内存，各占一半（由storageFraction控制，默认0.5）。Storage内存用来缓存RDD、DataFrame等数据块，Execution内存用来给shuffle聚合、排序、join等操作使用。关键设计是两边可以互相借用，但Execution借Storage内存在某些条件下可以把Storage的缓存块挤出去。
我在实际调Shuffle时最关心的其实是Execution Memory这一块。Shuffle聚合、merge、sort都在这块内存里进行。如果storageFraction默认0.5，意味着统一内存里一半被Storage占着。如果你同时在作业里缓存了大量数据集，Storage内存占得满满当当，Execution内存即使能借用，也在激烈竞争，shuffle阶段的内存压力会显著上升。
所以一种很实用的战术就是：如果作业里没有大量缓存需求，把spark.memory.storageFraction调低，比如0.2~0.3，相当于把统一内存更多留给Execution，给Shuffle腾地方。这个调整在很多场景下比重接加executor内存还管用。
2.3 用GC日志和堆转储找具体“凶手”
UI只能看到表象，要确定具体是什么对象把内存吃满了，还得靠GC日志。
我给executor配的extraJavaOptions里常带这几项：
-XX:+PrintGCDetails -XX:+PrintGCDateStamps -XX:+PrintTenuringDistribution然后把日志输出到容器日志目录。出问题后把GC日志拉出来，看两点：
- 是否反复Full GC？如果是，说明对象一直在老年代堆积，且无法回收。
- 单次Full GC后老年代占用是否只降一点点？如果是，说明内存里有大量长生命周期的大对象，极可能就是shuffle聚合用的集合或缓存块。
GC日志配合jmap是更好，但容器崩了就没办法连上去dump，所以我的经验是：优先在Spark UI和GC日志里找结论，堆转储只是最后手段。如果真想dump，要在executor还没死之前用jmap把堆导出来，然后灌进MAT或VisualVM看柱状图。shuffle阶段里最常出现的“元凶”就是byte[]、Object[]或者ConcurrentHashMap$Node，前者对应大量序列化后的缓存数据，后两者对应shuffle聚合时的集合结构。
2.4 快速判断数据倾斜：task间数据量对比
数据倾斜是Shuffle OOM的头号原因。判断方法很直接：在Stage页找到最后一个失败的stage，点进去看Summary Metrics或Tasks列表，按Shuffle Read Size排序。
比如某个任务的Shuffle Read Size中位数是20MB，但是某3个task分到了800MB，这就是教科书级的倾斜。再进一步，到SQL的物理计划里，看这个Stage对应的SQL片段是group by哪个键，基本就能锁定热点key。
有些时候肉眼不好定，可以写一个简单的查询去统计key分布：
SELECT key, COUNT(*) AS cnt FROM ( SELECT your_shuffle_key AS key FROM the_table WHERE dt = '2025-01-01' ) t GROUP BY key ORDER BY cnt DESC LIMIT 20;能看到前几个key的数量级是后面的几十倍，就实锤了。这类倾斜处理我放在后面案例里详细讲，因为它不是单纯调参数能解决的，往往需要改代码逻辑。
3. shuffle调优关键参数逐项拆解
3.1 executor内存与进程内存的边界
先明确一组最小换算关系：
进程内存（container memory）= spark.executor.memory（JVM堆） + spark.executor.memoryOverhead（元空间、栈、DirectMemory等）这里的memoryOverhead在Yarn和K8s上都适用（K8s场景用spark.executor.kubernetes.memoryOverheadFactor，但底层逻辑一样）。设executor.memory=8g时，实际进程大约占8g + 1g（默认10%），再加上NodeManager自身的开销。所以单台物理机分配executor时，要把这层“看不见的内存”算进去，否则物理内存会被悄悄吃满，出现上一节说的“被杀”现象。
我通常给executor内存设置时遵循一个经验：
- 单Executor内存不要超过物理机内存的四分之一到三分之一。比如64G的机器，单个executor最多设16~20G（含overhead），宁可多开几个小executor，也好过一个巨大的executor把机器内存全占完。
- Executor核数控制在3~5个。核数太多时，一个executor上多个并发task同时做shuffle聚合，内存压力是叠加的，GC也会变得激烈，反而不稳。
- 在大集群里优先用中小内存、中核数（比如8~12G堆、4核），配合spark.dynamicAllocation弹性扩缩容，稳定性通常都优于“大内存executor”。
3.2 read端与write端关键参数表
下面是我在实际调优中常动的几组Shuffle参数。不同计算引擎或部署形态可能默认值有差异，但核心逻辑是相近的：
| 参数 | 说明 | 我的常用设置参考 |
|---|---|---|
| spark.sql.shuffle.partitions | 决定shuffle后的分区数/Reduce任务数 | 缺省200；按数据量调，通常100~1000 |
| spark.reducer.maxSizeInFlight | 单个Reduce Task每次同时拉取的shuffle数据字节上限 | 默认48MB，内存紧张时不要动；内存充足时最多拉到96MB |
| spark.reducer.maxReqsInFlight | 同时发起拉取请求的上限 | 保持默认 |
| spark.reducer.maxBlocksInFlightPerAddress | 从同一上游Executive同时拉取的块数上限 | 默认5，可以不变 |
| spark.shuffle.file.buffer | Map端shuffle写文件时的缓冲大小 | 默认32KB，一般不动 |
| spark.shuffle.spill.compress | 是否压缩spill到磁盘的数据 | 默认true，保持开启 |
| spark.shuffle.sort.bypassMergeThreshold | 小于此分区数时shuffle走bypass路线，不排序只合并 | 默认200，分区数较少时可以不开 |
| spark.shuffle.sort.spill.threshold | 触发spill的缓冲记录数阈值（旧参数） | 新版本不建议手动改 |
| spark.memory.fraction | 统一内存占堆的比例 | 默认0.6 |
| spark.memory.storageFraction | 统一内存中Storage占的比例 | 默认0.5；无大量缓存时可降到0.2~0.3 |
| spark.executor.memoryOverhead | 堆外内存预留 | 显式设置，如2g~4g，或按堆的10~20% |
| spark.sql.autoBroadcastJoinThreshold | 小表自动广播的阈值 | 默认10MB，按实际可用内存调整，谨慎放大 |
| spark.sql.adaptive.enabled | 开启AQE | Spark 3.x建议开启 |
| spark.sql.adaptive.skewJoin.enabled | AQE自动处理倾斜Join | Spark 3.2+建议开启 |
这张表不是让你全调一遍，多数时候动其中两三个就够了，后面案例里你会看到选择顺序。
3.3 并行度与分区数怎么定
Shuffle OOM的一个隐蔽推手是Task并行度太低。Spark默认shuffle分区数是200，如果数据量有几百GB，200个task每个要拉取1GB以上，放在8G内存的executor里，不死也脱层皮。
比较合理的思路是按照“单个Reduce Task的输入量约等于50~200MB”来倒推分区数。假设Stage的Shuffle Read总量是200GB，想让单task拉到100MB左右，那么分区数就应该设为2000左右。
并行度也不是越高越好。分区数过高会带来两个新问题：
- 大量小task导致调度开销变大，运行时间反而变长。
- 下游stage的输出被切成几千个小文件，影响后续读写效率和文件系统压力。
所以我在调分区数时，会同时看两个指标：Shuffle Read Size是否回到合理区间，以及最终生成的输出文件是否过多。一般有个公式参考：
分区数 ≥ 集群总executor核数 × 2如果数据量大，再往这个基础上按数据量放大。Spark SQL作业里直接设spark.sql.shuffle.partitions，RDD作业里就是rdd.repartition(n)或coalesce(n, shuffle = true)。
3.4 高级手段：AQE自动优化与广播阈值调整
如果你的引擎还在Spark 2.x，Shuffle调优基本上靠手动。如果已经是Spark 3.x，强烈建议开启AQE（Adaptive Query Execution），它能在运行时动态合并过小分区、自动感知倾斜、自动调整join策略。
几个开AQE后变化巨大的场景：
- 动态合并分区。默认shuffle分区数设很大，但实际大部分task数据量并不大。AQE会把这些小task合并，避免调度开销，所需内存也自然下降。
- 自动处理Join倾斜。开启spark.sql.adaptive.skewJoin.enabled后，遇到倾斜的join key，它会把倾斜端拆分，增加并发来拉平负载。这跟手动加盐的思路一致，但免去了改SQL的成本。
- 动态优化Join策略。小表小于阈值时自动从SortMergeJoin切成Broadcast Hash Join，规避大Shuffle。
关于broadcast阈值，我见过有人为了全用broadcast join，把autoBroadcastJoinThreshold开到几百MB，结果Driver端或者每个executor都缓存了一份大表，内存直接爆炸。广播表本质上是“以内存换Shuffle”，要结合executor内存和表大小来决定。我的建议是：先保证广播表序列化后的大小不超过executor堆的十分之一，再考虑调阈值。
4. Shuffle OOM典型调优流程实战
4.1 案例背景与第一现场
我之前处理过一个某业务线跑离线日志分析的场景。环境大致是20多台物理机，单机64GB内存、16核CPU。任务每天处理数百GB到1~2TB的日志数据，核心逻辑是先用一个大事务表和一个用户维表做join，再按渠道、时间维度做指标聚合。
某天开始，任务在join之后的第一个聚合Stage大面积报优：
ExecutorLostFailure (executor 11 exited caused by Container killed by OOM killer)伴随部分FetchFailedException，任务反复重试，最终失败。
第一轮我先加了executor内存，从8G调到12G，同时也把shuffle分区数从默认200调到500。结果面临的只是把失败时间往后推了半小时，最终还是崩。这里其实就暴露了我自己需要反思的一点：没有先观察UI就直接堆参数，这是典型错误。旁边监控一打开，才发现失败Stage里少数task的Shuffle Read Size高达700MB，大多数task只有几十MB——典型的倾斜场景，而不是单纯内存不够。
4.2 逐步定位：从倾斜key到参数修正
确认倾斜后，第一步先定位热点key，写了个统计SQL把join key拿出来按count排序，发现某个渠道ID的记录数至少是其他渠道的几十倍。业务上这个渠道是主推活动带来的流量，数据多不奇怪，但走Shuffle时全部挤到同一个task上，就直接把内存打爆了。
这种场景我给两个方向：
方向一：改SQL，对倾斜key单独处理。思路是先把热点key和大表分离开，对热点key在map端提前过滤或广播处理，不和全量数据一起走SortMergeJoin。
实现上可以先取出热点key，再对热点key所在的小范围单独做一次map端关键值校验或者广播join，而非热点数据继续走常规join，最后用union把结果合并。这个方案改动稍大，但效果最可控。
方向二：代码成语。在join key上拼接一个0到N的随机前缀，把之前的数据打散成N份。这个方法实现简单，但要注意防“数据爆炸”——如果不拆大表对应的key，join会产生N倍笛卡尔积，反而更爆炸。因此在加随机盐时，我会对大表侧也做对应改造，让两边都带上相同盐值再join，得到结果后再去掉盐值聚合。
实际这个案例里，我对热度key做了单独处理，把热点key的数据量摊到10个task上，同时保持其他key不走随机逻辑，这样既解决了倾斜，又没有放大join量。
参数上也同步做了几件具体的事：
- 分区数调到1200。因为总量有1TB以上，200个分区每个task平均5GB多，明显不合理；1200分区后单task均值约800MB，可控了不少。
- 把sort merge join换成broadcast join：那个用户维表实际只有几百MB，我把它手动广播了，同时将autoBroadcastJoinThreshold调大到一个相对合适的值。这一改，join这关完全不做大Shuffle，后续聚合的压力少了一大截。
- 开启AQE并打开自动倾斜处理，当作兜底。
同时我把executor核数从6降到4，让并发task数变少，每个task拿到的内存资源更充沛，GC压力也下来了一些。executor内存保持12G，overhead显式设到2G，避免“看起来堆没满实际上进程被杀”的情况再出现。
4.3 调优效果与“为什么有效”
调完之后，同一份作业从原先的3小时左右缩短到约1小时，且不再OOM。失败率归零。
回头看为什么有效，简要说三点：
- 先手处理倾斜，而不是硬扛内存。倾斜才是导致局部内存破口的直接原因，加内存只是给“破口”变宽的墙，但水压不变，迟早还是漏。
- 减少不必要的Shuffle数据量。广播维表、避免大表全量Join，等于从源头把Shuffle总流量降下来，内存压力自然缓解。
- 让内存分配更匹配Task的实际规模。于是并行度高了、并发task少了、堆外预留充足了，每个task能分到的内存空间处于“刚好够用但不会溢出”的区间。
这里我也提醒一句：加随机盐处理倾斜时，盐的倍数不是越大越好。曾经有人直接随机加100个盐，等于把数据量平空放大100倍，不仅没有解决，反而把整个作业的IO和时间拖垮。做盐值处理前，最好先确认热点key的数据量和其他key差几个数量级，再定加几份盐。
5. 参数调优后仍然OOM的常见坑
5.1 高频误区速查
光套参数而不理解内存模型，很容易掉进下面的坑。
| 误区 | 本质 | 解决办法 |
|---|---|---|
| 一直加大executor内存 | 不解决局部倾斜，物理机内存可能被吃满 | 先查倾斜，再按需加内存 |
| 忘了调memoryOverhead | 进程被杀而非JVM OOM | 显式调spark.executor.memoryOverhead |
| groupByKey用得多 | 同key值在内存中无法增量聚合 | 改用reduceByKey/aggregateByKey |
| storageFraction过高 | Storage缓存挤压Execution内存 | 无缓存时降到0.2~0.3 |
| 分区数过多 | 小文件/调度开销反而压垮任务 | 按“单task几十~几百MB”校准 |
| 广播阈值盲目调大 | 大表广播占满所有executor堆 | 上限为executor堆的10% |
| 忽视spill指标 | 已疯狂溢写还不调内存 | 看Shuffle Spill (Memory/Disk)评估 |
| 没开AQE | 手动优化滞后且漏掉倾斜 | Spark3开启adaptive+skewJoin |
| 只信报错文字不看UI | 拿FetchFailed当网络问题 | 先定位ExecutorLost根因 |
| 一次调多个参数 | 无法区分是谁起效 | 一次只改一个变量并记录 |
5.2 针对典型场景的对症措施
如果你还没理清方向，可以参考下面这套选择逻辑：
- 现象是某个executor直接被杀，且JVM没报OOM堆栈 → 优先调memoryOverhead，检查单机总内存分配。
- 现象是整个Stage所有task都报heap space → 优先并行度/分区数，再看executor内存是否真的过小。
- 现象是一部分task报OOM，另一部分跑得很快 → 数据倾斜，优先加盐或拆分热点key，同时配上AQE skewJoin。
- 现象是Shuffle Spill (Disk)巨大，作业极慢但未必OOM → 说明内存已经严重不足，调高内存或降低storageFraction，减少spill。
- 现象是代码里用了groupByKey做求和/计数 → 用它改成reduceByKey或直接用SQL聚合，内存立省。
很多人在最后一步卡住，往往不是因为知识不够，而是参数与代码夹杂着同时调整，出了问题难定位。我现在的习惯是：一次改动只围绕一个假设展开，改完跑一小批数据验证，记录前后差异。反复几次后，哪个变量起了决定性作用，心里非常清楚。
6. 写在最后：调优的边界和我的体会
Shuffle阶段的OOM，和数据量、资源规格、代码质量、引擎版本全都相关，没有一组“万能参数”能覆盖所有情况。我必须承认，最难的步骤其实是识别出瓶颈到底在哪儿，而不是执行调整动作本身。先看UI、再抓GC、再定位热点key，这套流程走下来，大部分Shuffle OOM都能被拆解成普通问题。
我个人的经验是：在调优之前，先把作业本身的Shuffle数据总量、task分布、缓存用量打底理解透。数据总量都不知道，参数改来改去就是盲人摸象。另外，稳定调优之后，一定要把参数固化到作业配置或团队规范里，否则下一次作业重提，可能又回到老路。给作业留一个必要的运行基线，把GC时间、耗时、资源使用率记录清楚，对以后每次变更做对比都很有价值。
最后分享一个我反复踩过的坑：遇到OOM先冷静，别急着在群里喊加内存。加内存只解决了“资源不够”的表象，却经常掩盖了“数据不平衡”的实质。与其给某个task准备足够吞下800MB数据的堆，不如改掉那个让数据全挤到一个task里的逻辑。这一层想通了，Shuffle相关的问题基本就成功了大半。
