spark-shuffle调优

原创
2016/10/29 15:45
阅读数 157

什么情况下会发生shuffle,然后shuffle的原理是什么?

  • 在spark中,主要是以下几个算子:groupByKey、reduceByKey、countByKey、join,等等。
  • groupByKey,要把分布在集群各个节点上的数据中的同一个key,对应的values,都给集中到一块儿,集中到集群中同一个节点上,更严密一点说,就是集中到一个节点的一个executor的一个task中。然后呢,集中一个key对应的values之后,才能交给我们来进行处理,<key, Iterable<value>>;reduceByKey,算子函数去对values集合进行reduce操作,最后变成一个value;countByKey,需要在一个task中,获取到一个key对应的所有的value,然后进行计数,统计总共有多少个value;join,RDD<key, value>,RDD<key, value>,只要是两个RDD中,key相同对应的2个value,都能到一个节点的executor的task中,给我们进行处理。
  • 问题在于,同一个单词,比如说(hello, 1),可能散落在不同的节点上;对每个单词进行累加计数,就必须让所有单词都跑到同一个节点的一个task中,给一个task来进行处理;
  • 每一个shuffle的前半部分stage的task,每个task都会创建下一个stage的task数量相同的文件,比如下一个stage会有100个task,那么当前stage每个task都会创建100份文件;会将同一个key对应的values,一定是写入同一个文件中的;
  • shuffle的后半部分stage的task,每个task都会从各个节点上的task写的属于自己的那一份文件中,拉取key, value对;然后task会有一个内存缓冲区,然后会用HashMap,进行key, values的汇聚;(key ,values);
  • task会用我们自己定义的聚合函数,比如reduceByKey(_+_),把所有values进行一对一的累加,聚合出来最终的值。就完成了shuffle;
  • shuffle,一定是分为两个stage来完成的。因为这其实是个逆向的过程,不是stage决定shuffle,是shuffle决定stage。
  • reduceByKey(_+_),在某个action触发job的时候,DAGScheduler,会负责划分job为多个stage。划分的依据,就是,如果发现有会触发shuffle操作的算子,比如reduceByKey,就将这个操作的前半部分,以及之前所有的RDD和transformation操作,划分为一个stage;shuffle操作的后半部分,以及后面的,直到action为止的RDD和transformation操作,划分为另外一个stage;
  • shuffle前半部分的task在写入数据到磁盘文件之前,都会先写入一个一个的内存缓冲,内存缓冲满溢之后,再spill溢写到磁盘文件中。

如果不合并map端输出文件的话,会怎么样?

  1. 减少网络传输、disk io、减少reduce端内存缓冲
    • 问题来了:默认的这种shuffle行为,对性能有什么样的恶劣影响呢?
      • 实际生产环境的条件:
        • 100个节点(每个节点一个executor):100个executor,每个executor:2个cpu core,总共1000个task:每个executor平均10个task,上游1000个task,下游1000个task,每个节点,10个task,每个节点或者说每一个executor会输出多少份map端文件?10 * 1000=1万个文件(M*R)
        • 总共有多少份map端输出文件?100 * 10000 = 100万。
    • shuffle中的写磁盘的操作,基本上就是shuffle中性能消耗最为严重的部分。
    • 通过上面的分析,一个普通的生产环境的spark job的一个shuffle环节,会写入磁盘100万个文件。
    • 磁盘IO对性能和spark作业执行速度的影响,是极其惊人和吓人的。
    • 基本上,spark作业的性能,都消耗在shuffle中了,虽然不只是shuffle的map端输出文件这一个部分,但是这里也是非常大的一个性能消耗点。
      new SparkConf().set("spark.shuffle.consolidateFiles", "true")

       

  2. 开启shuffle map端输出文件合并的机制;默认情况下,是不开启的,就是会发生如上所述的大量map端输出文件的操作,严重影响性能。

  • 开启了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为自己创建的那份输出文件了;


提醒一下(map端输出文件合并):

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

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

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

实际生产环境的条件:
100个节点(每个节点一个executor):100个executor
每个executor:2个cpu core
总共1000个task:每个executor平均10个task
上游1000个task,下游1000个task


每个节点,2个cpu core,有多少份输出文件呢?2 * 1000 = 2000个(C*R)
总共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,主要拉取1000 * 2 = 2000个文件即可;网络传输的性能消耗是不是也大大减少分享一下,实际在生产环境中,使用了spark.shuffle.consolidateFiles机制以后,实际的性能调优的效果:对于上述的这种生产环境的配置,性能的提升,还是相当的客观的。

