小分区合并问题-Coalesce()方法和repartition方法

系统管理员 2022-12-13 00:44 153阅读 0赞

小分区合并问题介绍：

在使用spark进行数据处理的过程中，常会使用filter方法对数据进行一些预处理，过滤掉一些不符合条件的数据。在使用该方法对数据进行频繁过滤或者是过滤掉的数据量过大的情况下就会造成大量小分区的生成。在spark内部会对每一个分区分配一个task来执行任务，如果task过多，那么每个task处理的数据量就会很小，造成线程频繁在task之间切换，使得资源开销较大，且很多任务等待执行，并行度不高，最终造成集群工作效益低下。

为了解决这一个问题，常采用RDD中重分区的函数（coalesce函数或rePartition函数）来进行数据紧缩，减少分区数量，将小分区合并为大分区，从而提高效率。

先介绍下**宽依赖（发生shuffle）**和**窄依赖（不发生shuffle）**

*  窄依赖：父Rdd的分区最多只能被一个子Rdd的分区所引用，即一个父Rdd的分区对应一个子Rdd的分区，或者多个父Rdd的分区对应一个子Rdd的分区。即**一对一**或**多对一**，如下图左边所示。
 *  宽依赖：RDD的分区依赖于父RDD的多个分区或所有分区，即存在一个父RDD的一个分区对应一个子RDD的多个分区。1个父RDD分区对应多个子RDD分区，这其中又分两种情况：1个父RDD对应所有子RDD分区（未经协同划分的Join）或者1个父RDD对应非全部的多个RDD分区（如groupByKey）。即**一对多**

![这里写图片描述][70]

### Coalesce()方法和rePartition()方法 ###

Coalesce()方法源码：

def coalesce(numPartitions: Int, shuffle: Boolean = false)(implicit ord: Ordering[T] = null)
        : RDD[T] = withScope {
      if (shuffle) {
        /** Distributes elements evenly across output partitions, starting from a random partition. */
        val distributePartition = (index: Int, items: Iterator[T]) => {
          var position = (new Random(index)).nextInt(numPartitions)
          items.map { t =>
            // Note that the hash code of the key will just be the key itself. The HashPartitioner
            // will mod it with the number of total partitions.
            position = position + 1
            (position, t)
          }
        } : Iterator[(Int, T)]
    
        // include a shuffle step so that our upstream tasks are still distributed
        new CoalescedRDD(
          new ShuffledRDD[Int, T, T](mapPartitionsWithIndex(distributePartition),
          new HashPartitioner(numPartitions)),
          numPartitions).values
      } else {
        new CoalescedRDD(this, numPartitions)
      }
    }

rePartition()方法源码：

def repartition(numPartitions: Int)(implicit ord: Ordering[T] = null): RDD[T] = withScope {
        coalesce(numPartitions, shuffle = true)
      }

通过源码可以看出**两者的区别**：**coalesce()方法的参数shuffle默认设置为false，repartition()方法就是coalesce()方法shuffle为true的情况。**

### 使用情景 ###

假设RDD有N个分区，需要重新划分成M个分区：

1.  N < M: 一般情况下N个分区有数据分布不均匀的状况，利用HashPartitioner函数将数据重新分区为M个，这时需要将shuffle设置为true。**因为重分区前后相当于宽依赖，会发生shuffle过程**，此时可以使用coalesce(shuffle=true)，或者直接使用repartition()。
2.  如果N > M并且N和M相差不多(假如N是1000，M是100): 那么就可以将N个分区中的若干个分区合并成一个新的分区，最终合并为M个分区，这是**前后是窄依赖关系**，可以使用coalesce(shuffle=false)。
3.  如果 N> M并且两者相差悬殊: 这时如果将shuffle设置为false，父子ＲＤＤ是窄依赖关系，他们同处在一个Ｓｔａｇｅ中，就可能造成spark程序的并行度不够，从而影响性能，如果在M为1的时候，为了使coalesce之前的操作有更好的并行度，可以将shuffle设置为true。

### 总结 ###

**如果传入的参数大于现有的分区数目，而shuffle为false，RDD的分区数不变，也就是说不经过shuffle，是无法将RDDde分区数变多的。**

[70]: /images/20221123/de2cc7429568477593e63d2fc09773a3.png