标签：长度 部分 hpa wpa 设置 block += park 参数
RDD四种依赖关系，分别是 ShuffleDependency、PrunDependency、RangeDependency和OneToOneDependency四种依赖关系。如下图所示：org.apache.spark.Dependency有两个一级子类，分别是 ShuffleDependency 和 NarrowDependency。其中，NarrowDependency 是一个抽象类，它有三个实现类，分别是OneToOneDependency、RangeDependency和 PruneDependency。
我们先来看窄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的getParent实现如下：
override def getParents(partitionId: Int): List[Int] = List(partitionId)
就一行代码，实现比较简单，子RDD对应的partition index 跟父 RDD 的partition 的 index 一样。相当于父RDD 的 每一个partition 复制到 子RDD 的对应分区中，分区的关系是一对一的。RDD的关系也是一对一的。
其次，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的 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 的关系是 一对一的。
下面重点看 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 定义了 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 的窄依赖和宽依赖都介绍清楚了。
标签：长度 部分 hpa wpa 设置 block += park 参数
原文地址：https://www.cnblogs.com/johnny666888/p/11111957.html
