ITPub博客

首页 > 大数据 > 数据分析 > spark Shuffle相关

spark Shuffle相关

数据分析 作者:破棉袄 时间:2015-09-30 14:21:53 0 删除 编辑

 

Shuffle 相关

Shuffle操作大概是对Spark性能影响最大的步骤之一(因为可能涉及到排序,磁盘IO,网络IO等众多CPU或IO密集的操作),这也是为什么在Spark 1.1的代码中对整个Shuffle框架代码进行了重构,将Shuffle相关读写操作抽象封装到Pluggable的Shuffle Manager中,便于试验和实现不同的Shuffle功能模块。例如为了解决Hash Based的Shuffle Manager在文件读写效率方面的问题而实现的Sort Base的Shuffle Manager。

spark.shuffle.manager

用来配置所使用的Shuffle Manager,目前可选的Shuffle Manager包括默认的org.apache.spark.shuffle.sort.HashShuffleManager(配置参数值为hash)和新的org.apache.spark.shuffle.sort.SortShuffleManager(配置参数值为sort)。

这两个ShuffleManager如何选择呢,首先需要了解他们在实现方式上的区别。

HashShuffleManager,故名思义也就是在Shuffle的过程中写数据时不做排序操作,只是将数据根据Hash的结果,将各个Reduce分区的数据写到各自的磁盘文件中。带来的问题就是如果Reduce分区的数量比较大的话,将会产生大量的磁盘文件。如果文件数量特别巨大,对文件读写的性能会带来比较大的影响,此外由于同时打开的文件句柄数量众多,序列化,以及压缩等操作需要分配的临时内存空间也可能会迅速膨胀到无法接受的地步,对内存的使用和GC带来很大的压力,在Executor内存比较小的情况下尤为突出,例如Spark on Yarn模式。

SortShuffleManager,是1.1版本之后实现的一个试验性(也就是一些功能和接口还在开发演变中)的ShuffleManager,它在写入分区数据的时候,首先会根据实际情况对数据采用不同的方式进行排序操作,底线是至少按照Reduce分区Partition进行排序,这样来至于同一个Map任务Shuffle到不同的Reduce分区中去的所有数据都可以写入到同一个外部磁盘文件中去,用简单的Offset标志不同Reduce分区的数据在这个文件中的偏移量。这样一个Map任务就只需要生成一个shuffle文件,从而避免了上述HashShuffleManager可能遇到的文件数量巨大的问题

两者的性能比较,取决于内存,排序,文件操作等因素的综合影响。

对于不需要进行排序的Shuffle操作来说,如repartition等,如果文件数量不是特别巨大,HashShuffleManager面临的内存问题不大,而SortShuffleManager需要额外的根据Partition进行排序,显然HashShuffleManager的效率会更高。

而对于本来就需要在Map端进行排序的Shuffle操作来说,如ReduceByKey等,使用HashShuffleManager虽然在写数据时不排序,但在其它的步骤中仍然需要排序,而SortShuffleManager则可以将写数据和排序两个工作合并在一起执行,因此即使不考虑HashShuffleManager的内存使用问题,SortShuffleManager依旧可能更快。

spark.shuffle.sort.bypassMergeThreshold

这个参数仅适用于SortShuffleManager,如前所述,SortShuffleManager在处理不需要排序的Shuffle操作时,由于排序带来性能的下降。这个参数决定了在这种情况下,当Reduce分区的数量小于多少的时候,在SortShuffleManager内部不使用Merge Sort的方式处理数据,而是与Hash Shuffle类似,直接将分区文件写入单独的文件,不同的是,在最后一步还是会将这些文件合并成一个单独的文件。这样通过去除Sort步骤来加快处理速度,代价是需要并发打开多个文件,所以内存消耗量增加,本质上是相对HashShuffleMananger一个折衷方案。 这个参数的默认值是200个分区,如果内存GC问题严重,可以降低这个值。

spark.shuffle.consolidateFiles

这个配置参数仅适用于HashShuffleMananger的实现,同样是为了解决生成过多文件的问题,采用的方式是在不同批次运行的Map任务之间重用Shuffle输出文件,也就是说合并的是不同批次的Map任务的输出数据,但是每个Map任务所需要的文件还是取决于Reduce分区的数量,因此,它并不减少同时打开的输出文件的数量,因此对内存使用量的减少并没有帮助。只是HashShuffleManager里的一个折中的解决方案。

需要注意的是,这部分的代码实现尽管原理上说很简单,但是涉及到底层具体的文件系统的实现和限制等因素,例如在并发访问等方面,需要处理的细节很多,因此一直存在着这样那样的bug或者问题,导致在例如EXT3上使用时,特定情况下性能反而可能下降,因此从Spark 0.8的代码开始,一直到Spark 1.1的代码为止也还没有被标志为Stable,不是默认采用的方式。此外因为并不减少同时打开的输出文件的数量,因此对性能具体能带来多大的改善也取决于具体的文件数量的情况。所以即使你面临着Shuffle文件数量巨大的问题,这个配置参数是否使用,在什么版本中可以使用,也最好还是实际测试以后再决定。

spark.shuffle.spill

shuffle的过程中,如果涉及到排序,聚合等操作,势必会需要在内存中维护一些数据结构,进而占用额外的内存。如果内存不够用怎么办,那只有两条路可以走,一就是out of memory 出错了,二就是将部分数据临时写到外部存储设备中去,最后再合并到最终的Shuffle输出文件中去。

这里spark.shuffle.spill 决定是否Spill到外部存储设备(默认打开),如果你的内存足够使用,或者数据集足够小,当然也就不需要Spill,毕竟Spill带来了额外的磁盘操作。

