SparkShuffle原理讲解以及调优【小二讲堂】

一、Spark Shuffle

在DAG调度的过程中&＃xff0c;Stage阶段的划分是根据是否有shuffle过程&＃xff0c;也就是存在ShuffleDependency宽依赖的时候&＃xff0c;需要进行shuffle,这时候会将作业job划分成多个Stage&＃xff1b;并且在划分Stage的时候&＃xff0c;构建ShuffleDependency的时候进行shuffle注册&＃xff0c;获取后续数据读取所需要的ShuffleHandle,最终每一个job提交后都会生成一个ResultStage和若干个ShuffleMapStage&＃xff0c;其中ResultStage表示生成作业的最终结果所在的Stage. ResultStage与ShuffleMapStage中的task分别对应着ResultTask与ShuffleMapTask。一个作业&＃xff0c;除了最终的ResultStage外&＃xff0c;其他若干ShuffleMapStage中各个ShuffleMapTask都需要将最终的数据根据相应的Partitioner对数据进行分组&＃xff0c;然后持久化分区的数据。

• shuffle
  就是在不同stage上的相同key的数据文件都汇聚成一个文件&＃xff0c;当然不同节点上都会形成&＃xff0c;reduce task拉取数据的时候&＃xff0c;将不同节点不同分区中相同key的数据拉取到一起&＃xff0c;形成一个文件&＃xff0c;这个文件中所有的key是相同的&＃xff0c;是所有节点上相同key的集合。
• 聚合类型
  – shuffle write:上一个stage的map task就必须保障自己处理的当前分区的数据相同的key写入到同一个分区文件中
  – shuffle read:reduce task就会从上一个stage的所有task所在的节点上寻找属于自己的分区文件。这样就可以保证每一个key&＃xff0c;所对应的value,都会汇聚到同一个节点上去处理。
  

上图是由task任务执行的stage中每个partition中的数据最后溢写成不同的小文件
在HashShuffle没有优化之前&＃xff0c;每一个ShufflleMapTask会为每一个ReduceTask创建一个bucket缓存&＃xff0c;并且会为每一个bucket创建一个文件。这个bucket存放的数据就是经过Partitioner操作(默认是HashPartitioner)之后找到对应的bucket然后放进去&＃xff0c;最后将数据刷新bucket缓存的数据到磁盘上&＃xff0c;即对应的block file。
然后ShuffleMapTask将输出作为MapStatus发送到DAGScheduler的MapOutputTrackerMaster&＃xff0c;每一个MapStatus包含了每一个ResultTask要拉取的数据的位置和大小

ResultTask然后去利用BlockStoreShuffleFetcher向MapOutputTrackerMaster获取MapStatus&＃xff0c;看哪一份数据是属于自己的&＃xff0c;然后底层通过BlockManager将数据拉取过来

拉取过来的数据会组成一个内部的ShuffleRDD&＃xff0c;优先放入内存&＃xff0c;内存不够用则放入磁盘&＃xff0c;然后ResulTask开始进行聚合&＃xff0c;最后生成我们希望获取的那个MapPartitionRDD
缺点&＃xff1a;
如上图所示&＃xff1a;在这里有1个worker&＃xff0c;2个executor&＃xff0c;每一个executor运行2个ShuffleMapTask&＃xff0c;有三个ReduceTask&＃xff0c;所以总共就有4 * 3&＃61;12个bucket和12个block file。
#如果数据量较大&＃xff0c;将会生成MR个小文件&＃xff0c;比如ShuffleMapTask有100个&＃xff0c;ResultTask有100个&＃xff0c;这就会产生100100&＃61;10000个小文件
#bucket缓存很重要&＃xff0c;需要将ShuffleMapTask所有数据都写入bucket&＃xff0c;才会刷到磁盘&＃xff0c;那么如果Map端数据过多&＃xff0c;这就很容易造成内存溢出&＃xff0c;尽管后面有优化&＃xff0c;bucket写入的数据达到刷新到磁盘的阀值之后&＃xff0c;就会将数据一点一点的刷新到磁盘&＃xff0c;但是这样磁盘I/O就多了

