Spark 的shuffle机制

一、Spark ShuffleManager 发展历程

• Spark 1.1.0 之前
  • 在 Spark 1.1.0 之前，Spark 使用 BlockStoreShuffleFetcher 来处理 Shuffle 操作。这个实现主要依赖于直接从 BlockManager 获取 Shuffle 数据，并通过网络进行交换。
    
• Spark 1.1.x（默认使用 HashShuffleManager）
  • HashShuffleManager 使用哈希算法将数据划分到不同的分区。在进行 Shuffle 操作时，Spark 会为每个键计算一个哈希值，然后根据该值将数据分配到相应的分区。
    
• Spark 1.2.x 及以后版本（默认使用 SortShuffleManager）
  • 
• Spark 2.0.x 及以后版本（不在使用HashShuffleManager,默认使用 SortShuffleManager）
  • 

二、HashShuffleManager 原理

假设：每个 Executor 只有1个CPU core，也就是说，无论这个 Executor 上分配多少个 task
线程，同一时间都只能执行一个 task 线程。

• 未经优化的HashShuffleManager工作原理
  

1.Shuffle Write 阶段

• 将每个task处理的数据按照key进行hash算法，从而将相同key都写入同一个磁盘文件，而每一个磁盘文件都只属于下游stage的一个task，在数据写入磁盘之前，会现将数据写入内存缓冲区中，当内存缓冲区填满以后，才会溢写到磁盘文件中去。
• 下一个stage的task有多少个，当前stage的每个task就要创建多少分磁盘文件，比如当前stage有20个task,总共有4个Executor,每个Executor执行5个task，下一个stage总共有40个task，那么每个Executor上就要创建200个磁盘文件，所有Executor会创建800个磁盘文件，由此可见，未经过优化的shuffle write操作所产生的磁盘文件数据是惊人的。

2. Shuffle Read 阶段

• 将上一个stage的计算结果中所有相同的key，从各个节点上通过网络都拉取到自己所在的节点上，然后按照key进行聚合或连接等操作。
• 由于shuffle write阶段，map task给下游stage的每个reduce task都创建了一个磁盘文件，因此shuffle read阶段，每个reduce task只要从上游stage的所有map task所在节点上拉取属于自己的那个磁盘文件即可。
• shuffle read的拉取过程是一边拉取一边聚合的，每个shuffle read task都有一个自己的buffer缓冲，每次只能拉取与buffer缓冲相同大小的数据，然后在内存中进行聚合等操作，聚合完一批数据，再拉取下一批，以此类推，直接所有数据拉取完，并得到最终结果。

• 优化后的HashShuffleManager工作原理
  

为了优化 HashShuffleManager，可以启用参数 spark.shuffle.consolidateFiles，该参数的默认值为 false，启用后设置为 true，可以启动优化机制。

开启优化机制后的效果：

1.Shuffle Write 阶段

• 在 shuffle write 过程中，task就不是为下游stage的每个task创建一个磁盘文件了，此时会出现shuffleFileGroup的概念，每个shuffleFileGroup会对应一批磁盘文件，磁盘文件的数量与下游stage的task数量是相同的。
• 一个Executor上有多少个 CPU core，就可以并行执行多少个task。而第一批并行执行的每个task都会创建一个 shuffleFileGroup，并将数据写入对应的磁盘文件内。
• 当执行下一批task时，下一批task就会复用之前已有的 shuffleFileGroup，包括其中的磁盘文件。
• consolidate 机制允许不同的task复用同一批磁盘文件，这样就可以有效将多个task的磁盘文件进行一定程度上的合并，从而大幅度减少磁盘文件的数量，进而提升 shuffle write 的性能。
• 比如当前stage有20个task,总共有4个Executor,每个Executor执行5个task，下一个stage总共有40个task，那么每个Executor上就要创建40个磁盘文件，所有Executor会创建160个磁盘文件，由此可见，优化后shuffle write操作所产生的磁盘文件较优化前明显减少。

