3. Spark Shuffle 原理：两种实现与完整流程

Shuffle 这个词，说白了就是“重新洗牌”。在 Spark 里，它发生在宽依赖（Wide Dependency）阶段——当某个 RDD 的分区数据需要被重新分发到下游 RDD 的不同分区时，Shuffle 就登场了。

我刚开始接触 Spark 时，总觉得 Shuffle 就是个黑盒。直到有一次线上任务频繁报 OOM，我才被迫把它从头到尾啃了一遍。嗯，今天我们就来拆开这个盒子。

3.1 为什么需要两种 Shuffle 实现？

Spark 早期只有一种实现——Hash Shuffle。后来社区发现它在处理大规模数据时性能堪忧，于是又引入了 Sort Shuffle。现在这两者并存，但默认走的是 Sort Shuffle。

你可能会问：为什么不直接淘汰掉 Hash Shuffle？其实在某些场景下，它依然有优势。我们来看对比。

特性 Hash Shuffle Sort Shuffle
核心思路 每个 Map 任务为每个 Reduce 任务生成一个文件 每个 Map 任务只生成一个文件，内部排序
文件数量 M × R（M=Map 数，R=Reduce 数） M × 2（数据文件 + 索引文件）
磁盘 I/O 大量小文件，随机写 少量大文件，顺序写
内存开销 较低 较高（需要排序缓冲区）
适用场景 小规模数据、Reduce 数少 大规模数据、通用场景

核心结论：Sort Shuffle 用排序换来了更少的文件数和更稳定的 I/O 性能。这是 Spark 默认选择它的根本原因。

3.2 Hash Shuffle：简单但粗暴

Hash Shuffle 的逻辑非常直接：每个 Map 任务根据 key 的哈希值，将数据写入对应 Reduce 分区的文件中。

举个例子：假设有 3 个 Map 任务、4 个 Reduce 任务。每个 Map 任务会创建 4 个文件，总共就是 3 × 4 = 12 个文件。

我早期维护过一个 Spark 1.x 的集群，有一次跑 200 个 Map 任务、500 个 Reduce 任务，结果产生了 10 万个文件。文件系统直接扛不住了，NameNode 的 RPC 延迟飙升到几十秒。

避坑指南：我曾经在 Reduce 数超过 1000 时用过 Hash Shuffle，结果磁盘写入成了瓶颈。大量小文件导致磁盘寻道时间剧增，任务跑了一个多小时还没结束。后来切到 Sort Shuffle，同样的数据量只用了 20 分钟。

Hash Shuffle 的流程可以概括为三步：

  1. Map 端写入：每个 Map 任务为每个 Reduce 分区创建一个临时文件，数据直接追加写入。
  2. Reduce 端拉取：每个 Reduce 任务从所有 Map 任务拉取属于自己分区的数据。
  3. 合并处理：Reduce 任务将拉取到的数据合并后执行聚合操作。

它的优点只有一个：实现简单，没有排序开销。但缺点很明显——文件数量爆炸，磁盘 I/O 极差。

3.3 Sort Shuffle：稳定且高效

Sort Shuffle 的出现就是为了解决 Hash Shuffle 的文件数量问题。它的核心思想是：每个 Map 任务只生成一个数据文件和一个索引文件。

我习惯把 Sort Shuffle 的流程分成两个阶段来看：Shuffle Write 和 Shuffle Read。下面我们逐个拆解。

3.4 Shuffle Write 详细流程

Shuffle Write 发生在 Map 阶段。它的核心任务是把数据按分区整理好，写入磁盘。

具体步骤如下：

  1. 数据写入内存缓冲区：Map 任务将输出数据写入一个内存缓冲区（默认 32KB）。缓冲区满了就溢写到磁盘。
  2. 分区 + 排序：在溢写之前，数据会按分区 ID 和 key 进行排序。排序的目的是让同一个分区的数据连续存放。
  3. 溢写文件：排序后的数据写入磁盘，生成一个临时文件。
  4. 合并溢写文件：如果多次溢写，最终会将所有临时文件合并成一个数据文件，同时生成一个索引文件。

索引文件记录了每个分区在数据文件中的起始偏移量和长度。这样 Reduce 任务拉取数据时，可以直接定位到目标位置，不需要扫描整个文件。

个人经验：我建议把 spark.shuffle.file.buffer 参数调大到 64KB 或 128KB，特别是当你的数据量较大时。这样可以减少溢写次数，提升写入性能。但别调太大，否则内存压力会上升。

下面这张图展示了 Shuffle Write 的完整流程：

Shuffle Write 流程图 Map 任务 内存缓冲区 分区 + 排序 溢写文件 合并文件 数据文件 + 索引文件 说明 • 缓冲区默认 32KB • 排序按分区 ID + key • 多次溢写后合并

3.5 Shuffle Read 详细流程

Shuffle Read 发生在 Reduce 阶段。它的核心任务是从各个 Map 任务拉取属于自己分区的数据，然后进行聚合或计算。

流程如下：

  1. 请求索引文件：Reduce 任务向 Driver 或直接向 Map 任务的 BlockManager 请求索引文件，获取目标分区的数据位置。
  2. 拉取数据：根据索引信息，从各个 Map 任务的数据文件中拉取对应的数据块。
  3. 数据合并：拉取到的数据块可能来自不同的 Map 任务，Reduce 任务会按 key 进行合并。
  4. 聚合计算：合并后的数据执行用户定义的聚合操作（如 reduceByKey、groupByKey）。

关键点：Shuffle Read 是网络 I/O 最密集的阶段。我曾经遇到过因为网络带宽不足导致 Shuffle 拉取超时的问题。解决办法是调整 spark.reducer.maxSizeInFlight 参数，默认 48MB，可以适当调小来避免网络拥塞。

这里有个细节值得注意：Reduce 任务并不是一次性拉取所有数据，而是分批拉取。每批拉取的数据量由 spark.reducer.maxSizeInFlight 控制。这样做的好处是避免内存溢出。

3.6 两种实现的选型建议

说了这么多，到底该用哪种？我个人的经验是：

  • 默认用 Sort Shuffle：99% 的场景都适用，稳定可靠。
  • 考虑 Hash Shuffle 的场景：数据量很小（比如几百 MB），且 Reduce 数少于 50。这时候 Hash Shuffle 没有排序开销，反而更快。
  • 不要用 Hash Shuffle 的场景：Reduce 数超过 200，或者数据量超过 10GB。文件数量会让你崩溃。

一个小技巧：如果你不确定当前任务用的是哪种 Shuffle，可以在 Spark UI 的 Stages 页面查看。如果看到 "SortShuffleWriter"，那就是 Sort Shuffle；如果是 "HashShuffleWriter"，那就是 Hash Shuffle。

3.7 总结

Shuffle 是 Spark 性能调优的重中之重。理解 Hash Shuffle 和 Sort Shuffle 的区别，掌握 Shuffle Write 和 Shuffle Read 的完整流程，是优化 Spark 作业的第一步。

嗯，这一章的内容就到这里。记住一句话：Shuffle 的本质是数据重分布，而优化的核心是减少数据移动和文件数量。

← 上一章 📖 返回目录 下一章 →