在task任务进行向内存中溢写的时候&＃xff0c;首先在executor中会根据reduce的个数分配三个缓冲区&＃xff0c;task处理的数据&＃xff0c;进入executor时会写入buffer缓冲区中&＃xff0c;这里会通过key.hashCode%numBuffer&＃xff0c;用key的哈希值对Buffer缓冲区的个数取模进行分配数据取那个缓冲区&＃xff0c;这个Buffer缓冲区的大小默认是32k,当buffer缓冲区中的数据写满32k时&＃xff0c;会将数据进行溢写到磁盘上&＃xff0c;溢写成一个磁盘文件&＃xff0c;这样当一个task执行完毕时&＃xff0c;就会形成很多个task磁盘文件&＃xff0c;当task很多时&＃xff0c;这样形成的小文件就是特别多的&＃xff0c;这样就会造成很多问题&＃xff1a;产生的文件个数&＃xff1a;MR(map个数reduce个数)

1.造成了IO量的剧增&＃xff0c;当reduce进行拉取数据时&＃xff0c;效率也就低下了。
2.初次之外还在进行溢写磁盘文件时&＃xff0c;肯定会创建大量对象的&＃xff0c;这里GC压力过大&＃xff0c;导致OOM。
3.进行拉取数据时&＃xff0c;首先可定会进行大量的连接&＃xff0c;大连接数过多时&＃xff0c;肯定会造成连接的中断问题过多。这样会造成taskScheduler进行重试&＃xff08;默认重试3次&＃xff09;&＃xff0c;三次不成功&＃xff0c;DAGScheduler进行重新计算分配Task任务&＃xff0c;重新跑task任务&＃xff0c;这样的效率是极为低下的。

二、Spark 1.6.x

对上面问题进行的优化
1.HashShuffle

由上图可以看出运行原理基本和上面相同&＃xff0c;就是在进行想缓冲去中写文件时进行了优化&＃xff0c;由一个executor中的两个task分别共用一个buffer,这样两个task任务会将取模后相同的buffer缓冲区的数据写到一块 &＃xff0c;最后溢写成对应的一个磁盘小文件。减少了至少一半的磁盘小文件。虽然减少了磁盘小文件但是&＃xff0c;在面对大数据的巨量数据集下&＃xff0c;这种优化还是有着很大的问题的&＃xff0c;和上面一样的问题。
磁盘文件个数&＃xff1a;ER&＃xff08;一个Executor产生的小文件个数Reduce的个数&＃xff09;。

三、Sort Shuffle&＃xff08;Spark 2.3.x&＃xff0c;Spark 1.6.x&＃xff09;

1.普通运行机制

在task进行运行的时候,有一个估算机制&＃xff0c;会加你估算执行task任务之前所需要的缓冲区大小。
在申请内存的时候&＃xff0c;估算机制是&＃xff1a;2*估算值-当前。executor会专门开启一块内存用于后续内存的申请。
首先task会将数据一块内存中&＃xff0c;然后达到一定的阈值之后进行溢写文件&＃xff0c;溢写文件的时候&＃xff0c;将数据向事先估算好的内存缓冲中&＃xff0c;但是在这之前会将数据的key的hashCode对缓冲区的个数进行取模&＃xff0c;这样根据取模的结果进行分配不同的数据到不同的缓冲区中&＃xff0c;写到之前首先会进行排序&＃xff0c;这里的排序和MapReduce中的分区排序一样。将数据加载到不同的内存缓冲区中。并且到内存缓冲区的值达到一定阈值后会溢写成响应的磁盘文件&＃xff0c;最后当map task 执行完毕的时候&＃xff0c;会将所有缓冲区中的文件进行溢写到对应的磁盘文件上。当mpa task执行完毕之后会通过归并排序将所有的数据文件合并成一个大文件&＃xff0c;这里比如有0&＃xff0c;1&＃xff0c;2好对应的的文件&＃xff0c;则会将不同的分区文件拉取到一起并且0号对应的文件会在大文件的前面&＃xff0c;1号在后面&＃xff0c;并且归并拉取的同时会创建一个索引文件&＃xff0c;来记录数据的位置。
最后由reduce task会将0号对应的磁盘文件拉取过来&＃xff0c;包括不同节点上的磁盘文件。

• 总结&＃xff1a;
  在执行过程中map task产生的磁盘小文件是2*M个&＃xff0c;即一个map task会产生两个磁盘小文件&＃xff0c;即一个是数据文件、另一个是数据文件对应的索引文件。
  这样大大减少了磁盘文件&＃xff0c;磁盘IO访问量&＃xff0c;以及数据拉取时的网络IO访问量。

