本文共 2556 字，大约阅读时间需要 8 分钟。

SparkShuffle概念

– Shuffle Write：上一个stage的每个map task就必须保证将自己处理的当前分区的数据相同的key写入一个分区文件中，可能会写入多个不同的分区文件中。

– Shuffle Read：reduce task就会从上一个stage的所有task所在的机器上寻找属于己的那些分区文件，这样就可以保证每一个key所对应的value都会汇聚到同一个节点上去处理和聚合。 Spark中有两种Shuffle类型，HashShuffle和SortShuffle，Spark1.2之前是HashShuffle默认的分区器是HashPartitioner，Spark1.2引入SortShuffle默认的分区器是RangePartitioner。

HashShuffle 普通机制

执行流程

a) 每一个map task将不同结果写到不同的buffer中，每个buffer的大小为32K。buffer起到数据缓存的作用。

b) 每个buffer文件最后对应一个磁盘小文件。 c) reduce task来拉取对应的磁盘小文件。

总结

① map task的计算结果会根据分区器（默认是hashPartitioner）来决定写入到哪一个磁盘小文件中去。ReduceTask会去Map端拉取相应的磁盘小文件。

② 产生的磁盘小文件的个数： M（map task的个数）*R（reduce task的个数）

导致问题

产生的磁盘小文件过多，会导致以下问题：

a) 在Shuffle Write过程中会产生很多写磁盘小文件的对象。 b) 在Shuffle Read过程中会产生很多读取磁盘小文件的对象。 c) 在JVM堆内存中对象过多会造成频繁的gc,gc还无法解决运行所需要的内存 的话，就会OOM。 d) 在数据传输过程中会有频繁的网络通信，频繁的网络通信出现通信故障的可能性大大增加，一旦网络通信出现了故障会导致shuffle file cannot find 由于这个错误导致的task失败，TaskScheduler不负责重试，由DAGScheduler负责重试Stage。

问题来了：默认的这种shuffle行为，对性能有什么样的恶劣影响呢？

实际生产环境的条件：

100个节点（每个节点一个executor）：100个executor 每个executor：2个cpu core 总共1000个task：每个executor平均10个task

每个节点，10个task，每个节点会输出多少份map端文件？10 * 1000=1万个文件

总共有多少份map端输出文件？100 * 10000 = 100万。

HashShuffle合并机制

new SparkConf().set("spark.shuffle.consolidateFiles", "true")

总结与注意事项

*产生磁盘小文件的个数：C(core的个数)R（reduce的个数）

开启了map端输出文件的合并机制之后： 第一个stage，同时就运行cpu core个task，比如cpu core是2个，并行运行2个task；每个task都创建下一个stage的task数量个文件； 第一个stage，并行运行的2个task执行完以后；就会执行另外两个task；另外2个task不会再重新创建输出文件；而是复用之前的task创建的map端输出文件，将数据写入上一批task的输出文件中。 第二个stage，task在拉取数据的时候，就不会去拉取上一个stage每一个task为自己创建的那份输出文件了；而是拉取少量的输出文件，每个输出文件中，可能包含了多个task给自己的map端输出。

提醒一下（map端输出文件合并）：

只有并行执行的task会去创建新的输出文件；下一批并行执行的task，就会去复用之前已有的输出文件；但是有一个例外，比如2个task并行在执行，但是此时又启动要执行2个task；那么这个时候的话，就无法去复用刚才的2个task创建的输出文件了；而是还是只能去创建新的输出文件。

要实现输出文件的合并的效果，必须是一批task先执行，然后下一批task再执行，才能复用之前的输出文件；负责多批task同时起来执行，还是做不到复用的。

开启map端输出机器，变化

开启了map端输出文件合并机制之后，生产环境上的例子，会有什么样的变化？

实际生产环境的条件：

100个节点（每个节点一个executor）：100个executor 每个executor：2个cpu core 总共1000个task：每个executor平均10个task

每个节点，2个cpu core，有多少份输出文件呢？2 * 1000 = 2000个

总共100个节点，总共创建多少份输出文件呢？100 * 2000 = 20万个文件

相比较开启合并机制之前的情况，100万个

map端输出文件，在生产环境中，立减5倍！

调节效果？

合并map端输出文件，对咱们的spark的性能有哪些方面的影响呢？

1、map task写入磁盘文件的IO，减少：100万文件 -> 20万文件

2、第二个stage，原本要拉取第一个stage的task数量份文件，1000个task，第二个stage的每个task，都要拉取1000份文件，走网络传输；合并以后，100个节点，每个节点2个cpu core，第二个stage的每个task，主要拉取100 * 2 = 200个文件即可；网络传输的性能消耗是不是也大大减少

分享一下，实际在生产环境中，使用了spark.shuffle.consolidateFiles机制以后，实际的性能调优的效果：对于上述的这种生产环境的配置，性能的提升，还是相当的客观的。spark作业，5个小时 -> 2~3个小时。

大家不要小看这个map端输出文件合并机制。实际上，在数据量比较大，你自己本身做了前面的性能调优，executor上去->cpu core上去->并行度（task数量）上去，shuffle没调优，shuffle就很糟糕了；大量的map端输出文件的产生。对性能有比较恶劣的影响。

转载地址：http://bhblz.baihongyu.com/