本文假设我们已经对spark有一定的使用经验，并对spark常见对一些名词有一定的理解。

Spark DAG图中的Stage的划分依据是宽依赖和窄依赖，遇到宽依赖则会划分新的stage。今天主要分析下spark依赖的源码实现，spark依赖定义在org.apache.spark.Dependency中，如图所示：

image.png

依赖基类 org.apache.spark.Dependency

所有的依赖均继承于该类，包括宽依赖（也称ShufferDependency）和窄依赖

/**
 * :: DeveloperApi ::
 * Base class for dependencies.
 */
@DeveloperApi
abstract class Dependency[T] extends Serializable {
  def rdd: RDD[T]
}

窄依赖 org.apache.spark.NarrowDependency

窄依赖(Narrow Dependency):如果子RDD的每个分区依赖于父RDD的少量分区，那么这种关系我们称为窄依赖。换言之，每一个父RDD中的Partition最多被子RDD的1个Partition所使用。

image.png

窄依赖分两种：一对一的依赖关系（OneToOneDependency）和范围依赖关系（RangeDependency）。
图中map，filter，join with inputs co-partitioned(对输入进行协同划分的join操作，典型的reduceByKey，先按照key分组然后shuffle write的时候一个父分区对应一个子分区)属于一对一的依赖关系；union属于范围依赖关系。

窄依赖基类

其中getParents方法定义了如何获取父RDD对应的partitionId

/**
 * :: DeveloperApi ::
 * Base class for dependencies where each partition of the child RDD depends on a small number
 * of partitions of the parent RDD. Narrow dependencies allow for pipelined execution.
 */
