spark 源码分析之一 -- RDD的四种依赖关系

Posted 2022-03-04 johnny666888

　RDD的四种依赖关系

　　RDD四种依赖关系，分别是 ShuffleDependency、PrunDependency、RangeDependency和OneToOneDependency四种依赖关系。如下图所示：org.apache.spark.Dependency有两个一级子类，分别是 ShuffleDependency 和 NarrowDependency。其中，NarrowDependency 是一个抽象类，它有三个实现类，分别是OneToOneDependency、RangeDependency和 PruneDependency。

　　

　RDD的窄依赖

　　我们先来看窄RDD是如何确定依赖的父RDD的分区的呢？NarrowDependency 定义了一个抽象方法，如下：

/**
   * Get the parent partitions for a child partition.
   * @param partitionId a partition of the child RDD
   * @return the partitions of the parent RDD that the child partition depends upon
   */
  def getParents(partitionId: Int): Seq[Int]

　　其输入参数是子RDD 的 分区Id，输出是子RDD 分区依赖的父RDD 的 partition 的 id 序列。

　　下面，分别看三种子类的实现：

　 OneToOneDependency

　　首先，OneToOneDependency的getParent实现如下：

override def getParents(partitionId: Int): List[Int] = List(partitionId)

　　就一行代码，实现比较简单，子RDD对应的partition index 跟父 RDD 的partition 的 index 一样。相当于父RDD 的 每一个partition 复制到 子RDD 的对应分区中，分区的关系是一对一的。RDD的关系也是一对一的。

　 RangeDependency

　　其次，RangeDependency的 getParent 实现如下：

　　

/**
 * :: DeveloperApi ::
 * Represents a one-to-one dependency between ranges of partitions in the parent and child RDDs.
 * @param rdd the parent RDD
 * @param inStart the start of the range in the parent RDD
 * @param outStart the start of the range in the child RDD
 * @param length the length of the range
 */
@DeveloperApi
class RangeDependency[T](rdd: RDD[T], inStart: Int, outStart: Int, length: Int)
  extends NarrowDependency[T](rdd) 

  override def getParents(partitionId: Int): List[Int] = 
    if (partitionId >= outStart && partitionId < outStart + length) 
      List(partitionId - outStart + inStart)
     else 
      Nil
    
  

　　首先解释三个变量：inStart：父RDD  range 的起始位置；outStart：子RDD range 的起始位置；length：range 的长度。

　　获取 父RDD 的partition index 的规则是：如果子RDD 的 partition index 在父RDD 的range 内，则返回的 父RDD partition是 子RDD partition index - 父 RDD 分区range 起始 + 子RDD 分区range 起始。其中，（- 父 RDD 分区range 起始 + 子RDD 分区range 起始）即 子RDD 的分区的 range 起始位置和 父RDD 的分区的 range 的起始位置 的相对距离。子RDD 的 parttion index 加上这个相对距离就是 对应父的RDD partition。否则是无依赖的父 RDD 的partition index。父子RDD的分区关系是一对一的。RDD 的关系可能是一对一（length 是1 ，就是特殊的 OneToOneDependency），也可能是多对一，也可能是一对多。

　 PruneDependency

      最后，PruneDependency的 getParent 实现如下：

 1 /**
 2  * Represents a dependency between the PartitionPruningRDD and its parent. In this
 3  * case, the child RDD contains a subset of partitions of the parents‘.
 4  */
 5 private[spark] class PruneDependency[T](rdd: RDD[T], partitionFilterFunc: Int => Boolean)
 6   extends NarrowDependency[T](rdd) 
 7 
 8   @transient
 9   val partitions: Array[Partition] = rdd.partitions
10     .filter(s => partitionFilterFunc(s.index)).zipWithIndex
11     .map  case(split, idx) => new PartitionPruningRDDPartition(idx, split) : Partition 
12 
13   override def getParents(partitionId: Int): List[Int] = 
14     List(partitions(partitionId).asInstanceOf[PartitionPruningRDDPartition].parentSplit.index)
15   
16 

　　首先，解释三个变量: rdd 是指向父RDD 的实例引用；partitionFilterFunc 是一个回调函数，作用是过滤出符合条件的父 RDD 的 partition 集合；PartitionPruningRDDPartition类声明如下：

private[spark] class PartitionPruningRDDPartition(idx: Int, val parentSplit: Partition)
  extends Partition 
  override val index = idx

　　partitions的生成过程如下： 先根据父RDD 引用获取父RDD 对应的 partition集合，然后根据过滤函数和partition index ，过滤出想要的父RDD 的 partition 集合并且从0 开始编号，最后，根据父RDD 的 partition 和 新编号实例化新的PartitionPruningRDDPartition实例，并放入到 partitions 集合中，相当于是先对parent RDD 的分区做Filter 剪枝操作。

　　在getParent 方法中， 先根据子RDD 的 partition index 获取 到对应的 parent RDD 的对应分区，然后获取Partition 的成员函数 index，该index 就是 父RDD 的 partition 在父RDD 的所有分区中的 index。 子RDD partition 和 父RDD partition的关系是 一对一的， 父RDD 和子RDD 的关系是 多对一，也可能是一对多，也可能是一对一。

　　简言之，在窄依赖中，子RDD 的partition 和 父RDD 的 partition 的关系是 一对一的。

　RDD 的宽依赖

　　下面重点看 ShuffleDependency，ShuffleDependency代表的是 一个 shuffle stage 的输出。先来看其构造方法，即其依赖的变量或实例：

