149、Spark核心编程进阶之Shuffle相关

shuffle操作原理

是spark中一些特殊的算子操作会触发的一种操作
shuffle操作&＃xff0c;会导致大量的数据在不同的机器和节点之间进行传输&＃xff0c;因此也是spark中最复杂、最消耗性能的一种操作

我们可以通过reduceByKey操作作为一个例子&＃xff0c;来理解shuffle操作
先看图

shuffle操作原理.png

reduceByKey算子会将上一个RDD中的每个key对应的所有value都聚合成一个value&＃xff0c;然后生成一个新的RDD
新的RDD的元素类型就是对的格式&＃xff0c;每个key对应一个聚合起来的value
这里最大的问题就在于&＃xff0c;对于上一个RDD来说&＃xff0c;并不是一个key对应的所有value都是在一个partition中的&＃xff0c;也更不太可能说key的所有value都在一台机器上
所以对于这种情况来说&＃xff0c;就必须在整个集群中&＃xff0c;将各个节点上&＃xff0c;同一个key对应的values&＃xff0c;统一传输到一个节点上来聚合处理
这个过程中就会发生大量的网络数据的传输

在进行一个key对应的values的聚合时
首先&＃xff0c;上一个stage的每个map task就必须保证将自己处理的当前分区中的数据&＃xff0c;相同的key写入一个分区文件中&＃xff0c;可能会写多个不同的分区文件
接着下一个stage的reduce task就必须从上一个stage所有task所在的机器上&＃xff0c;将各个task写入的多个分区文件中&＃xff0c;找到属于自己的那个分区文件
接着将属于自己的分区数据&＃xff0c;拉取过来&＃xff0c;这样就可以保证每个key对应的所有values都汇聚到一个节点上去处理和聚合
这个过程就称之为shuffle

shuffle是分为shuffle write和shuffle read两个部分的,是在两个不同的stage中进行的

shuffle操作过程中进行数据排序

默认情况下&＃xff0c;shuffle操作是不会对每个分区中的数据进行排序的

如果想要对每个分区中的数据进行排序&＃xff0c;那么可以使用以下三种方法&＃xff1a;

1. 使用mapPartitions算子处理每个partition&＃xff0c;对每个partition中的数据进行排序
2. 使用repartitionAndSortWithinPartitions&＃xff0c;对RDD进行重分区&＃xff0c;在重分区的过程中同时就进行分区内数据的排序
3. 使用sortByKey对数据进行全局排序

上述三种方法中&＃xff0c;相对来说&＃xff0c;mapPartitions的代价比较小&＃xff0c;因为不需要进行额外的shuffle操作
repartitionAndSortWithinPartitions和sortByKey可能会进行额外的shuffle操作的&＃xff0c;因此性能并不是很高

val rdd2 &＃61; rdd1.reduceByKey(_ &＃43; _)
rdd2.mapPartitions(tuples.sort)
rdd2.repartitionAndSortWithinPartitions()&＃xff0c;重分区&＃xff0c;重分区的过程中&＃xff0c;就进行分区内的key的排序&＃xff0c;重分区的原理和repartition一样
rdd2.sortByKey&＃xff0c;直接对rdd按照key进行全局性的排序

spark中会导致shuffle操作

spark中会导致shuffle操作的有以下几种算子

1. repartition类的操作&＃xff1a;比如repartition、repartitionAndSortWithinPartitions、coalesce等
2. byKey类的操作&＃xff1a;比如reduceByKey、groupByKey、sortByKey等
3. join类的操作&＃xff1a;比如join、cogroup等