spark.shuffle.memoryFraction / spark.shuffle.safetyFraction

在启用Spill的情况下,spark.shuffle.memoryFraction(1.1后默认为0.2)决定了当Shuffle过程中使用的内存达到总内存多少比例的时候开始Spill。

通过spark.shuffle.memoryFraction可以调整Spill的触发条件,即Shuffle占用内存的大小,进而调整Spill的频率和GC的行为。总的来说,如果Spill太过频繁,可以适当增加spark.shuffle.memoryFraction的大小,增加用于Shuffle的内存,减少Spill的次数。当然这样一来为了避免内存溢出,对应的可能需要减少RDD cache占用的内存,即减小spark.storage.memoryFraction的值,这样RDD cache的容量减少,有可能带来性能影响,因此需要综合考虑。

由于Shuffle数据的大小是估算出来的,一来为了降低开销,并不是每增加一个数据项都完整的估算一次,二来估算也会有误差,所以实际暂用的内存可能比估算值要大,这里spark.shuffle.safetyFraction(默认为0.8)用来作为一个保险系数,降低实际Shuffle使用的内存阀值,增加一定的缓冲,降低实际内存占用超过用户配置值的概率。

spark.shuffle.spill.compress / spark.shuffle.compress

这两个配置参数都是用来设置Shuffle过程中是否使用压缩算法对Shuffle数据进行压缩,前者针对Spill的中间数据,后者针对最终的shuffle输出文件,默认都是True

理论上说,spark.shuffle.compress设置为True通常都是合理的,因为如果使用千兆以下的网卡,网络带宽往往最容易成为瓶颈。此外,目前的Spark任务调度实现中,以Shuffle划分Stage,下一个Stage的任务是要等待上一个Stage的任务全部完成以后才能开始执行,所以shuffle数据的传输和CPU计算任务之间通常不会重叠,这样Shuffle数据传输量的大小和所需的时间就直接影响到了整个任务的完成速度。但是压缩也是要消耗大量的CPU资源的,所以打开压缩选项会增加Map任务的执行时间,因此如果在CPU负载的影响远大于磁盘和网络带宽的影响的场合下,也可能将spark.shuffle.compress 设置为False才是最佳的方案

对于spark.shuffle.spill.compress而言,情况类似,但是spill数据不会被发送到网络中,仅仅是临时写入本地磁盘,而且在一个任务中同时需要执行压缩和解压缩两个步骤,所以对CPU负载的影响会更大一些,而磁盘带宽(如果标配12HDD的话)可能往往不会成为Spark应用的主要问题,所以这个参数相对而言,或许更有机会需要设置为False。

总之,Shuffle过程中数据是否应该压缩,取决于CPU/DISK/NETWORK的实际能力和负载,应该综合考虑。

spark.locality.wait

spark.locality.wait和spark.locality.wait.process,spark.locality.wait.node, spark.locality.wait.rack这几个参数影响了任务分配时的本地性策略的相关细节。

Spark中任务的处理需要考虑所涉及的数据的本地性的场合,基本就两种,一是数据的来源是HadoopRDD; 二是RDD的数据来源来自于RDD Cache(即由CacheManager从BlockManager中读取,或者Streaming数据源RDD)。其它情况下,如果不涉及shuffle操作的RDD,不构成划分Stage和Task的基准,不存在判断Locality本地性的问题,而如果是ShuffleRDD,其本地性始终为No Prefer,因此其实也无所谓Locality。

在理想的情况下,任务当然是分配在可以从本地读取数据的节点上时(同一个JVM内部或同一台物理机器内部)的运行时性能最佳。但是每个任务的执行速度无法准确估计,所以很难在事先获得全局最优的执行策略,当Spark应用得到一个计算资源的时候,如果没有可以满足最佳本地性需求的任务可以运行时,是退而求其次,运行一个本地性条件稍差一点的任务呢,还是继续等待下一个可用的计算资源已期望它能更好的匹配任务的本地性呢?

这几个参数一起决定了Spark任务调度在得到分配任务时,选择暂时不分配任务,而是等待获得满足进程内部/节点内部/机架内部这样的不同层次的本地性资源的最长等待时间。默认都是3000毫秒。

基本上,如果你的任务数量较大和单个任务运行时间比较长的情况下,单个任务是否在数据本地运行,代价区别可能比较显著,如果数据本地性不理想,那么调大这些参数对于性能优化可能会有一定的好处。反之如果等待的代价超过带来的收益,那就不要考虑了。

特别值得注意的是:在处理应用刚启动后提交的第一批任务时,由于当作业调度模块开始工作时,处理任务的Executors可能还没有完全注册完毕,因此一部分的任务会被放置到No Prefer的队列中,这部分任务的优先级仅次于数据本地性满足Process级别的任务,从而被优先分配到非本地节点执行,如果的确没有Executors在对应的节点上运行,或者的确是No Prefer的任务(如shuffleRDD),这样做确实是比较优化的选择,但是这里的实际情况只是这部分Executors还没来得及注册上而已。这种情况下,即使加大本节中这几个参数的数值也没有帮助。针对这个情况,有一些已经完成的和正在进行中的PR通过例如动态调整No Prefer队列,监控节点注册比例等等方式试图来给出更加智能的解决方案。不过,你也可以根据自身集群的启动情况,通过在创建SparkContext之后,主动Sleep几秒的方式来简单的解决这个问题。

来自 “ ITPUB博客 ” ,链接:http://blog.itpub.net/29754888/viewspace-1812412/,如需转载,请注明出处,否则将追究法律责任。

上一篇: Akka与Storm的对比
请登录后发表评论 登录
全部评论

注册时间:2014-07-16

  • 博文量
    180
  • 访问量
    1111232