4. Spark内存管理与Shuffle机制
Spark的性能问题，十有八九能追溯到内存和Shuffle。这一章我们不讲抽象概念，而是把统一内存管理的账算清楚，再把Shuffle Write/Read的流程拆开，最后用Hash Shuffle和Sort Shuffle的对比，帮你建立“看到Stage边界就知道瓶颈在哪”的判断力。
一、统一内存管理：一块堆内内存怎么分
Spark 1.6之后引入统一内存管理（Unified Memory Management），把执行内存（Execution Memory）和存储内存（Storage Memory）放在同一个池子里动态借用。以Spark 3.x默认配置为例，堆内内存的划分逻辑是：
# 关键参数（默认值）
spark.memory.fraction = 0.6 # 用于执行+存储的内存比例
spark.memory.storageFraction = 0.5 # 存储内存的初始占比
spark.memory.offHeap.enabled = false # 默认不使用堆外内存假设Executor堆内内存为10GB，那么可用于执行和存储的总内存约为 (10GB - 300MB) × 0.6 ≈ 5.82GB，其中存储内存初始约2.91GB，执行内存初始约2.91GB。注意这里的300MB是spark.memory.reservedMemory保留内存，用于避免OOM。
统一内存管理的核心规则有三条，面试中经常被追问：
- 执行内存可以抢占存储内存：当Shuffle、Join、Sort、Aggregation需要内存时，可以驱逐存储内存中缓存的数据块。被驱逐的Block如果使用了MEMORY_AND_DISK级别，会落盘；如果是MEMORY_ONLY，则直接丢弃，后续需要时重新计算。
- 存储内存不能抢占执行内存：存储内存只能使用执行内存的空闲部分，一旦执行内存需要，存储内存必须让路。这是为了避免缓存数据把Shuffle操作挤到磁盘，导致性能雪崩。
- 存储内存有“保护额度”：
storageFraction定义了存储内存的保底比例。在执行内存未占满时，存储内存可以借用超过该比例的空间；但当执行内存回收时，存储内存需要归还到保底额度以内。
实际调优时，如果任务频繁发生Full GC或Spill，优先检查spark.memory.fraction是否被调得过小，以及是否存在大量MEMORY_ONLY缓存导致执行内存被反复驱逐。一个常见的经验值是：Shuffle密集的作业把spark.memory.fraction调到0.7~0.8，同时把storageFraction降到0.3左右，给执行内存留出更多空间。
二、Shuffle Write/Read流程：数据从Map端到Reduce端经历了什么
Shuffle的本质是跨Executor的数据重分区。以groupByKey为例，Map端每个Task根据分区器（默认HashPartitioner）计算每条记录的partitionId = key.hashCode % numPartitions，然后按分区写入本地磁盘。Reduce端每个Task拉取属于自己的那部分数据。
Shuffle Write阶段的关键动作：
- Map Task处理完数据后，将结果按
(partitionId, key)写入内存缓冲区。缓冲区大小由spark.shuffle.file.buffer控制，默认32KB。 - 当缓冲区达到阈值（
spark.shuffle.spill.numElementsForceSpillThreshold或内存压力触发），触发Spill，将数据排序后写入磁盘临时文件。 - 所有Spill文件最终合并成一个
data文件和一个index文件。index记录了每个Reduce分区在data文件中的偏移量和长度。
Shuffle Read阶段的关键动作：
- Reduce Task通过BlockManager向Map端发起Fetch请求，默认使用
spark.reducer.maxSizeInFlight=48MB控制单次拉取量。 - 拉取到的数据先放入内存缓冲区，如果使用了Sort Shuffle，会进行归并排序；如果是Hash Shuffle，则直接聚合。
- 最终结果写入内存或磁盘，供后续算子使用。
这里有一个容易被忽略的细节：Shuffle Read的拉取是并发的，默认spark.reducer.maxReqsInFlight=Int.MaxValue，但在网络不稳定或Map端数量很多时，建议调小到32~64，避免大量并发请求压垮Map端。
三、Hash Shuffle与Sort Shuffle对比
Hash Shuffle是Spark早期的默认实现，每个Map Task为每个Reduce Partition生成一个独立文件。假设有M个Map Task、R个Reduce Task，则会产生M × R个文件。当M=1000、R=1000时，就是100万个文件，磁盘I/O和文件句柄都会成为瓶颈。
Sort Shuffle在Spark 1.2之后引入，并在1.6之后成为默认。它不再为每个Reduce Partition单独建文件，而是所有分区写入同一个data文件，通过index文件记录偏移量。文件数量从M × R降到2M（每个Map Task一个data+一个index）。
| 对比维度 | Hash Shuffle | Sort Shuffle |
|---|---|---|
| 文件数量 | M × R | 2M |
| 是否排序 | 不排序 | 按key排序（可配置bypass） |
| 内存占用 | 较高，需维护多个缓冲区 | 较低，统一缓冲区 |
| 适用场景 | Reduce Partition少（<200） | Reduce Partition多、数据量大 |
| 是否可绕过排序 | 不涉及 | 支持bypass机制 |
Sort Shuffle有一个重要的优化：bypass机制。当Reduce Task数量小于spark.shuffle.sort.bypassMergeThreshold=200时，Sort Shuffle会跳过排序，直接按分区写入不同文件，最后合并。这样既避免了排序开销，又不会产生M×R个文件。判断依据是：如果Reduce Partition数量不多，且不需要全局排序，bypass能显著降低CPU消耗。
实际调优时，如果发现Shuffle Write阶段耗时过长，先看spark.sql.shuffle.partitions（默认200）是否合理。对于小数据量作业，200个分区会导致大量小文件；对于大数据量作业，200个分区又会导致单个分区过大。一个可操作的估算方式是：目标分区大小控制在128MB~256MB。假设Shuffle数据总量为50GB，则分区数约为50GB / 200MB ≈ 256，可以设置spark.sql.shuffle.partitions=256。
四、一个可复用的Shuffle调优检查清单
下面这份清单可以直接用于日常排查，字段和判断依据都来自上面的分析：
| 检查项 | 观察指标 | 判断依据 | 调整动作 |
|---|---|---|---|
| Shuffle分区数 | Stage的Task数量 | 分区大小是否在128~256MB | 调整spark.sql.shuffle.partitions |
| Shuffle Spill | Spill (memory) / Spill (disk) | Spill disk远大于0 | 增大spark.memory.fraction或减少分区数据量 |
| GC时间 | GC Time占比 | 超过10% | 检查缓存级别，降低storageFraction |
| Fetch失败 | Fetch Failed次数 | 大于0 | 调小maxReqsInFlight，检查网络 |
| 文件数量 | Shuffle Write文件数 | 接近M×R | 确认是否误用了Hash Shuffle |
最后强调一点：统一内存管理和Shuffle机制不是孤立的。执行内存不足会触发Spill，Spill会放大Shuffle Write的磁盘压力；而Shuffle Read的缓冲区又依赖执行内存。面试中如果被问到“Spark OOM怎么排查”，正确的切入路径是：先看是Driver OOM还是Executor OOM，再看是执行内存不足还是存储内存不足，最后结合Shuffle Spill和GC日志定位具体阶段。这条链路，就是本章内容要帮你建立的核心判断力。