RDD转换操作算子 --- 分区类

RDD转换操作算子 — 分区类

• repartition — 表示重新分区，可大可小 ==> shuffle默认为true
• coalesce — 表示重分区时，不可增大 ==> shuffle默认为false，修改为true即可增大

@Test
  def repartitionTest: Unit ={
    // 创建RDD,赋予初始2个分区
    val rdd = sc.parallelize(Seq(1,2,3,4,5,6,7,8),2)
    println(rdd.repartition(4).partitions.size) // 4
    println(rdd.repartition(1).partitions.size) // 1
  }

@Test
  def coalesceTest: Unit ={
    val rdd = sc.parallelize(Seq(1,2,3,4,5,6,7,8),2)
    println(rdd.coalesce(4).partitions.size) // 不是4，还是2
    println(rdd.coalesce(4,shuffle = true).partitions.size) // 4
    println(rdd.coalesce(1).partitions.size) // 1
  }

• ​​partitionby​​ 针对 K-V 型RDD重分区

@Test
  def partitionby(): Unit ={
    /*
     * 针对 K-V 型RDD重分区
    */
    import org.apache.spark._
    val source = sc.parallelize(Seq(("a 5",1),("b 6",2),("c 7",3),("d 8",2),("a 5",2),("b 6",4)),2)
    val pb  = source.partitionBy(new HashPartitioner(4))
    println(pb.partitions.size) // 4
  }

补充：

@Test
  def textFileTest: Unit ={

    // 读取外部数据时也可以指定分区的数量，但是这只是最小分区数
    val data: RDD[String] = sc.textFile("src/main/scala/Rdd算子/测验/fix_1.csv",2)
    println(data.partitions.size) // 3

  }

1.分区的作用：

    1. RDD 经常需要通过读取外部数据来创建，外部数据存储系统往往是支持分片的，分区后能够使得与外部系统一一对应
    2. RDD 分区后支持了并行运算，也就是说分区是实现RDD并行运算的一个手段


2.分区和shuffle的关系：

    RDD分区的主要作用是支持并行运算，本质上与shuffle没什么关系。但是往往在进行数据处理的时候，例如 reduceByKey、groupByKey等聚合操作，
    要把Key相同的value值聚合到一起，有可能相同的key的value不在同一个分区中，所以要理解了分区，shuffle自然就可以简单些理解

3.shuffle操作的特点：

    1.除了 repartition 算子外，只有Key-Value型的RDD数据才可以进行shuffle操作
    2.shuffle 是从早期的 Hash base shuffle 进化而来的，更适用于大吞吐的场景


4.RDD 的 shuffle 原理 --- Hash base shuffle 和 Sort base shuffle：

     Hash base shuffle

        大致原理是分桶：

             * 假设Reducer有R个，则每个Mapper中有R个桶，按照Key的Hash将数据映射到不同桶中，Reduce找到每个Mapper中对应自己的桶拉取数据

             * 假设整个Mapper的数量为M，那么整个集群的文件数量为M*R，过多的文件会导致系统文件打开过多的文件描述符，占用资源，只适合处理中小文件。

     Sort base shuffle

        每个Map侧只产生一个输出文件，Reduce侧的Task来进行拉取：

              * Map侧将数据全部放入一个叫AppendOnlyMap的组件中，同时可以进行聚合操作
              * 然后通过类似于MergeSort的排序算法TimeSort对AppendOnlyMap底层的Array排序
                 - 先按照Partition ID排，再按照HashCode排
              * 最终每个MapTask生成一个输出文件，ReduceTask来来拉取自己对应的数据