1 @DeveloperApi
2 class ShuffleDependency[K: ClassTag, V: ClassTag, C: ClassTag](
3     @transient private val _rdd: RDD[_ <: Product2[K, V]],
4     val partitioner: Partitioner,
5     val serializer: Serializer = SparkEnv.get.serializer,
6     val keyOrdering: Option[Ordering[K]] = None,
7     val aggregator: Option[Aggregator[K, V, C]] = None,
8     val mapSideCombine: Boolean = false)
9   extends Dependency[Product2[K, V]]

　　其中，_rdd 代指父RDD实例；partitioner是用于给shuffle的输出分区的分区器；serializer，主要用于序列化，默认是org.apache.spark.serializer.JavaSerializer，可以通过`spark.serializer` 参数指定；keyOrdering RDD shuffle的key 的顺序。aggregator，map或reduce 端用于RDD shuffle的combine聚合器；mapSideCombine 是否执行部分的聚合（即 map端的预聚合，可以提高网络传输效率和reduce 端的执行效率），默认是false。因为并不是所有的都适合这样做。比如求全局平均值，均值，平方差等，但像全局最大值，最小值等是适合用mapSideCombine 的。注意，当mapSideCombine 为 true时， 必须设置combine聚合器，因为 shuffle 前需要使用聚合器做 map-combine 操作。

　 partitioner的7种实现

　　partitioner 定义了 RDD 里的key-value 对 是如何按 key 来分区的。映射每一个 key 到一个分区 id，从 0 到 分区数 - 1； 注意，分区器必须是确定性的，即给定同一个 key，必须返回同一个分区，便于任务失败时，追溯分区数据，确保了每一个要参与计算的分区数据的一致性。即 partition 确定了 shuffle 过程中 数据是要流向哪个具体的分区的。

　　org.apache.spark.Partition的 7 个实现类如下：

　　

　　我们先来看Partitioner 的方法定义：

1 abstract class Partitioner extends Serializable 
2   def numPartitions: Int
3   def getPartition(key: Any): Int
4 

　　其中，numPartitions 是返回 子RDD 的 partition 数量；getPartition 会根据指定的 key 返回 子RDD 的 partition index。

　　HashPartitioner 的 getPartition 的 实现如下，思路是 key.hashcode() mod 子RDD的 partition 数量：

1 def getPartition(key: Any): Int = key match 
2     case null => 0
3     case _ => Utils.nonNegativeMod(key.hashCode, numPartitions)
4   

　　RangePartitioner 的 getPartition 的实现如下：

 1 def getPartition(key: Any): Int = 
 2     val k = key.asInstanceOf[K]
 3     var partition = 0
 4     if (rangeBounds.length <= 128)  // 不大于 128 分区
 5       // If we have less than 128 partitions naive search
 6       while (partition < rangeBounds.length && ordering.gt(k, rangeBounds(partition))) 
 7         partition += 1
 8       
 9      else  // 大于 128 个分区数量
10       // Determine which binary search method to use only once.
11       partition = binarySearch(rangeBounds, k) // 二分查找
12       // binarySearch either returns the match location or -[insertion point]-1
13       if (partition < 0) 
14         partition = -partition-1
15       
16       if (partition > rangeBounds.length) 
17         partition = rangeBounds.length
18       
19     
20     if (ascending) 
21       partition
22      else 
23       rangeBounds.length - partition
24     
25   

　　PythonPartitioner 的 getPartition 如下，跟hash 很相似：

1 override def getPartition(key: Any): Int = key match 
2     case null => 0
3     // we don‘t trust the Python partition function to return valid partition ID‘s so
4     // let‘s do a modulo numPartitions in any case
5     case key: Long => Utils.nonNegativeMod(key.toInt, numPartitions)
6     case _ => Utils.nonNegativeMod(key.hashCode(), numPartitions)
7   

　　PartitionIdPassthrough 的 getPartition 如下：

1 override def getPartition(key: Any): Int = key.asInstanceOf[Int]

　　GridPartitioner 的 getPartition 如下，思想，二元组定位到网格的partition：

 1 override val numPartitions: Int = rowPartitions * colPartitions
 2 
 3   /**
 4    * Returns the index of the partition the input coordinate belongs to.
 5    *
 6    * @param key The partition id i (calculated through this method for coordinate (i, j) in
 7    *            `simulateMultiply`, the coordinate (i, j) or a tuple (i, j, k), where k is
 8    *            the inner index used in multiplication. k is ignored in computing partitions.
 9    * @return The index of the partition, which the coordinate belongs to.
10    */
11   override def getPartition(key: Any): Int = 
12     key match 
13       case i: Int => i
14       case (i: Int, j: Int) =>
15         getPartitionId(i, j)
16       case (i: Int, j: Int, _: Int) =>
17         getPartitionId(i, j)
18       case _ =>
19         throw new IllegalArgumentException(s"Unrecognized key: $key.")
20     
21   
22 
23   /** Partitions sub-matrices as blocks with neighboring sub-matrices. */
24   private def getPartitionId(i: Int, j: Int): Int = 
25     require(0 <= i && i < rows, s"Row index $i out of range [0, $rows).")
26     require(0 <= j && j < cols, s"Column index $j out of range [0, $cols).")
27     i / rowsPerPart + j / colsPerPart * rowPartitions
28   

　　包括匿名类，还有好多种，就不一一介绍了。总而言之，宽依赖是根据partitioner 确定 分区内的数据具体到哪个分区。

　　至此，RDD 的窄依赖和宽依赖都介绍清楚了。