2. Shuffle Read 阶段

• 由于shuffle write阶段，每个Executor仅为下游每个reduce task创建一个磁盘文件，在shuffle read阶段，每个reduce task只要从上游stage的所有map task所在节点上拉取属于自己的那个磁盘文件即可。

三、SortShuffleManager 原理

SortShuffleManager 的运行机制主要分成两种，一种是普通运行机制，另一种是bypass运行机制。 当 shuffle
read task 的数量小于等于 spark.shuffle.sort.bypassMergeThreshold参数(默认为
200)的值且不是聚合类的shuffle算子时，就会启用 bypass 机制。

• 普通运行机制的SortShuffleManager工作原理
  

• 在该模式下，数据会先写入一个内存数据结构中，此时根据不同的 shuffle 算子， 可能选用不同的数据结构。如果是 reduceByKey 这种聚合类的 shuffle 算子，那么会 选用 Map 数据结构，一边通过 Map 进行聚合，一边写入内存;如果是join 这种普通的 shuffle 算子，那么会选用 Array 数据结构，直接写入内存。接着，每写一条数据进入内存数据结构之后，就会判断一下，是否达到了某个临界阈值。如果达到临界阈值的话，那么就会尝试将内存数据结构中的数据溢写到磁盘，然后清空内存数据结构。在溢写到磁盘文件之前，会先根据 key 对内存数据结构中已有的数据进行排序。 排序过后，会分批将数据写入磁盘文件。默认的 batch 数量是10000 条，也就是说，排序好的数据，会以每批 1 万条数据的形式分批写入磁盘文件。写入磁盘文件是通 过 Java 的BufferedOutputStream 实现的。BufferedOutputStream 是 Java的缓冲输出流，首先会将数据缓冲在内存中，当内存缓冲填满之后再一次写入磁盘文件中，这样可以减少磁盘 IO 次数，提升性能。
• 一个 task 将所有数据写入内存数据结构的过程中，会发生多次磁盘溢写操作， 也就会产生多个临时文件。最后会将之前所有的临时磁盘文件都进行合并，这就是 merge 过程，此时会将之前所有临时磁盘文件中的数据读取出来，然后依次写入最 终的磁盘文件之中。此外，由于一个 task就只对应一个磁盘文件，也就意味着该 task 为下游 stage 的 task 准备的数据都在这一个文件中，因此还会单独写一份索引文件，其中标识了下游各个 task 的数据在文件中的 start offset 与 end offset。
• SortShuffleManager 由于有一个磁盘文件 merge 的过程，因此大大减少了文件数量。比如第一个 stage 有 20 个 task，总共有 4 个 Executor，每个 Executor 执行 5 个 task，而第二个 stage 有 40 个 task。由于每个 task 最终只有一个磁盘文件，因此 此时每个 Executor 上只有 5 个磁盘文件，所有 Executor 只有
  20 个磁盘文件。

• bypass运行机制的SortShuffleManager工作原理
  

• 每个 task 会为每个下游 task 都创建一个临时磁盘文件，并将数据按 key进行 hash 然后根据 key 的 hash 值，将 key 写入对应的磁盘文件之中。当然，写入磁盘文件时也是先写入内存缓冲，缓冲写满之后再溢写到磁盘文件的。最后，同样会将所有临时磁盘文件都合并成一个磁盘文件，并创建一个单独的索引文件。
• 该过程的磁盘写机制其实跟未经优化的 HashShuffleManager 是一模一样的，因为都要创建数量惊人的磁盘文件，只是在最后会做一个磁盘文件的合并而已。因此 少量的最终磁盘文件，也让该机制相对未经优化的 HashShuffleManager 来说，shuffle read 的性能会更好。
• 而该机制与普通 SortShuffleManager 运行机制的不同在于:第一，磁盘写机制 不同;第二，不会进行排序。也就是说，启用该机制的最大好处在于，shuffle write
  过程中，不需要进行数据的排序操作，也就节省掉了这部分的性能开销。