浪尖说spark的coalesce的利弊及原理

浪尖的粉丝应该很久没见浪尖发过spark源码解读的文章，今天浪尖在这里给大家分享一篇文章，帮助大家进一步理解rdd如何在spark中被计算的，同时解释一下coalesce降低分区的原理及使用问题。

主要是知识星球有人问到过coalesce方法的使用和原理的问题，并且参考阅读了网上关于coalesce方法的错误介绍，有了错误的理解，所以浪尖忙里偷闲给大家解释一下。

浪尖这里建议多看看spark源码上，spark源码我觉得是注释最全的一套源码了，而且整体代码逻辑比较清晰，就是scala高阶函数的使用会使得前期阅读的时候很头疼，但是不可否认spark是大家学习scala编程规范性的参考代码。

这里不得不吐槽一下：flink的代码写的很挫，注释又不好，感觉不太适合人们阅读学习。

1.  coalesce  函数start

对于Spark 算子使用，大家还是要经常翻看一下源码上的注释及理解一下spark 算子的源码实现逻辑，注释很多时候已经很清楚了讲了算子的应用场景及原理，比如本文要讲的关于coalesce函数的注释如下：

 /*** Return a new RDD that is reduced into `numPartitions` partitions.** This results in a narrow dependency, e.g. if you go from 1000 partitions* to 100 partitions, there will not be a shuffle, instead each of the 100* new partitions will claim 10 of the current partitions. If a larger number* of partitions is requested, it will stay at the current number of partitions.** However, if you're doing a drastic coalesce, e.g. to numPartitions = 1,* this may result in your computation taking place on fewer nodes than* you like (e.g. one node in the case of numPartitions = 1). To avoid this,* you can pass shuffle = true. This will add a shuffle step, but means the* current upstream partitions will be executed in parallel (per whatever* the current partitioning is).** @note With shuffle = true, you can actually coalesce to a larger number* of partitions. This is useful if you have a small number of partitions,* say 100, potentially with a few partitions being abnormally large. Calling* coalesce(1000, shuffle = true) will result in 1000 partitions with the* data distributed using a hash partitioner. The optional partition coalescer* passed in must be serializable.*/

注释的大致意思就是假设父rdd 1000分区，然后调用coalesce(100)，实际上就是将父rdd的1000分区分成100组，每组10个，叫做partitionGroup，每个partitionGroup作为coalescedrdd的一个分区，在compute方法中迭代处理，以此来避免shuffle。

coalesce函数总共三个参数：分区数，是否进行shuffle(默认不shuffle)，Coalesce分区器(用来决定哪些父rdd的分区组成一组，作为一个partitiongroup，也即是决定了coalescedrdd的分区情况)。