spark作业,5个小时 -> 2~3个小时。

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

这个时候,去开启这个机制,可以很有效的提升性能。
spark.shuffle.manager hash M*R 个小文件
spark.shuffle.manager sort   C*R 个小文件  (默认的shuffle管理机制

spark.shuffle.file.buffer,默认32k
spark.shuffle.memoryFraction,0.2

        默认情况下,shuffle的map task,输出到磁盘文件的时候,统一都会先写入每个task自己关联的一个内存缓冲区。这个缓冲区大小,默认是32kb。每一次,当内存缓冲区满溢之后,才会进行spill操作,溢写操作,溢写到磁盘文件中去reduce端task,在拉取到数据之后,会用hashmap的数据格式,来对各个key对应的values进行汇聚。针对每个key对应的values,执行我们自定义的聚合函数的代码,比如_ + _(把所有values累加起来)reduce task,在进行汇聚、聚合等操作的时候,实际上,使用的就是自己对应的executor的内存,executor(jvm进程,堆),默认executor内存中划分给reduce task进行聚合的比例,是0.2。问题来了,因为比例是0.2,所以,理论上,很有可能会出现,拉取过来的数据很多,那么在内存中,放不下;这个时候,默认的行为,就是说,将在内存放不下的数据,都spill(溢写)到磁盘文件中去。

原理说完之后,来看一下,默认情况下,不调优,可能会出现什么样的问题?

默认,map端内存缓冲是每个task,32kb。
默认,reduce端聚合内存比例,是0.2,也就是20%。

如果map端的task,处理的数据量比较大,但是呢,你的内存缓冲大小是固定的。
可能会出现什么样的情况?

每个task就处理320kb,32kb,总共会向磁盘溢写320 / 32 = 10次。
每个task处理32000kb,32kb,总共会向磁盘溢写32000 / 32 = 1000次。

在map task处理的数据量比较大的情况下,而你的task的内存缓冲默认是比较小的,32kb。可能会造成多次的map端往磁盘文件的spill溢写操作,发生大量的磁盘IO,从而降低性能。

reduce端聚合内存,占比。默认是0.2。如果数据量比较大,reduce task拉取过来的数据很多,那么就会频繁发生reduce端聚合内存不够用,频繁发生spill操作,溢写到磁盘上去。而且最要命的是,磁盘上溢写的数据量越大,后面在进行聚合操作的时候,很可能会多次读取磁盘中的数据,进行聚合。

默认不调优,在数据量比较大的情况下,可能频繁地发生reduce端的磁盘文件的读写。

这两个点之所以放在一起讲,是因为他们俩是有关联的。数据量变大,map端肯定会出点问题;
reduce端肯定也会出点问题;出的问题是一样的,都是磁盘IO频繁,变多,影响性能。

调优:

调节map task内存缓冲:spark.shuffle.file.buffer,默认32k(spark 1.3.x不是这个参数,
后面还有一个后缀,kb;spark 1.5.x以后,变了,就是现在这个参数)
调节reduce端聚合内存占比:spark.shuffle.memoryFraction,0.2

在实际生产环境中,我们在什么时候来调节两个参数?

看Spark UI,如果你的公司是决定采用standalone模式,那么很简单,你的spark跑起来,会显示一个Spark UI的地址,4040的端口,进去看,依次点击进去,可以看到,你的每个stage的详情,有哪些executor,有哪些task,每个task的shuffle write和shuffle read的量,shuffle的磁盘和内存,读写的数据量;如果是用的yarn模式来提交,课程最前面,从yarn的界面进去,点击对应的application,进入Spark UI,查看详情。

如果发现shuffle 磁盘的write和read,很大,可以调节这两个参数

调节上面说的那两个参数。调节的时候的原则。spark.shuffle.file.buffer,每次扩大一倍,然后看看效果,64,128;spark.shuffle.memoryFraction,每次提高0.1,看看效果。不能调节的太大,太大了以后过犹不及,因为内存资源是有限的,你这里调节的太大了,其他环节的内存使用就会有问题了。

调节了以后,效果?map task内存缓冲变大了,减少spill到磁盘文件的次数;reduce端聚合内存变大了,
减少spill到磁盘的次数,而且减少了后面聚合读取磁盘文件的数量。
 

展开阅读全文
打赏
1
2 收藏
分享
加载中
更多评论
打赏
0 评论
2 收藏
1
分享
返回顶部
顶部