@DeveloperApi
abstract class NarrowDependency[T](_rdd: RDD[T]) extends Dependency[T] {
  /**
   * 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]

  override def rdd: RDD[T] = _rdd
}

一对一的依赖 org.apache.spark.OneToOneDependency

直译就是一对一的依赖关系。在这种关系下，父RDD的partitionId和子RDD的partitionId相等。

/**
 * :: DeveloperApi ::
 * Represents a one-to-one dependency between partitions of the parent and child RDDs.
 */
@DeveloperApi
class OneToOneDependency[T](rdd: RDD[T]) extends NarrowDependency[T](rdd) {
  override def getParents(partitionId: Int): List[Int] = List(partitionId)
}

范围依赖关系 org.apache.spark.RangeDependency

有范围的依赖关系是指父RDD的partitionId和子RDD的partitionId在一定范围区间内一一对应。

/**
 * :: 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
    }
  }
}

算法图解

image.png

宽依赖 org.apache.spark.ShuffleDependency

宽依赖（Wide Dependency | Shuffle Dependency）：从英文名可以看出，宽依赖是存在shuffle的。换言之，一个父RDD的partition会被多个子RDD的partition所使用。

image.png

源码解读

/**
 * :: DeveloperApi ::
 * Represents a dependency on the output of a shuffle stage. Note that in the case of shuffle,
 * the RDD is transient since we don't need it on the executor side.
 *
 * @param _rdd the parent RDD
 * @param partitioner partitioner used to partition the shuffle output
 * @param serializer [[org.apache.spark.serializer.Serializer Serializer]] to use. If not set
 *                   explicitly then the default serializer, as specified by `spark.serializer`
 *                   config option, will be used.
 * @param keyOrdering key ordering for RDD's shuffles
 * @param aggregator map/reduce-side aggregator for RDD's shuffle
 * @param mapSideCombine whether to perform partial aggregation (also known as map-side combine)
 * @param shuffleWriterProcessor the processor to control the write behavior in ShuffleMapTask
 */
@DeveloperApi
class ShuffleDependency[K: ClassTag, V: ClassTag, C: ClassTag](
    @transient private val _rdd: RDD[_ <: Product2[K, V]],
    val partitioner: Partitioner,
    val serializer: Serializer = SparkEnv.get.serializer,
    val keyOrdering: Option[Ordering[K]] = None,
    val aggregator: Option[Aggregator[K, V, C]] = None,
    val mapSideCombine: Boolean = false,
    val shuffleWriterProcessor: ShuffleWriteProcessor = new ShuffleWriteProcessor)
  extends Dependency[Product2[K, V]] {

  if (mapSideCombine) {
    require(aggregator.isDefined, "Map-side combine without Aggregator specified!")
  }
  override def rdd: RDD[Product2[K, V]] = _rdd.asInstanceOf[RDD[Product2[K, V]]]

  private[spark] val keyClassName: String = reflect.classTag[K].runtimeClass.getName
  private[spark] val valueClassName: String = reflect.classTag[V].runtimeClass.getName
  // Note: It's possible that the combiner class tag is null, if the combineByKey
  // methods in PairRDDFunctions are used instead of combineByKeyWithClassTag.
  private[spark] val combinerClassName: Option[String] =
    Option(reflect.classTag[C]).map(_.runtimeClass.getName)

  val shuffleId: Int = _rdd.context.newShuffleId()

  val shuffleHandle: ShuffleHandle = _rdd.context.env.shuffleManager.registerShuffle(
    shuffleId, _rdd.partitions.length, this)

  _rdd.sparkContext.cleaner.foreach(_.registerShuffleForCleanup(this))
}

我们看到shuffle依赖会生成对应的shuffleId作为标示，另外，还会在shuffleManager中进行注册。

RDD依赖测试

测试方案：前面我们说groupByKey会产生shuffle生成宽依赖，所以groupByKey返回的rdd的依赖类型应该是ShuffleDependency。
测试代码

val rdd = sc.parallelize( 1 to 10, 5)
// map产生一对一窄依赖
val mapRdd = rdd.map(key => (key, 1))
// 窄依赖类
val narrowDependency = mapRdd.dependencies.head
// groupByKey产生宽依赖
val groupRdd = mapRdd.groupByKey
// 宽依赖类
val dependency = groupRdd.dependencies.head
val rdd2 = sc.parallelize( 10 to 100, 10)
// union产生范围窄依赖
val unionRdd = rdd.union(rdd2)
// 范围窄依赖
val rangeDependence = unionRdd.dependencies.head

测试结果符合预期

[hadoop@jms-master-01 ~]$ spark-shell
2019-03-19 23:46:25 WARN  NativeCodeLoader:62 - Unable to load native-hadoop library for your platform... using builtin-java classes where applicable
Setting default log level to "WARN".
To adjust logging level use sc.setLogLevel(newLevel). For SparkR, use setLogLevel(newLevel).
Spark context Web UI available at http://jms-master-01:4040
Spark context available as 'sc' (master = local[*], app id = local-1553010403826).
Spark session available as 'spark'.
Welcome to
      ____              __
     / __/__  ___ _____/ /__
    _\ \/ _ \/ _ `/ __/  '_/
   /___/ .__/\_,_/_/ /_/\_\   version 2.4.0
      /_/

Using Scala version 2.11.12 (Java HotSpot(TM) 64-Bit Server VM, Java 1.8.0_191)
Type in expressions to have them evaluated.
Type :help for more information.

scala> val rdd = sc.parallelize( 1 to 10, 5)
rdd: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[0] at parallelize at <console>:24

scala> val mapRdd = rdd.map(key => (key, 1))
mapRdd: org.apache.spark.rdd.RDD[(Int, Int)] = MapPartitionsRDD[1] at map at <console>:25

scala> val narrowDependency = mapRdd.dependencies.head
narrowDependency: org.apache.spark.Dependency[_] = org.apache.spark.OneToOneDependency@55882ff2

scala> val groupRdd = mapRdd.groupByKey
groupRdd: org.apache.spark.rdd.RDD[(Int, Iterable[Int])] = ShuffledRDD[2] at groupByKey at <console>:25

scala> val dependency = groupRdd.dependencies.head
dependency: org.apache.spark.Dependency[_] = org.apache.spark.ShuffleDependency@219aab91

scala> val rdd2 = sc.parallelize( 10 to 100, 10)
rdd2: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[3] at parallelize at <console>:24

scala> val unionRdd = rdd.union(rdd2)
unionRdd: org.apache.spark.rdd.RDD[Int] = UnionRDD[4] at union at <console>:27

scala> val rangeDependence = unionRdd.dependencies.head
rangeDependence: org.apache.spark.Dependency[_] = org.apache.spark.RangeDependency@5afefa87

scala>