def coalesce(numPartitions: Int, shuffle: Boolean = false,partitionCoalescer: Option[PartitionCoalescer] = Option.empty)(implicit ord: Ordering[T] = null): RDD[T] = withScope {require(numPartitions > 0, s"Number of partitions ($numPartitions) must be positive.")if (shuffle) {/** Distributes elements evenly across output partitions, starting from a random partition. */val distributePartition = (index: Int, items: Iterator[T]) => {var position = new Random(hashing.byteswap32(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 distributednew CoalescedRDD(new ShuffledRDD[Int, T, T](mapPartitionsWithIndexInternal(distributePartition, isOrderSensitive = true),new HashPartitioner(numPartitions)),numPartitions,partitionCoalescer).values} else {new CoalescedRDD(this, numPartitions, partitionCoalescer)}}

研究coalescedrdd源码之前，浪尖觉得应该要强调一下rdd的五大特性，大家要不理解rdd的五大特性的话，很难理解本文的内容。

吐槽一下：

其实，大家对于RDD五大特性都能背诵入流，但是要说真正理解，浪尖感觉很多人还是差点。

 *  - A list of partitions*  - A function for computing each split*  - A list of dependencies on other RDDs*  - Optionally, a Partitioner for key-value RDDs (e.g. to say that the RDD is hash-partitioned)*  - Optionally, a list of preferred locations to compute each split on (e.g. block locations for*    an HDFS file)

也即：

每个RDD都有一系列的分区，每个rdd都有一系列的父rdd，也有一个针对rdd的当前分区的compute计算函数，可选的分区器和可选的本地性策略。

那么，提个问题: RDD和父RDD的分区关系是如何确定的？

这里又要强调五大特性了：

所有的RDD的分区数都是由getPartitions函数来确定分区，所有的RDD都是通过getDependencies()函数来确定依赖关系：窄依赖和宽依赖。而所有的rdd都是通过compute方法来计算rdd数据的。

coalesce函数的shuffle的我们这里就暂时不介绍了，只介绍不进行shuffle操作的功能，也即是：

new CoalescedRDD(this, numPartitions, partitionCoalescer)

2. getPartitions 分区分组

默认coalesce函数的partitionCoalescer为空，所以你要想自己实现父RDD分区分组策略也是可以的。对于CoalescedRDD，默认指定分区器为空，那么看一下其getPartitions函数，会使用默认的分区器DefaultPartitionCoalescer。

override def getPartitions: Array[Partition] = {val pc = partitionCoalescer.getOrElse(new DefaultPartitionCoalescer())pc.coalesce(maxPartitions, prev).zipWithIndex.map {case (pg, i) =>val ids = pg.partitions.map(_.index).toArrayCoalescedRDDPartition(i, prev, ids, pg.prefLoc)}}

可以看看DefaultPartitionCoalescer分区器的coalesce方法，实际上就是将父RDD的分区分组缩减为指定的分区数,该函数返回的就是Array[PartitionGroup]，每个PartitionGroup代表一组父RDD分区，也代表一个CoalescedRDD的分区。

 /*** Runs the packing algorithm and returns an array of PartitionGroups that if possible are* load balanced and grouped by locality** @return array of partition groups*/def coalesce(maxPartitions: Int, prev: RDD[_]): Array[PartitionGroup] = {val partitionLocs = new PartitionLocations(prev)// setup the groups (bins)setupGroups(math.min(prev.partitions.length, maxPartitions), partitionLocs)// assign partitions (balls) to each group (bins)throwBalls(maxPartitions, prev, balanceSlack, partitionLocs)getPartitions}

一个PartitionGroup实际上就是按照一定的规则组合的父RDD的partition数组，可以看一下该类。

/*** ::DeveloperApi::* A group of `Partition`s* @param prefLoc preferred location for the partition group*/
@DeveloperApi
class PartitionGroup(val prefLoc: Option[String] = None) {val partitions = mutable.ArrayBuffer[Partition]()def numPartitions: Int = partitions.size
}

3. getDependencies 血缘

上面说了，CoalescedRDD的getPartitions()方法，也就是完成了父RDD的分区到当前RDD分区的映射关系。这个映射关系的使用实际上就是通过getDependencies方法来调用的。具体如下：

 override def getDependencies: Seq[Dependency[_]] = {Seq(new NarrowDependency(prev) {def getParents(id: Int): Seq[Int] =partitions(id).asInstanceOf[CoalescedRDDPartition].parentsIndices})}

partitions数组是在RDD类里实现的，其实调用了getPartitions函数。

/*** Get the array of partitions of this RDD, taking into account whether the* RDD is checkpointed or not.*/final def partitions: Array[Partition] = {checkpointRDD.map(_.partitions).getOrElse {if (partitions_ == null) {partitions_ = getPartitionspartitions_.zipWithIndex.foreach { case (partition, index) =>require(partition.index == index,s"partitions($index).partition == ${partition.index}, but it should equal $index")}}partitions_}}

再说回窄依赖 NarrowDependency，其实他的getParents方法就是通过当前分区的id获取一个coalescedRDDPartition，也即一个父RDD分区数组。该数组是通过CoalescedRDD的getPartitions中实现的对父RDD分区分组得到的。

4. compute 计算分区

compute五大特性之一，针对分区的计算函数，对于CoalescedRDD，那么其计算函数的实现如下：

 override def compute(partition: Partition, context: TaskContext): Iterator[T] = {partition.asInstanceOf[CoalescedRDDPartition].parents.iterator.flatMap { parentPartition =>firstParent[T].iterator(parentPartition, context)}}

观察上述方法就会发现，是针对CoalescedRDDPartition的计算，这个其实是就是针对一个PartitionsGroup进行计算，也即使一个父RDD的分组。在getPartitions方法里生成的哦。

到这里就很明显了，coalescedrdd的compute方法虽然是针对Coalescedrdd的一个分区计算，实际上是计算的父RDD的一组RDD分区，降低了父RDD 的并行度哦，所以大家使用要慎重哦。

该使用shuffle决不能手软

5. shuffle模式 开篇

对于支持shuffle的Coalesce函数，我们可以看到其实是外层包括了一个shuffleRDD，同时CoalescedRDD传入的分区数和构建的父shuffleRDD一样，就实现了一对一分区转化,以此来实现shuffle功能的，针对shuffleRDD我们星球里分析分享。

if (shuffle) {/** Distributes elements evenly across output partitions, starting from a random partition. */val distributePartition = (index: Int, items: Iterator[T]) => {var position = new Random(hashing.byteswap32(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 distributednew CoalescedRDD(new ShuffledRDD[Int, T, T](mapPartitionsWithIndexInternal(distributePartition, isOrderSensitive = true),new HashPartitioner(numPartitions)),numPartitions,partitionCoalescer).values

欢迎大家加入浪尖知识星球哦~