目录
（1）Shuffle概述
Shuffle 就是对数据进行重组，是把一组无规则的数据尽量转换成一组具有一定规则的数 据。由于分布式计算的特性和要求，在实现细节上更加繁琐和复杂。
在 MapReduce 框架，Shuffle 是连接 Map 和 Reduce 之间的桥梁，Map 阶段通过 shuffle 读取 数据并输出到对应的 Reduce。而 Reduce 阶段负责从 Map 端拉取数据并进行计算。在整个 shuffle 过程中，往往伴随着大量的磁盘和网络 I/O。所以 shuffle 性能的高低也直接决定 了整个程序的性能高低。下图为 Hadoop Shuffle 过程。
Spark 也有自己的 shuffle 实现过程。在 DAG 调度的过程中，Stage 阶段的划分是根据是否 有 shuffle 过程，也就是存在 ShuffleDependency 宽依赖的时候，需要进行 shuffle。 并且在划分 Stage 并构建 ShuffleDependency 的时候进 行 shuffle 注册，获取后续数据读取所需要的 ShuffleHandle, 最终每一个 job 提交后都会 生成一个 ResultStage 和若干个 ShuffleMapStage，其中 ResultStage 表示生成作业的最终 结果所在的 Stage。ResultStage 与 ShuffleMapStage 中的 task 分别对应着 ResultTask 与 ShuffleMapTask。一个作业，除了最终的 ResultStage 外，其他若干 ShuffleMapStage 中各 个 ShuffleMapTask 都需要将最终的数据根据相应的 Partitioner 对数据进行分组，然后持 久化分区的数据。
（2）Hash Shuffle机制
（2.1）Hash Shuffle概述
在 spark-1.6 版本之前，一直使用 HashShuffle，在 spark-1.6 版本之后使用 Sort Shuffle，因为 HashShuffle 存在的不足所以就替换了 HashShuffle。
我们知道，Spark 的运行主要分为 2 部分：一部分是驱动程序，其核心是 SparkContext。另 一部分是 Worker 节点上 Task,它是运行实际任务的。程序运行的时候，Driver 和 Executor 进程相互交互，Driver 会分配 Task 到 Executor，也就是 Driver 跟 Executor 会进行网络 传输。另外，当前 Task 要抓取其他上游的 Task 的数据结果，所以这个过程中就不断的产 生网络结果。其中下一个 Stage 向上一个 Stage 要数据这个过程，我们就称之为 Shuffle。
（2.2）没有优化之前的Hash Shuffle机制
HashShuffle没有优化之前的细节过程：
在 HashShuffle 没有优化之前，每一个 ShufflleMapTask 会为每一个 ReduceTask 创建一个bucket 缓存，并且会为每一个 bucket 创建一个文件。这个 bucket 存放的数据就是经过 Partitioner 操作(默认是 HashPartitioner)之后找到对应的 bucket 然后放进去，最后将数 据刷新 bucket 缓存的数据到磁盘上，即对应的 block file。
然 后 ShuffleMapTask 将 输 出 作 为 MapStatus 发 送 到 DAGScheduler 的 MapOutputTrackerMaster，每一个 MapStatus 包含了每一个 ResultTask 要拉取的数据的位 置和大小。
接下来 ResultTask 去利用 BlockStoreShuffleFetcher 向 MapOutputTrackerMaster 获取 MapStatus，看哪一份数据是属于自己的，然后底层通过 BlockManager 将数据拉取过来。
拉取过来的数据会组成一个内部的 ShuffleRDD，优先放入内存，内存不够用则放入磁盘， 然后 ResulTask 开始进行聚合，最后生成我们希望获取的那个 MapPartitionRDD。
自己的话总结：
每一个上游ShufflleMapTask 根据下游 ReduceTask数量，产生对应多个的bucket内存，这个bucket存放的数据是经过Partition操作（默认是Hashpartition）之后找到对应的 bucket 然后放进去，bucket内存大小默认是32k，最后将bucket缓存的数据溢写到磁盘，即为对应的block file。接下来Reduce Task底层通过 BlockManager 将数据拉取过来。拉取过来的数据会组成一个内部的 ShuffleRDD，优先放入内存，内存不够用则放入磁盘。
没有优化的Hash Shuffle的缺点
如上图所示：在这里有 1 个 worker， 2 个 executor， 每一个 executor 运行 2 个 ShuffleMapTask， 有三个 ReduceTask， 计算方式为：executor 数量每个 executor 的 ShuffleMapTask 数量ReduceTask 数量。所以总共就有 2 * 2 * 3=12 个 bucket 以及对应 12 个 block file(分区文件)。
- 如果数据量较大，将会生成 M*R 个小文件，比如 ShuffleMapTask 有 100 个，ResultTask 有 100 个，这就会产生 100 * 100=10000 个小文件
- bucket 缓存很重要，需要将 ShuffleMapTask 所有数据都写入 bucket，然后再刷到磁盘。 那么如果 Map 端数据过多，这就很容易造成内存溢出。尽管后面有优化，bucket 写入的数 据达到刷新到磁盘的阀值之后，就会将数据一点一点的刷新到磁盘，但是这样磁盘 I/O 就多 了。
（2.3）优化后的Hash Shuffle机制
HashShuffle优化之后的细节过程：
每一个 Executor 进程根据核数，决定 Task 的并发数量，比如 executor 核数是 2，那就可 以并发运行两个 task，如果是一个则只能运行一个 task。
假设 executor 核数是 1，ShuffleMapTask 数量是 M,那么它依然会根据 ResultTask 的数量 R， 创建 R 个 bucket 缓存，然后对 key 进行 hash，数据进入不同的 bucket 中，每一个 bucket 对应着一个 block file,用于刷新 bucket 缓存里的数据。
然后下一个 task 运行的时候，就不会再创建新的 bucket 和 block file，而是复用之前的 task 已经创建好的 bucket 和 block file。即所谓同一个 Executor 进程里所有 Task 都会把 相同的 key 放入相同的 bucket 缓冲区中。
这样的话， 生成文件的数量就是(本地 worker 的所有 executor 对应的 cores 的总数 *ResultTask 数量)如上图所示，即 2 * 3 = 6 个文件，每一个 Executor 的 shuffleMapTask 数量 100,ReduceTask 数量即为 100。
接下来举例比较一下，未优化的 HashShuffle 的文件数是 2 * 100 * 100 =20000，优化之后的 数量是 2*100 = 200 文件，相当于少了 100 倍。
自己的话总结：
每一个 Executor 进程根据核数，决定 Task 的并发数量，如果executor 核数是1，则只能运行一个task。ShuffleMapTask 会根据 ResultTask 的数量，产生对应多个的bucket内存，然后对 key 进行 hash分区，数据进入不同的 bucket 中，每一个 bucket 对应着一个 block file,用于刷新 bucket缓存里的数据。然后下一个 task 运行的时候，就不会再创建新的 bucket 和 block file，而是复用之前的 task 已经创建好的 bucket 和 block file。即所谓同一个 Executor 进程里所有 Task 都会把 相同的 key放入相同的 bucket 缓冲区中。
优化过的Hash Shuffle的缺点
如果 Reducer 端的并行任务或者是数据分片过多的话则 Core * Reducer Task 依旧 过大，也会产生很多小文件。
（3）Sort Shuffle机制
为了缓解 Shuffle 过程产生文件数过多和 Writer 缓存开销过大的问题，spark 引入了类似 于 hadoop Map-Reduce 的 shuffle 机制。该机制每一个 ShuffleMapTask 不会为后续的任务 创建单独的文件，而是会将所有的 Task 结果写入同一个文件，并且对应生成一个索引文件。 以前的数据是放在内存缓存中，等到缓存读取完数据后再刷到磁盘，现在为了减少内存的使 用，在内存不够用的时候，可以将输出溢写到磁盘。结束的时候，再将这些不同的文件联合 内存（缓存）的数据一起进行归并，从而减少内存的使用量。一方面文件数量显著减少，另 一方面减少 Writer 缓存所占用的内存大小，而且同时避免 GC 的风险和频率。
Sort Shuffle 有几种不同的策略：
- BypassMergeSortShuffleWriter（Bypass 机制）
- SortShuffleWriter（普通机制）
- UnsafeShuffleWriter
对于 BypassMergeSortShuffleWriter，使用这个模式的特点为：
- 主要用于处理不需要排序和聚合的 Shuffle 操作，所以数据是直接写入文件，数据量较大 的时候，网络 I/O 和内存负担较重。
- 主要适合处理 Reducer 任务数量比较少的情况。
- 将每一个分区写入一个单独的文件，最后将这些文件合并,减少文件数量。但是这种方式 需要并发打开多个文件，对内存消耗比较大。
因为 BypassMergeSortShuffleWriter 这种方式比 SortShuffleWriter 更快，所以如果在 Reducer 数 量 不 大 ， 又 不 需 要 在 map 端 聚 合 和 排 序 ， 而 且 Reducer 的 数 目 小 于 spark.shuffle.sort.bypassMergeThreshold 指定的阀值（默认 200）时，就是用的是这种 方式（即启用条件）。
对于 SortShuffleWriter(普通机制),使用这个模式的特点为：
- 比较适合数据量很大的场景或者集群规模很大。
- 引入了外部排序器，可以支持在 Map 端进行本地聚合或者不聚合。
- 如果外部排序器 enable 了 spill 功能，如果内存不够，可以先将输出溢写到本地磁盘， 最后将内存结果和本地磁盘的溢写文件进行合并。
另外，这个 Sort-Based Shuffle 跟 Executor 核数没有关系，即跟并发度没有关系，它是每 一个 ShuffleMapTask 都会产生一个 data 文件和 index 文件， 所谓合并也只是将该 ShuffleMapTask 的各个 partition 对应的分区文件合并到 data 文件而已。所以这个就需要 和 Hash-BasedShuffle 的 consolidation 机制区别开来。
（4）Spark Shuffle调优
-
调节map端缓冲区大小
- map端缓冲的默认配置是32KB
val conf = new SparkConf().set("spark.shuffle.file.buffer", "64")
-
调节reduce端拉取数据缓冲区大小
- 如果内存资源较为充足，适当增加拉取数据缓冲区的大小，可以减少拉取数据的次数，也就可以减少网络传输的次数，进而提升性能
- reduce端数据拉取缓冲区大小默认48MB
val conf = new SparkConf().set("spark.reducer.maxSizeInFlight", "96")
-
调节reduce端拉取数据重试次数
- reduce端拉取数据重试次数默认为3
val conf = new SparkConf().set("spark.shuffle.io.maxRetries", "6")
-
调节reduce端拉取数据等待间隔
- reduce端拉取数据等待间隔默认为5s
val conf = new SparkConf().set("spark.shuffle.io.retryWait", "60s")
-
调节SortShuffle排序操作阈值
- SortShuffleManager排序操作阈值默认为200
val conf = new SparkConf().set("spark.shuffle.sort.bypassMergeThreshold", "400")
以上内容仅供参考学习，如有侵权请联系我删除！
如果这篇文章对您有帮助，左下角的大拇指就是对博主最大的鼓励。
您的鼓励就是博主最大的动力！