重分区: 一般会shuffle&＃xff0c;因为需要在整个集群中&＃xff0c;对之前所有的分区的数据进行随机&＃xff0c;均匀的打乱&＃xff0c;然后把数据放入下游新的指定数量的分区内
byKey类的操作&＃xff1a;因为你要对一个key&＃xff0c;进行聚合操作&＃xff0c;那么肯定要保证集群中&＃xff0c;所有节点上的&＃xff0c;相同的key&＃xff0c;一定是到同一个节点上进行处理
join类的操作&＃xff1a;两个rdd进行join&＃xff0c;就必须将相同join key的数据&＃xff0c;shuffle到同一个节点上&＃xff0c;然后进行相同key的两个rdd数据的笛卡尔乘积

所以对于上述的操作
首先第一原则&＃xff0c;就是&＃xff0c;能不用shuffle操作&＃xff0c;就尽量不用shuffle操作&＃xff0c;尽量使用不shuffle的操作
第二原则&＃xff0c;就是&＃xff0c;如果使用了shuffle操作&＃xff0c;那么肯定要进行shuffle的调优&＃xff0c;甚至是解决碰到的数据倾斜的问题

shuffle操作对性能消耗的原理

shuffle操作是spark中唯一最最消耗性能的地方
因此也就成了最需要进行性能调优的地方&＃xff0c;最需要解决线上报错的地方&＃xff0c;也是唯一可能出现数据倾斜的地方
因为shuffle过程中&＃xff0c;会产生大量的磁盘IO、数据序列化和反序列化、网络IO

为了实施shuffle操作
spark中才有了stage的概念&＃xff0c;在发生shuffle操作的算子中&＃xff0c;进行stage的拆分
shuffle操作的前半部分&＃xff0c;是上一个stage来进行&＃xff0c;也称之为map task&＃xff0c;shuffle操作的后半部分&＃xff0c;是下一个stage来进行&＃xff0c;也称之为reduce task
其中map task负责数据的组织&＃xff0c;也就是将同一个key对应的value都写入同一个下游task对应的分区文件中
其中reduce task负责数据的聚合&＃xff0c;也就是将上一个stage的task所在节点上&＃xff0c;将属于自己的各个分区文件&＃xff0c;都拉取过来聚合
这种模型&＃xff0c;是参考和模拟了MapReduce的shuffle过程来的

map task会将数据先保存在内存中&＃xff0c;如果内存不够时&＃xff0c;就溢写到磁盘文件中去
reduce task会读取各个节点上属于自己的分区磁盘文件&＃xff0c;到自己节点的内存中&＃xff0c;并进行聚合

shuffle操作会消耗大量的内存&＃xff0c;因为无论是网络传输数据之前&＃xff0c;还是之后&＃xff0c;都会使用大量的内存中数据结构来实施聚合操作
比如reduceByKey和aggregateByKey操作&＃xff0c;会在map side使用内存中的数据结构进行预先聚合
其他的byKey类的操作&＃xff0c;都是在reduce side&＃xff0c;使用内存数据结构进行聚合
在聚合过程中&＃xff0c;如果内存不够&＃xff0c;只能溢写到磁盘文件中去&＃xff0c;此时就会发生大量的磁盘IO&＃xff0c;降低性能

此外&＃xff0c;shuffle过程中&＃xff0c;还会产生大量的中间文件&＃xff0c;也就是map side写入的大量分区文件
比如Spark 1.3版本&＃xff0c;这些中间文件会一致保留着&＃xff0c;直到RDD不再被使用&＃xff0c;而且被垃圾回收掉了&＃xff0c;才会去清理中间文件
这主要是为了&＃xff0c;如果要重新计算shuffle后的RDD&＃xff0c;那么map side不需要重新做一次磁盘写操作
但是这种情况下&＃xff0c;如果我们的应用程序中&＃xff0c;一直保持着对RDD的引用&＃xff0c;导致很长时间以后才会进行RDD垃圾回收操作
保存中间文件的目录&＃xff0c;由spark.local.dir属性指定

内存的消耗、磁盘IO、网络数据传输&＃xff08;IO&＃xff09;

shuffle操作所有相关参数详解以及性能调优

我们可以通过对一系列的参数进行调优&＃xff0c;来优化shuffle的性能
spark 1.5.2版本