2…bypass优化机制

和sortshuffle机制的运行原理相同&＃xff0c;只不过少了排序的过程&＃xff0c;这里是针对不需要进行排序的需求进行的运行机制。比如repartition&＃xff0c;它只是进行重新分区而不需要文件排序。
这里的运行效率就大大提高了

• 使用bypass的条件&＃xff1a;
  1.reduce端不能有聚合类型的操作。
  2.reduce task的个数必须要小于spark.shuffle.sort.bypassMergeThreshold 参数的值默认是200个,如果reducetask个数过多时&＃xff0c;可以设置这个参数的值。

四、Spark Shuffle总结

Shuffle 过程本质上都是将 Map 端获得的数据使用分区器进行划分&＃xff0c;并将数据发送给对应的 Reducer 的过程。
shuffle作为处理连接map端和reduce端的枢纽&＃xff0c;其shuffle的性能高低直接影响了整个程序的性能和吞吐量。map端的shuffle一般为shuffle的Write阶段&＃xff0c;reduce端的shuffle一般为shuffle的read阶段。Hadoop和spark的shuffle在实现上面存在很大的不同&＃xff0c;spark的shuffle分为两种实现&＃xff0c;分别为HashShuffle和SortShuffle。

HashShuffle又分为普通机制和合并机制&＃xff0c;普通机制因为其会产生MR个数的巨量磁盘小文件而产生大量性能低下的Io操作&＃xff0c;从而性能较低&＃xff0c;因为其巨量的磁盘小文件还可能导致OOM&＃xff0c;HashShuffle的合并机制通过重复利用buffer从而将磁盘小文件的数量降低到CoreR个&＃xff0c;但是当Reducer 端的并行任务或者是数据分片过多的时候&＃xff0c;依然会产生大量的磁盘小文件。

SortShuffle也分为普通机制和bypass机制&＃xff0c;普通机制在内存数据结构(默认为5M)完成排序&＃xff0c;会产生2M个磁盘小文件。而当shuffle map task数量小于spark.shuffle.sort.bypassMergeThreshold参数的值。或者算子不是聚合类的shuffle算子(比如reduceByKey)的时候会触发SortShuffle的bypass机制&＃xff0c;SortShuffle的bypass机制不会进行排序&＃xff0c;极大的提高了其性能。

在Spark 1.6以前&＃xff0c;默认的shuffle计算引擎是HashShuffleManager&＃xff0c;因为HashShuffleManager会产生大量的磁盘小文件而性能低下&＃xff0c;在Spark 1.6以后的版本中&＃xff0c;默认的ShuffleManager改成了SortShuffleManager。SortShuffleManager相较于HashShuffleManager来说&＃xff0c;有了一定的改进。主要就在于&＃xff0c;每个Task在进行shuffle操作时&＃xff0c;虽然也会产生较多的临时磁盘文件&＃xff0c;但是最后会将所有的临时文件合并(merge)成一个磁盘文件&＃xff0c;因此每个Task就只有一个磁盘文件。在下一个stage的shuffle read task拉取自己的数据时&＃xff0c;只要根据索引读取每个磁盘文件中的部分数据即可。

细节解释

• shuffle中的定时器&＃xff1a;
  定时器会检查内存数据结构的大小&＃xff0c;如果内存数据结构空间不够&＃xff0c;那么会申请额外的内存&＃xff0c;申请的大小满足如下公式&＃xff1a;
  applyMemory&＃61;nowMenory2-oldMemory
  申请的内存&＃61;当前的内存情况2-上一次的内嵌情况
  意思就是说内存数据结构的大小的动态变化&＃xff0c;如果存储的数据超出内存数据结构的大小&＃xff0c;将申请内存数据结构存储的数据*2-内存数据结构的设定值的内存大小空间。申请到了&＃xff0c;内存数据结构的大小变大&＃xff0c;内存不够&＃xff0c;申请不到&＃xff0c;则发生溢写。
• 排序
  在溢写到磁盘文件之前&＃xff0c;会先根据key对内存数据结构中已有的数据进行排序。
