RDD 的成員之一是依賴(lài)集桶蛔，依賴(lài)集也關(guān)系到任務(wù)調(diào)度

源碼

Dependency代碼主要在一個(gè)源文件中：core/Dependency代碼中有5個(gè)類(lèi)霜运。
除此以外在 core/rdd/PartitionPruningRDD還有一個(gè)PruneDependency類(lèi)。
他們的名稱(chēng)和繼承關(guān)系如下圖：

3017B56B-EBD0-4E3A-BAD8-777B872194D4.png

通過(guò)閱讀代碼可以得到以下信息：

1. 依賴(lài)的根類(lèi)是Dependency，只有一個(gè)RDD 成員，表示依賴(lài)的對(duì)象。這類(lèi)繼承了Serializable類(lèi)少办，是可以序列化的。
2. 依賴(lài)分為兩大種诵原，一個(gè)叫窄依賴(lài)(Narrow)英妓，另一個(gè)就洗牌依賴(lài)(Shuffle，很多材料也叫作“寬”依賴(lài))绍赛。
3. 從數(shù)量關(guān)系上說(shuō)蔓纠，“1-to-1”(OneToOne)、“n-to-1”(Range)吗蚌、“1-to-部分分區(qū)”(Prune腿倚，剪枝)是窄依賴(lài)，寬依賴(lài)是“n-to-n”的蚯妇。
4. “1-to-1”是RDD的默認(rèn)依賴(lài)敷燎。上節(jié)中的MapPartitionRDD是一對(duì)一的轉(zhuǎn)換暂筝，就包含“1-to-1“依賴(lài)。
5. ”n-to-1“的依賴(lài)只有一個(gè)使用場(chǎng)景——UnionRDD硬贯，“交”運(yùn)算焕襟，多個(gè) RDD 合并到一個(gè) RDD 中。
6. 剪枝依賴(lài)是個(gè)私有對(duì)象饭豹，用于優(yōu)化鸵赖，減少數(shù)據(jù)載入。
7. 洗牌依賴(lài)復(fù)雜一些拄衰。

只有 RDD 的轉(zhuǎn)換（Transformations）才用到依賴(lài)它褪，RDD 的操作（Actions，如reduce翘悉、collect等）就不需要依賴(lài)列赎，直接運(yùn)行SparkContext.runjob()函數(shù)。RDD 有哪些轉(zhuǎn)換和操作镐确？閱讀文檔： Spark Programming Guide - Spark 2.0.2 Documentation

洗牌依賴(lài)

洗牌依賴(lài)也只是相對(duì)復(fù)雜，代碼也不長(zhǎng)饼煞。

@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)
  extends Dependency[Product2[K, V]] {

  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))
}

是的源葫，只有8行代碼，核心內(nèi)容是：

1. 這個(gè)類(lèi)有三個(gè)泛型類(lèi)型砖瞧，K=key息堂，V=value，C=combiner块促；
2. 洗牌依賴(lài)只能用于Product2[K, V]及其父類(lèi)荣堰，即 KV 數(shù)據(jù)；
3. 成員有 分區(qū)器(partitioner) 竭翠、序列器(serializer)振坚、排序器(keyOrdering)、聚合器(aggregator)斋扰、map 端聚合開(kāi)關(guān)(mapSideCombine)渡八；
4. _rdd.context.newShuffleId() 獲得一個(gè)自增的 ID；
5. _rdd.context.env.shuffleManager.registerShuffle 獲得幾個(gè)洗牌的句柄传货。通過(guò)core/shuffle/sort/SortShuffleManager代碼可以知道屎鳍，一共有三種句柄：
   1. 分區(qū)數(shù)很少（小于變量spark.shuffle.sort.bypassMergeThreshold，默認(rèn)200）時(shí)问裕，用BypassMergeSortShuffleHandle逮壁，直接發(fā)送數(shù)據(jù)合并，不用耗時(shí)的序列化和反序列化粮宛；
   2. 否則窥淆，如果能序列化卖宠，則用SerializedShuffleHandle，用序列化和反序列化祖乳，降低網(wǎng)絡(luò) IO捻艳；
   3. 否則臭增，使用基礎(chǔ)的BaseShuffleHandle。

Scala語(yǔ)法

在 Java 中使用null 是非常容易出錯(cuò)的，在 Guava( GitHub - google/guava: Google Core Libraries for Java 6+ ) 中提供了 Optional 來(lái)避免使用 null槐瑞。
同樣的 Scala 自帶了 Option 來(lái)避免使用 null。

Option 有兩個(gè)子類(lèi)鸳慈，Some 和 None

scala> val capitals = Map("France"->"Paris", "Japan"->"Tokyo", "China"->"Beijing")
capitals: scala.collection.immutable.Map[String,String] = Map(France -> Paris, Japan -> Tokyo, China -> Beijing)

scala> capitals get "France"
res0: Option[String] = Some(Paris)

scala> capitals get "North Pole"
res1: Option[String] = None

None.get 會(huì)報(bào)錯(cuò)拂募，用 getOrElse 可以給取不到值的時(shí)候賦默認(rèn)值：

scala> capitals get "North Pole" get
warning: there was one feature warning; re-run with -feature for details
java.util.NoSuchElementException: None.get
  at scala.None$.get(Option.scala:347)
  at scala.None$.get(Option.scala:345)
  ... 33 elided

scala> capitals get "France" get
warning: there was one feature warning; re-run with -feature for details
res3: String = Paris

scala> (capitals get "North Pole") getOrElse "Oops"
res7: String = Oops

scala> capitals get "France" getOrElse "Oops"
res8: String = Paris

在 Spark 中大量使用了 Option。

疑問(wèn)列表

我將閱讀過(guò)程中的未解內(nèi)容記錄下來(lái)帅刊，留待以后閱讀代碼時(shí)解答纸泡。疑問(wèn)一個(gè)一個(gè)劃掉，就是成長(zhǎng)的過(guò)程赖瞒。

1. reduce 等 RDD 操作是如何執(zhí)行的女揭？
2. groupByKey 等洗牌操作是如何執(zhí)行的？不同的洗牌類(lèi)型有什么用栏饮？

結(jié)論

1. 在 RDD 的轉(zhuǎn)換時(shí)吧兔，會(huì)用到依賴(lài)
2. 依賴(lài)包括窄依賴(lài)（1-to-1、n-to-1關(guān)系）袍嬉、洗牌依賴(lài)（n-to-n 關(guān)系）
3. 洗牌依賴(lài)包含分區(qū)器境蔼、序列器、排序器伺通、聚合器箍土、map聚合開(kāi)關(guān)、ID罐监、洗牌類(lèi)型句柄等成分組成吴藻。洗牌類(lèi)型句柄有三種。

本文源碼

Dependency spark/Dependency.scala at master · apache/spark · GitHub
三種洗牌句柄 spark/SortShuffleManager.scala at master · apache/spark · GitHub
Option示例來(lái)自： 【Scala】使用Option弓柱、Some调缨、None，避免使用null - 簡(jiǎn)書(shū)

@ Kangying Village, Beijing, China

Spark源碼/Dependency