Spark Shuffle 调优实战：从热点定位到内存优化的系统方案

Spark 大数据量场景下 Shuffle Spill 和 OOM 的根因是同一个链条：数据在 Shuffle 阶段无法全部驻留内存，被迫溢写磁盘，当溢写量超过磁盘 I/O 承载能力或 Executor 内存被彻底耗尽时，作业就会报出 OOM 异常。系统化调优需要从定位热点 Task 开始，沿分区数、内存、压缩三个层级逐级优化。

第一步：用 Spark UI 精准定位热点

调优的第一要素不是盲目调整参数，而是确认问题发生在哪个 Stage、哪个 Task。

打开 Spark Web UI 的 Stages 页面，重点关注两个指标：每个 Task 的 Shuffle Read 数据量和运行时间。如果绝大多数 Task 在数秒内完成，但个别 Task 的运行时间达到数分钟甚至数十分钟，且其 Shuffle Read 数据量远超其他 Task，就说明存在数据倾斜导致的热点。

进一步检查 Stage 详情页的 Shuffle Write 和 Shuffle Read 统计。如果 Shuffle Spill（Disk）的数值很大，说明 Executor 内存不足以容纳 Shuffle 数据，数据被大量溢写到磁盘。正常情况下 Shuffle Spill 应该为零或接近零；如果达到 Shuffle Read 量的 50% 以上，就意味着内存配置存在明显瓶颈。

根据美团技术团队的实践经验，数据倾斜的典型表现是：总共 1000 个 Task，997 个在 1 分钟内完成，剩余 2 到 3 个需要运行 1 到 2 小时。整个 Stage 的运行时间由最慢的那个 Task 决定。定位热点后，需要查看 Stage 与代码的对应关系：每个 Shuffle 类算子（如 reduceByKey、groupByKey、join、repartition）都会划分出 Stage 边界，热点所在 Stage 对应的代码中一定包含这类算子。

第二步：分区数调整——第一级优化

spark.sql.shuffle.partitions 控制 Shuffle Read 的并行度，默认值为 200。对于大数据量场景，这个默认值通常过小，导致每个 Task 需要处理过多数据。

调整策略的核心原则是：让每个 Task 处理的数据量落在 100MB 到 200MB 之间。如果总数据量为 50GB，将分区数设为 256 到 512 是合理的起点。如果数据量达到 TB 级别，分区数需要设为数千甚至上万。

一个关键细节是：spark.sql.shuffle.partitions 必须在作业启动时就设置生效，而非在代码中途动态修改。如果需要对已有 DataFrame 重新分区，必须在 Action 操作（如 count()、write()）前调用 repartition() 并触发执行，确保分区变更真正生效。

对于数据倾斜导致的热点，单纯增加分区数的效果有限——因为倾斜的 Key 无论如何都会被分配到同一个 Task。此时需要配合两阶段聚合（先给 Key 加随机前缀做局部聚合，再去前缀做全局聚合）或广播 Join（小表广播避免 Shuffle）来打散热点。

第三步：内存参数调优——第二级优化

Spark 的 Executor 内存分为多个区域，与 Shuffle 相关的关键参数包括以下几个。

spark.shuffle.memoryFraction（旧版本参数）控制 Executor 堆内存中分配给 Shuffle Read 聚合操作的比例，默认 0.2。在内存充足且不频繁使用 cache()/persist() 的场景下，建议提高到 0.3 到 0.4，给 Shuffle 聚合更多内存空间，减少溢写频率。美团的实践表明，合理调节此参数可将性能提升 10% 左右。

spark.executor.memoryOverhead 控制堆外内存，默认为 Executor 内存的 10%（最小 384MB）。Shuffle 过程中的序列化、Netty 网络缓冲都使用堆外内存。如果 OOM 异常栈指向 DirectBuffer 或 Netty，说明堆外内存不足，需要将此参数提高到 4096MB 甚至更高。

spark.shuffle.file.buffer 控制 Shuffle Write 的磁盘写入缓冲大小，默认 32KB。增大到 64KB 可以减少溢写次数，降低磁盘 I/O 频率，实践中通常带来 1% 到 5% 的性能提升。