• 溢写
  排序过后&＃xff0c;会分批将数据写入磁盘文件。默认的batch数量是10000条&＃xff0c;也就是说&＃xff0c;排序好的数据&＃xff0c;会以每批1万条数据的形式分批写入磁盘文件。写入磁盘文件是通过Java的BufferedOutputStream实现的。BufferedOutputStream是Java的缓冲输出流&＃xff0c;首先会将数据缓冲在内存中&＃xff0c;当内存缓冲满溢之后再一次写入磁盘文件中&＃xff0c;这样可以减少磁盘IO次数&＃xff0c;提升性能。
• merge
  一个task将所有数据写入内存数据结构的过程中&＃xff0c;会发生多次磁盘溢写操作&＃xff0c;也就会产生多个临时文件。最后会将之前所有的临时磁盘文件都进行合并&＃xff0c;这就是merge过程&＃xff0c;此时会将之前所有临时磁盘文件中的数据读取出来&＃xff0c;然后依次写入最终的磁盘文件之中。此外&＃xff0c;由于一个task就只对应一个磁盘文件&＃xff0c;也就意味着该task为Reduce端的stage的task准备的数据都在这一个文件中&＃xff0c;因此还会单独写一份索引文件&＃xff0c;其中标识了下游各个task的数据在文件中的start offset与end offset。
  SortShuffleManager由于有一个磁盘文件merge的过程&＃xff0c;因此大大减少了文件数量。比如第一个stage有50个task&＃xff0c;总共有10个Executor&＃xff0c;每个Executor执行5个task&＃xff0c;而第二个stage有100个task。由于每个task最终只有一个磁盘文件&＃xff0c;因此此时每个Executor上只有5个磁盘文件&＃xff0c;所有Executor只有50个磁盘文件。
  1)block file&＃61; 2M
  一个map task会产生一个索引文件和一个数据大文件

2. m*r>2m(r>2)&＃xff1a;SortShuffle会使得磁盘小文件的个数再次的减少

五、Spark shuffle调优

• spark.shuffle.file.buffer
  默认值&＃xff1a;32k
  参数说明&＃xff1a;该参数用于设置shuffle write task的BufferedOutputStream的buffer缓冲大小。将数据写到磁盘文件之前&＃xff0c;会先写入buffer缓冲中&＃xff0c;待缓冲写满之后&＃xff0c;才会溢写到磁盘。
  调优建议&＃xff1a;如果作业可用的内存资源较为充足的话&＃xff0c;可以适当增加这个参数的大小&＃xff08;比如64k&＃xff09;&＃xff0c;从而减少shuffle write过程中溢写磁盘文件的次数&＃xff0c;也就可以减少磁盘IO次数&＃xff0c;进而提升性能。在实践中发现&＃xff0c;合理调节该参数&＃xff0c;性能会有1%~5%的提升。
• spark.reducer.maxSizeInFlight
  默认值&＃xff1a;48m
  参数说明&＃xff1a;该参数用于设置shuffle read task的buffer缓冲大小&＃xff0c;而这个buffer缓冲决定了每次能够拉取多少数据。
  调优建议&＃xff1a;如果作业可用的内存资源较为充足的话&＃xff0c;可以适当增加这个参数的大小&＃xff08;比如96m&＃xff09;&＃xff0c;从而减少拉取数据的次数&＃xff0c;也就可以减少网络传输的次数&＃xff0c;进而提升性能。在实践中发现&＃xff0c;合理调节该参数&＃xff0c;性能会有1%~5%的提升。
• spark.shuffle.io.maxRetries
  默认值&＃xff1a;3
  参数说明&＃xff1a;shuffle read task从shuffle write task所在节点拉取属于自己的数据时&＃xff0c;如果因为网络异常导致拉取失败&＃xff0c;是会自动进行重试的。该参数就代表了可以重试的最大次数。如果在指定次数之内拉取还是没有成功&＃xff0c;就可能会导致作业执行失败。
  调优建议&＃xff1a;对于那些包含了特别耗时的shuffle操作的作业&＃xff0c;建议增加重试最大次数&＃xff08;比如60次&＃xff09;&＃xff0c;以避免由于JVM的full gc或者网络不稳定等因素导致的数据拉取失败。在实践中发现&＃xff0c;对于针对超大数据量&＃xff08;数十亿~上百亿&＃xff09;的shuffle过程&＃xff0c;调节该参数可以大幅度提升稳定性。
