zoukankan      html  css  js  c++  java
  • SparkShuffle

    SparkShuffle概念:

    reduceByKey会将上一个RDD中的每一个key对应的所有value聚合成一个value,然后生成一个新的RDD,元素类型是<key,value>对的形式,这样每一个key对应一个聚合起来的value。

    问题:聚合之前,每一个key对应的value不一定都是在一个partition中,也不太可能在同一个节点上,因为RDD是分布式的弹性的数据集,RDD的partition极有可能分布在各个节点上,那么它是如何聚合的呢?

    这就涉及到shuffle的读写的两个过程了:

    – Shuffle Write:上一个stage的每个map task就必须保证将自己处理的当前分区的数据相同的key写入一个分区文件中,可能会写入多个不同的分区文件中。

     – Shuffle Read:reduce task就会从上一个stage的所有task所在的机器上寻找属于己的那些分区文件,这样就可以保证每一个key所对应的value都会汇聚到同一个节点上去处理和聚合。

    Spark中有两种Shuffle类型,HashShuffle和SortShuffle

    (1)Spark1.2之前是HashShuffle默认的分区器是HashPartitioner

    (2)Spark1.2引入SortShuffle默认的分区器是RangePartitioner

    HashShuffle:

    Hash Shuffle又分为两种:

    (1)普通机制:

    a)     每一个map task将不同结果写到不同的buffer中,每个buffer的大小为32K。buffer起到数据缓存的作用。

    b)     每个buffer文件最后对应一个磁盘小文件。

    c)     reduce task来拉取对应的磁盘小文件。

    总结:

    ①   .map task的计算结果会根据分区器(默认是hashPartitioner)来决定写入到哪一个磁盘小文件中去。ReduceTask会去Map端拉取相应的磁盘小文件。

    ②   .产生的磁盘小文件的个数:M(map task的个数)*R(reduce task的个数)

    存在的问题:

    产生的磁盘小文件过多,会导致以下问题:

    a)     在Shuffle Write过程中会产生很多写磁盘小文件的对象。

    b)     在Shuffle Read过程中会产生很多读取磁盘小文件的对象。

    c)     在JVM堆内存中对象过多会造成频繁的gc,gc还无法解决运行所需要的内存 的话,就会OOM(Out Of Memory)。

    d)     在数据传输过程中会有频繁的网络通信,频繁的网络通信出现通信故障的可能性大大增加,一旦网络通信出现了故障会导致shuffle file cannot find 由于这个错误导致的task失败,TaskScheduler不负责重试,由DAGScheduler负责重试Stage。

    (2)     合并机制

    总结:

    产生磁盘小文件的个数:C(core的个数)*R(reduce的个数),并不是Excutor来决定的,图上的CPU为1Core的,如果是双Core的话,则是2*reduce的个数。

    SortShuffle:

    1)     普通机制

    执行过程:

    a)     map task 的计算结果会写入到一个内存数据结构里面,内存数据结构默认是5M

    b)     在shuffle的时候会有一个定时器,不定期的去估算这个内存结构的大小,当内存结构中的数据超过5M时,比如现在内存结构中的数据为5.01M,那么他会申请5.01*2-5=5.02M内存给内存数据结构。

    c)     如果申请成功不会进行溢写,如果申请不成功,这时候会发生溢写磁盘。

    d)     在溢写之前内存结构中的数据会进行排序分区

    e)     然后开始溢写磁盘,写磁盘是以batch的形式去写,一个batch是1万条数据,

    f)      map task执行完成后,会将这些磁盘小文件合并成一个大的磁盘文件,同时生成一个索引文件。

    g)     reduce task去map端拉取数据的时候,首先解析索引文件,根据索引文件再去拉取对应属于自己分区的数据。

    总结:

    产生磁盘小文件的个数: 2*M(map task的个数)

    1)     bypass机制

    总结

    ①   .bypass运行机制的触发条件如下:

    shuffle reduce task的数量小于spark.shuffle.sort.bypassMergeThreshold的参数值。这个值默认是200

    ②   .产生的磁盘小文件为:2*M(map task的个数)

  • 相关阅读:
    Flink -- Java Generics Programming
    SpringBoot-annotation
    RDMA in CloudComputing
    ARM-IoT
    Netty
    TiDB, Distributed Database
    microkernel architecture
    Web网站的几个并发量级
    vtiger 下载地址
    php分布式缓存系统 Memcached 入门
  • 原文地址:https://www.cnblogs.com/yehuili/p/10466137.html
Copyright © 2011-2022 走看看