一、Shuffle流程
1、Shuffle定义
Spark之所以出现Shuffle，主要是因为具有某种共同特征的一类数据需要最终汇聚到一个计算节点上进行计算。这些数据分布在各个存储节点上，并且由不用的计算单元进行处理。这种数据打乱然后汇聚到不同节点的过程就是Shuffle。
2、Shuffle流程
Spark是以Shuffle为边界，将一个Job划分为不同的Stage，这些Stage构成了一个大粒度的DAG。Spark的Shuffle主要分为Shuffle Write和Shuffle Read两个阶段。
执行Shuffle的主体是Stage中的并发任务，这些任务分为ShuffleMapTask和ResultTask两大类。ShuffleMapTask要进行Shuffle，ResultTask负责返回计算结果，一个Job中只有最后一个Stage采用ResultTask，其它均为ShuffleMapTask。
如果按照map端和reduce端来分析的话，ShuffleMapTask可以即是map端任务，又是reduce端任务。因为Spark中的Shuffle可以是串行的，ResultTask则只能充当reduce端任务的角色。
Spark Shuffle流程简单抽象为如下几个步骤：
Shuffle Write
如果需要的话，先在map端做数据预聚合
写入本地输出文件
Shuffle Read
fetch数据块
reduce端做数据预聚合
如果需要的话，进行数据排序
Shuffle Write阶段：发生于ShuffleMapTask对该Stage的最后一个RDD完成了map端的计算之后，首先会判断是否需要对计算结果进行聚合，然后将最终结果按照不同的reduce端进行区分，写入前节点的本地磁盘。
Shuffle Read阶段：开始于reduce端的任务读取ShuffledRDD之后，首先通过远程或者本地数据拉取获得Write阶段各个节点中属于当前任务的数据，根据数据的Key进行聚合，然后判断是否需要排序，最后生成新的RDD。
二、Shuffle技术演进
在Spark Shuffle的具体实现上，主要经历了：hash-based shuffle、sort-based shuffle、Tungsten-sort shuffle 三个大的阶段。
1、hash-based shuffle V1
在Spark 0.8 及之前的版本采用：hash-based shuffle机制。
1）Shuffle流程介绍
（1）Shuffle Write
在Shuffle Write过程会按照hash的方式重组partition的数据，不进行排序。每个map端的任务为每个reduce端的任务都生成一个文件，通过会产生大量的文件（假如map端task数量为m，reduce端task数量为n，则对应 m * n个中间文件），其中伴随着大量的随机磁盘IO操作与大量的内存开销。
（2）Shuffle Read
Reduce端任务首先将Shuffle write生成的文件fetch到本地节点，如果Shuffle Read阶段有combiner操作，则它会把拉到的数据保存在一个Spark封装的哈希表（AppendOnlyMap）中进行合并。
2）源码结构
在代码结构上：
org.apache.spark.storage.ShuffleBlockManager 负责 Shuffle Write
org.apache.spark.BlockStoreShuffleFetcher 负责 Shuffle Read
org.apache.spark.Aggregator 负责 combine，依赖于 AppendOnlyMap
3）优缺点
该版本的Spark Shuffle机制存在如下两个严重问题：
- 生成大量的文件，占用文件描述符，同时引入DiskObjectWriter带来的Writer Handler的缓存也非常消耗内存；
- 如果在Reduce Task时需要合并操作的话，会把数据放在一个HashMap中进行合并，如果数据量较大，很容易引发OOM。
2、Hash Shuffle V2
在Spark 0.8.1 针对原来的hash-based shuffle机制，引入了 File Consolidation 机制。
一个 Executor 上所有的 Map Task 生成的分区文件只有一份，即将所有的 Map Task 相同的分区文件合并，这样每个 Executor 上最多只生成 N 个分区文件。
这样就减少了文件数，但是假如下游 Stage 的分区数 N 很大，还是会在每个 Executor 上生成 N 个文件，同样，如果一个 Executor 上有 K 个 Core，还是会开 K*N 个 Writer Handler，所以这里仍然容易导致OOM。
是否采用File Consolidation机制，需要配置 spark.shuffle.consolidateFiles 参数。
3、Hash Shuffle V3
在Spark 0.9 引入了ExternalAppendOnlyMap。
在combine的时候，可以将数据spill到磁盘，然后通过堆排序merge。
4、Sort Shuffle V1
为了更好的解决上面的问题，Spark 参考了 MapReduce 中的 Shuffle 的处理方式，
在Spark 1.1 引入了 sort-based shuffle ，但是默认仍为 hash-based shuffle。在Spark1.2 将默认的Shuffle方式修改为sort-based shuffle。
每个 Task 不会为后续的每个 Task 创建单独的文件，而是将所有对结果写入同一个文件。该文件中的记录首先是按照 Partition Id 排序，每个 Partition 内部再按照 Key 进行排序，Map Task 运行期间会顺序写每个 Partition 的数据，同时生成一个索引文件记录每个 Partition 的大小和偏移量。
在Reduce阶段，Reduce Task拉取数据做Combine时不再是采用HashMap，而是采用ExternalAppendOnlyMap，该数据结构在做Combine时，如果内存不足，会刷写磁盘，很大程度上保证了系统的鲁棒性，避免了大数据情况下的OOM。
总体来看，Sort Shuffle解决了 Hash Shuffle 的所有弊端，但是因为需要其Shuffle过程需要对记录进行排序，所以在性能上有所损失。
代码结构
在代码结构上：
从以前的 ShuffleBlockManager 中分离出 ShuffleManager 专门管理 Shuffle Write 和 Shuffle Read。两种Shuffle方式分别对应：
org.apache.spark.shuffle.hash.HashShuffleManager
org.apache.spark.shuffle.sort.SortShuffleManager
org.apache.spark.util.collection.ExternalSorter 实现排序功能。可通过对spark.shuffle.spill 参数配置，决定是否可以在排序时将临时数据Spill到磁盘。
5、Tungsten-Sort Based Shuffle
从 Spark 1.5.0 开始，Spark 开始了钨丝计划（Tungsten），目的是优化内存和CPU的使用，进一步提升spark的性能。由于使用了堆外内存，而它基于 JDK Sun Unsafe API，故 Tungsten-Sort Based Shuffle 也被称为 Unsafe Shuffle。
它的做法是将数据记录用二进制的方式存储，直接在序列化的二进制数据上 Sort 而不是在 Java 对象上，这样一方面可以减少内存的使用和 GC 的开销，另一方面避免 Shuffle 过程中频繁的序列化以及反序列化。在排序过程中，它提供 cache-efficient sorter，使用一个 8 bytes 的指针，把排序转化成了一个指针数组的排序，极大的优化了排序性能。
但是使用 Tungsten-Sort Based Shuffle 有几个限制，Shuffle 阶段不能有 aggregate 操作，分区数不能超过一定大小（2^24-1，这是可编码的最大 Parition Id），所以像 reduceByKey 这类有 aggregate 操作的算子是不能使用 Tungsten-Sort Based Shuffle，它会退化采用 Sort Shuffle。
6、Sort Shuffle V2
从 Spark-1.6.0 开始，把 Sort Shuffle 和 Tungsten-Sort Based Shuffle 全部统一到 Sort Shuffle 中，如果检测到满足 Tungsten-Sort Based Shuffle 条件会自动采用 Tungsten-Sort Based Shuffle，否则采用 Sort Shuffle。从Spark-2.0.0开始，Spark 把 Hash Shuffle 移除，可以说目前 Spark-2.0 中只有一种 Shuffle，即为 Sort Shuffle。
三、总结
1、Shuffle Read相关问题
关于Shuffle Read，主要了解以下问题：
1）在什么时候获取数据，Parent Stage的一个ShuffleMapTask执行完还是等全部ShuffleMapTask执行完？
当Parent Stage的所有ShuffleMapTasks结束后再fetch。
2）边获取边处理还是一次性获取完再处理？
因为Spark不要求Shuffle后的数据全局有序，因此没必要等到全部数据shuffle完成后再处理，所以是边fetch边处理。
3）获取来的数据存放在哪里？
刚获取来的 FileSegment 存放在 softBuffer 缓冲区，经过处理后的数据放在内存 + 磁盘上。
内存使用的是AppendOnlyMap ，类似 Java 的HashMap，内存＋磁盘使用的是ExternalAppendOnlyMap，如果内存空间不足时，ExternalAppendOnlyMap可以将 records 进行 sort 后 spill（溢出）到磁盘上，等到需要它们的时候再进行归并
4）怎么获得数据的存放位置？
通过请求 Driver 端的 MapOutputTrackerMaster 询问 ShuffleMapTask 输出的数据位置。
2、Shuffle触发机制
如下算子会触发Shuffle：
repartition类：repartition、coalesce
*ByKey类：groupByKey、reduceByKey、combineByKey、aggregateByKey等
join相关：cogroup、
四、Spark Shuffle版本变更
Spark 0.8 及以前 Hash Based Shuffle
Spark 0.8.1 为 Hash Based Shuffle引入File Consolidation机制
Spark 0.9 引入 ExternalAppendOnlyMap
Spark 1.1 引入 Sort Based Shuffle，但默认仍为 Hash Based Shuffle
Spark 1.2 默认的 Shuffle 方式改为 Sort Based Shuffle
Spark 1.4 引入 Tungsten-Sort Based Shuffle
Spark 1.6 Tungsten-Sort Based Shuffle 并入 Sort Based Shuffle
Spark 2.0 Hash Based Shuffle 退出历史舞台