• spark.shuffle.io.retryWait
  默认值&＃xff1a;5s
  参数说明&＃xff1a;具体解释同上&＃xff0c;该参数代表了每次重试拉取数据的等待间隔&＃xff0c;默认是5s。
  调优建议&＃xff1a;建议加大间隔时长&＃xff08;比如60s&＃xff09;&＃xff0c;以增加shuffle操作的稳定性。
• spark.shuffle.memoryFraction
  默认值&＃xff1a;0.2
  参数说明&＃xff1a;该参数代表了Executor内存中&＃xff0c;分配给shuffle read task进行聚合操作的内存比例&＃xff0c;默认是20%。
  调优建议&＃xff1a;在资源参数调优中讲解过这个参数。如果内存充足&＃xff0c;而且很少使用持久化操作&＃xff0c;建议调高这个比例&＃xff0c;给shuffle read的聚合操作更多内存&＃xff0c;以避免由于内存不足导致聚合过程中频繁读写磁盘。在实践中发现&＃xff0c;合理调节该参数可以将性能提升10%左右。
• spark.shuffle.manager
  默认值&＃xff1a;sort
  参数说明&＃xff1a;该参数用于设置ShuffleManager的类型。Spark 1.5以后&＃xff0c;有三个可选项&＃xff1a;hash、sort和tungsten-sort。HashShuffleManager是Spark 1.2以前的默认选项&＃xff0c;但是Spark 1.2以及之后的版本默认都是SortShuffleManager了。tungsten-sort与sort类似&＃xff0c;但是使用了tungsten计划中的堆外内存管理机制&＃xff0c;内存使用效率更高。
  调优建议&＃xff1a;由于SortShuffleManager默认会对数据进行排序&＃xff0c;因此如果你的业务逻辑中需要该排序机制的话&＃xff0c;则使用默认的SortShuffleManager就可以&＃xff1b;而如果你的业务逻辑不需要对数据进行排序&＃xff0c;那么建议参考后面的几个参数调优&＃xff0c;通过bypass机制或优化的HashShuffleManager来避免排序操作&＃xff0c;同时提供较好的磁盘读写性能。这里要注意的是&＃xff0c;tungsten-sort要慎用&＃xff0c;因为之前发现了一些相应的bug。
• spark.shuffle.sort.bypassMergeThreshold
  默认值&＃xff1a;200
  参数说明&＃xff1a;当ShuffleManager为SortShuffleManager时&＃xff0c;如果shuffle read task的数量小于这个阈值&＃xff08;默认是200&＃xff09;&＃xff0c;则shuffle write过程中不会进行排序操作&＃xff0c;而是直接按照未经优化的HashShuffleManager的方式去写数据&＃xff0c;但是最后会将每个task产生的所有临时磁盘文件都合并成一个文件&＃xff0c;并会创建单独的索引文件。
  调优建议&＃xff1a;当你使用SortShuffleManager时&＃xff0c;如果的确不需要排序操作&＃xff0c;那么建议将这个参数调大一些&＃xff0c;大于shuffle read task的数量。那么此时就会自动启用bypass机制&＃xff0c;map-side就不会进行排序了&＃xff0c;减少了排序的性能开销。但是这种方式下&＃xff0c;依然会产生大量的磁盘文件&＃xff0c;因此shuffle write性能有待提高。
• spark.shuffle.consolidateFiles
  默认值&＃xff1a;false
  参数说明&＃xff1a;如果使用HashShuffleManager&＃xff0c;该参数有效。如果设置为true&＃xff0c;那么就会开启consolidate机制&＃xff0c;会大幅度合并shuffle write的输出文件&＃xff0c;对于shuffle read task数量特别多的情况下&＃xff0c;这种方法可以极大地减少磁盘IO开销&＃xff0c;提升性能。
  调优建议&＃xff1a;如果的确不需要SortShuffleManager的排序机制&＃xff0c;那么除了使用bypass机制&＃xff0c;还可以尝试将spark.shffle.manager参数手动指定为hash&＃xff0c;使用HashShuffleManager&＃xff0c;同时开启consolidate机制。在实践中尝试过&＃xff0c;发现其性能比开启了bypass机制的SortShuffleManager要高出10%~30%。

MapReduce shuffle讲解:https://blog.csdn.net/Mirror_w/article/details/89421705
小二讲堂&＃xff1a;https://blog.csdn.net/Mirror_w
Spark讲堂&＃xff1a;https://blog.csdn.net/Mirror_w/article/details/89408567