spark.reducer.maxSizeInFlight 控制 Shuffle Read 端每次拉取的数据块大小，默认 48MB。增大到 96MB 可以减少网络拉取次数，同样带来 1% 到 5% 的性能提升。

spark.shuffle.io.maxRetries 和 spark.shuffle.io.retryWait 控制 Shuffle Read 的重试策略，默认分别为 3 次和 5 秒。对于超大数据量的 Shuffle（数十亿到上百亿条记录），建议将重试次数提高到 60 次、等待间隔提高到 60 秒，避免因 JVM Full GC 或网络抖动导致的偶发性失败。

第四步：压缩与数据格式优化——第三级优化

开启 Shuffle 数据压缩是减少内存和磁盘压力的最直接手段。spark.shuffle.spill.compress 默认为 true，确保该参数开启。spark.shuffle.compress 控制 Shuffle Write 阶段的输出压缩，默认也是 true。

压缩算法的选择需要在 CPU 开销和压缩比之间权衡。LZ4 压缩速度快、压缩比适中，是默认推荐。Snappy 压缩速度更快但压缩比略低。ZSTD 提供更高的压缩比但 CPU 开销更大。对于 Shuffle 瓶颈在磁盘 I/O 而非 CPU 的场景，优先使用 LZ4 或 Snappy。

在数据格式层面，将宽表中的冗余列在 Shuffle 前过滤掉（使用 select() 只保留必要列），可以显著减少 Shuffle 数据量。将数字类型从 String 改为 Integer/Long、将日期从 String 改为 DateType，都能降低单条记录的内存占用。对于 Join 操作中的小表（数据量在数百 MB 以内），使用 broadcast() 将小表广播到所有 Executor，完全避免大表的 Shuffle，这是消除 Join 导致的数据倾斜的最有效方式。

第五步：ShuffleManager 的底层机制理解

理解 ShuffleManager 的工作机制有助于做出更精准的调优决策。Spark 1.2 以后默认使用 SortShuffleManager，取代了产生大量中间文件的 HashShuffleManager。SortShuffleManager 的核心改进是：每个 Task 最终只生成一个磁盘文件和一个索引文件，下游 Task 根据索引读取自己需要的数据分片。

SortShuffleManager 有两种运行模式：普通模式和 bypass 模式。当 Shuffle Read Task 数量小于 spark.shuffle.sort.bypassMergeThreshold（默认 200）且不是聚合类操作时，启用 bypass 模式——跳过排序步骤直接写磁盘，再合并为单文件。对于不需要排序的 Join 操作，适当提高此阈值可以跳过排序开销。

tungsten-sort 是第三种可选模式，使用堆外内存管理机制，内存使用效率更高。如果 Spark 版本支持且作业内存压力较大，可以尝试 spark.shuffle.manager=tungsten-sort。

调优顺序与常见误区

推荐的调优顺序是：先定位热点（Spark UI）→ 调整分区数 → 优化内存参数 → 开启压缩与精简数据 → 深入 ShuffleManager 底层配置。每一级优化后观察效果再决定是否进入下一级。

常见的误区包括：盲目增大 Executor 内存但不调整 Shuffle 内存比例，导致大部分内存被堆对象占用，Shuffle 聚合仍然溢写；设置 spark.sql.shuffle.partitions 为极大值但未配合 repartition() 触发，参数实际未生效；在广播 Join 场景下将广播阈值设得过大，导致 Driver 内存溢出。

EasyMR 平台基于 Hadoop 自研增强，统一承载批计算、流计算、OLAP 与湖仓存储，其内置的 Spark 调优建议模块可以根据作业的 Shuffle 指标自动生成参数推荐，降低了人工调优的试错成本。对于大规模集群的 Shuffle 性能问题，平台层面的资源隔离和动态分配能力也提供了额外的保障。

Shuffle 调优的本质是在有限的集群资源下，让数据在节点间高效流转。核心思路是：让每个 Task 处理适量数据（分区数调整）、让 Shuffle 数据尽量留在内存中（内存参数优化）、减少需要传输的数据量（压缩与精简）。把握住这三个方向，大部分 Shuffle 性能问题都能得到有效缓解。