zoukankan      html  css  js  c++  java
  • Spark Shuffle 机制

    1、spark shuffle:spark 的 shuffle 主要发生在 DAG 视图中的 stage 和 stage 之间,也就是RDD之间是宽依赖的时候,会发生 shuffle。

    补充:spark shuffle在很多地方也会参照mapreduce一样,将它分成两个阶段map阶段、reduce阶段。map阶段就是数据还在各个节点上的阶段,reduce阶段就是相同的key被拉到了相同的节点上后的阶段。

    2、shuffle对于spark性能的影响:shuffle过程包括大量的磁盘IO、序列化、网络数据传输等操作。因此,shuffle调优可以让spark作业性能得到提高。

    3、spark shuffle的四种策略:

      shuffle manager 就是 负责管理 shuffle 过程的执行、计算、处理的组件,即shuffle管理器。

      ShuffleManager随着Spark的发展有两种实现的方式,分别为HashShuffleManager(spark1.2之前使用)和SortShuffleManager,因此spark的Shuffle有Hash Shuffle和Sort Shuffle两种

      在Spark 1.2以前,默认的shuffle计算引擎是HashShuffleManager

      HashShuffleManager有着一个非常严重的弊端,就是会产生大量的中间磁盘文件,进而由大量的磁盘IO操作影响了性能。因此在Spark 1.2以后的版本中,默认的ShuffleManager改成了SortShuffleManager。

      SortShuffleManager相较于HashShuffleManager来说,有了一定的改进。主要就在于,每个Task在进行shuffle操作时,虽然也会产生较多的临时磁盘文件,但是最后会将所有的临时文件合并(merge)成一个磁盘文件,因此每个Task就只有一个磁盘文件。在下一个stage的shuffle read task拉取自己的数据时,只要根据索引读取每个磁盘文件中的部分数据即可。

    (1)未经优化的hashShuffle

      上游的stage的task对相同的key执行hash算法,从而将相同的key都写入到一个磁盘文件中,而每一个磁盘文件都只属于下游stage的一个task。在将数据写入磁盘之前,会先将数据写入到内存缓冲,当内存缓冲填满之后,才会溢写到磁盘文件中。但是这种策略的不足在于,下游有几个task,上游的每一个task都就都需要创建几个临时文件,每个文件中只存储key取hash之后相同的数据,导致了当下游的task任务过多的时候,上游会堆积大量的小文件。

    (2)经过优化的hashShuffle

      同一个excutor的所有task公用一个shuffleFileGroup

      在shuffle write过程中,上游stage的task就不是为下游stage的每个task创建一个磁盘文件了。此时会出现shuffleFileGroup的概念,每个shuffleFileGroup会对应一批磁盘文件,磁盘文件的数量与下游stage的task数量是相同的。一个Executor上有多少个CPU core,就可以并行执行多少个task。而第一批并行执行的每个task都会创建一个shuffleFileGroup,并将数据写入对应的磁盘文件内。当Executor的CPU core执行完一批task,接着执行下一批task时,下一批task就会复用之前已有的shuffleFileGroup,包括其中的磁盘文件。也就是说,此时task会将数据写入已有的磁盘文件中,而不会写入新的磁盘文件中。因此,consolidate机制允许不同的task复用同一批磁盘文件,这样就可以有效将多个task的磁盘文件进行一定程度上的合并,从而大幅度减少磁盘文件的数量,进而提升shuffle write的性能。

    注意:如果想使用优化之后的ShuffleManager,需要将:spark.shuffle.consolidateFiles调整为true。(当然,默认是开启的)

    未经优化:     上游的task数量:m   ,  下游的task数量:n    , 上游的executor数量:k (m>=k)   ,  总共的磁盘文件:m*n= 上游task数量  x  下游task的数量

    优化之后的: 上游的task数量:m    ,  下游的task数量:n   , 上游的executor数量:k (m>=k)   ,  总共的磁盘文件: k*n=  上游excutor的数量  x  下游task的数量

    (3)SortShuffle普通机制:

    shuffle write:mapper阶段,上一个stage得到最后的结果写出
    shuffle read :reduce阶段,下一个stage拉取上一个stage进行合并

      在普通模式下,数据会先写入一个内存数据结构中,此时根据不同的shuffle算子,可以选用不同的数据结构。如果是由聚合操作的shuffle算子,就是用map的数据结构(边聚合边写入内存),如果是join的算子,就使用array的数据结构(直接写入内存)。接着,每写一条数据进入内存数据结构之后,就会判断是否达到了某个临界值,如果达到了临界值的话,就会尝试的将内存数据结构中的数据溢写到磁盘,然后清空内存数据结构。
      在溢写到磁盘文件之前,会先根据key对内存数据结构中已有的数据进行排序,排序之后,会分批将数据写入磁盘文件。默认的batch数量是10000条,也就是说,排序好的数据,会以每批次1万条数据的形式分批写入磁盘文件,写入磁盘文件是通过Java的BufferedOutputStream实现的。BufferedOutputStream是Java的缓冲输出流,首先会将数据缓冲在内存中,当内存缓冲满溢之后再一次写入磁盘文件中,这样可以减少磁盘IO次数,提升性能。
      此时task将所有数据写入内存数据结构的过程中,会发生多次磁盘溢写,会产生多个临时文件,最后会将之前所有的临时文件都进行合并,最后会合并成为一个大文件。最终只剩下两个文件,一个是合并之后的数据文件,一个是索引文件(标识了下游各个task的数据在文件中的start offset与end offset)。最终再由下游的task根据索引文件读取相应的数据文件。最终文件的个数等于上游task的个数。

      自己的理解:sortshuffle机制就是先将数据写入到内存缓存,缓存达到阈值进行磁盘溢写,磁盘溢写是分批次进行的,并且在溢写之前对key进行排序。这样就形成了很多的批次排序磁盘文件,然后将这些批次key排序磁盘文件进行合并,就形成了一个task的总的key的排序文件,在为这个key排序文件创建个索引文件,这样下一个stage的task就可以根据索引去文件中拉取自己的数据了。

    (4)SortShuffle 的 ByPass 机制:

     

      此时上游stage的task会为每个下游stage的task都创建一个临时磁盘文件,并将数据按key进行hash然后根据key的hash值,将key写入对应的磁盘文件之中。当然,写入磁盘文件时也是先写入内存缓冲,缓冲写满之后再溢写到磁盘文件的。最后,同样会将所有临时磁盘文件都合并成一个磁盘文件,并创建一个单独的索引文件。

      自己的理解:bypass的就是不排序,还是用hash去为key分磁盘文件,分完之后再合并,形成一个索引文件和一个合并后的key hash文件。省掉了排序的性能。

     bypass机制与普通SortShuffleManager运行机制的不同在于:

      a、磁盘写机制不同;

      b、不会进行排序。

      也就是说,启用该机制的最大好处在于,shuffle write过程中,不需要进行数据的排序操作,也就节省掉了这部分的性能开销。

     触发bypass机制的条件:

    shuffle map task的数量小于 spark.shuffle.sort.bypassMergeThreshold 参数的值(默认200)或者不是聚合类的shuffle算子(比如groupByKey)

    参考博客:https://blog.csdn.net/qichangjian/article/details/88039576

  • 相关阅读:
    11.8-ros-navigation解析
    8.14-rqt_common_pluggins 详解
    8.1-roscomm详解
    8.15-ros-bag使用
    7.26-rosbridge-suit 解读
    7.26-roscpp_overview详解
    7.26-ROS其他有价值模块
    java dbutils查询数据库时无法给部分字段赋值原因
    java_获取多个文件夹下所有.java源码的总行数
    正则表达式-1-初识正则表达式
  • 原文地址:https://www.cnblogs.com/guoyu1/p/12049486.html
Copyright © 2011-2022 走看看