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;
- 使用mapPartitions算子处理每个partition&＃xff0c;对每个partition中的数据进行排序
- 使用repartitionAndSortWithinPartitions&＃xff0c;对RDD进行重分区&＃xff0c;在重分区的过程中同时就进行分区内数据的排序
- 使用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操作的有以下几种算子
- repartition类的操作&＃xff1a;比如repartition、repartitionAndSortWithinPartitions、coalesce等
- byKey类的操作&＃xff1a;比如reduceByKey、groupByKey、sortByKey等
- 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版本
| 属性名称 | 默认值 | 属性说明 |
|---|
| spark.reducer.maxSizeInFlight | 48m | reduce task的buffer缓冲&＃xff0c;代表了每个reduce task每次能够拉取的map side数据最大大小&＃xff0c;如果内存充足&＃xff0c;可以考虑加大大小&＃xff0c;从而减少网络传输次数&＃xff0c;提升性能 |
| spark.shuffle.blockTransferService | netty | shuffle过程中&＃xff0c;传输数据的方式&＃xff0c;两种选项&＃xff0c;netty或nio&＃xff0c;spark 1.2开始&＃xff0c;默认就是netty&＃xff0c;比较简单而且性能较高&＃xff0c;spark 1.5开始nio就是过期的了&＃xff0c;而且spark 1.6中会去除掉 |
| spark.shuffle.compress | true | 是否对map side输出的文件进行压缩&＃xff0c;默认是启用压缩的&＃xff0c;压缩器是由spark.io.compression.codec属性指定的&＃xff0c;默认是snappy压缩器&＃xff0c;该压缩器强调的是压缩速度&＃xff0c;而不是压缩率 |
| spark.shuffle.consolidateFiles | false | 默认为false&＃xff0c;如果设置为true&＃xff0c;那么就会合并map side输出文件&＃xff0c;对于reduce task数量特别的情况下&＃xff0c;可以极大减少磁盘IO开销&＃xff0c;提升性能 |
| spark.shuffle.file.buffer | 32k | map side task的内存buffer大小&＃xff0c;写数据到磁盘文件之前&＃xff0c;会先保存在缓冲中&＃xff0c;如果内存充足&＃xff0c;可以适当加大大小&＃xff0c;从而减少map side磁盘IO次数&＃xff0c;提升性能 |
| spark.shuffle.io.maxRetries | 3 | 网络传输数据过程中&＃xff0c;如果出现了网络IO异常&＃xff0c;重试拉取数据的次数&＃xff0c;默认是3次&＃xff0c;对于耗时的shuffle操作&＃xff0c;建议加大次数&＃xff0c;以避免full gc或者网络不通常导致的数据拉取失败&＃xff0c;进而导致task lost&＃xff0c;增加shuffle操作的稳定性 |
| spark.shuffle.io.retryWait | 5s | 每次重试拉取数据的等待间隔&＃xff0c;默认是5s&＃xff0c;建议加大时长&＃xff0c;理由同上&＃xff0c;保证shuffle操作的稳定性 |
| spark.shuffle.io.numConnectionsPerPeer | 1 | 机器之间的可以重用的网络连接&＃xff0c;主要用于在大型集群中减小网络连接的建立开销&＃xff0c;如果一个集群的机器并不多&＃xff0c;可以考虑增加这个值 |
| spark.shuffle.io.preferDirectBufs | true | 启用堆外内存&＃xff0c;可以避免shuffle过程的频繁gc&＃xff0c;如果堆外内存非常紧张&＃xff0c;则可以考虑关闭这个选项 |
| spark.shuffle.manager | sort | ShuffleManager&＃xff0c;Spark 1.5以后&＃xff0c;有三种可选的&＃xff0c;hash、sort和tungsten-sort&＃xff0c;sort-based ShuffleManager会更高效实用内存&＃xff0c;并且避免产生大量的map side磁盘文件&＃xff0c;从Spark 1.2开始就是默认的选项&＃xff0c;tungsten-sort与sort类似&＃xff0c;但是内存性能更高 |
| spark.shuffle.memoryFraction | 0.2 | 如果spark.shuffle.spill属性为true&＃xff0c;那么该选项生效&＃xff0c;代表了executor内存中&＃xff0c;用于进行shuffle reduce side聚合的内存比例&＃xff0c;默认是20%&＃xff0c;如果内存充足&＃xff0c;建议调高这个比例&＃xff0c;给reduce聚合更多内存&＃xff0c;避免内存不足频繁读写磁盘 |
| spark.shuffle.service.enabled | false | 启用外部shuffle服务&＃xff0c;这个服务会安全地保存shuffle过程中&＃xff0c;executor写的磁盘文件&＃xff0c;因此executor即使挂掉也不要紧&＃xff0c;必须配合spark.dynamicAllocation.enabled属性设置为true&＃xff0c;才能生效&＃xff0c;而且外部shuffle服务必须进行安装和启动&＃xff0c;才能启用这个属性 |
| spark.shuffle.service.port | 7337 | 外部shuffle服务的端口号&＃xff0c;具体解释同上 |
| spark.shuffle.sort.bypassMergeThreshold | 200 | 对于sort-based ShuffleManager&＃xff0c;如果没有进行map side聚合&＃xff0c;而且reduce task数量少于这个值&＃xff0c;那么就不会进行排序&＃xff0c;如果你使用sort ShuffleManager&＃xff0c;而且不需要排序&＃xff0c;那么可以考虑将这个值加大&＃xff0c;直到比你指定的所有task数量都打&＃xff0c;以避免进行额外的sort&＃xff0c;从而提升性能 |
| spark.shuffle.spill | true | 当reduce side的聚合内存使用量超过了spark.shuffle.memoryFraction指定的比例时&＃xff0c;就进行磁盘的溢写操作 |
| spark.shuffle.spill.compress | true | 同上&＃xff0c;进行磁盘溢写时&＃xff0c;是否进行文件压缩&＃xff0c;使用spark.io.compression.codec属性指定的压缩器&＃xff0c;默认是snappy&＃xff0c;速度优先 |
京公网安备 11010802041100号