spark RDD的依賴關(guān)系是什么

今天就跟大家聊聊有關(guān)spark RDD的依賴關(guān)系是什么，可能很多人都不太了解，為了讓大家更加了解，小編給大家總結(jié)了以下內(nèi)容，希望大家根據(jù)這篇文章可以有所收獲。

依賴關(guān)系

基本概念

RDD的依賴關(guān)系有一種類似于上下文之間的聯(lián)系，這種關(guān)系也是存在于各個RDD算子間的，相鄰兩個RDD間的關(guān)系被稱作依賴關(guān)系，多個連續(xù)的RDD之間的關(guān)系，被稱作血緣關(guān)系。
每個RDD都會保存血緣關(guān)系，就像是知道自己的父親是誰，自己的父親的父親是誰一樣。

RDD不會保存數(shù)據(jù)，因此當一個算子出錯的時候，為了能夠提高容錯性，需要通過算子間的依賴關(guān)系找到數(shù)據(jù)源頭，再按順序執(zhí)行，從而重新讀取計算。

def main(args: Array[String]): Unit = {
    val sparConf = new SparkConf().setMaster("local").setAppName("WordCount")
    val sc = new SparkContext(sparConf)

    val lines: RDD[String] = sc.makeRDD(List("hello world","hello spark"))
    println(lines.toDebugString)
    println("*************************")
    val words: RDD[String] = lines.flatMap(_.split(" "))
    println(words.toDebugString)
    println("*************************")
    val wordToOne = words.map(word=>(word,1))
    println(wordToOne.toDebugString)
    println("*************************")
    val wordToSum: RDD[(String, Int)] = wordToOne.reduceByKey(_+_)
    println(wordToSum.toDebugString)
    println("*************************")
    val array: Array[(String, Int)] = wordToSum.collect()
    array.foreach(println)
    sc.stop()
  }

輸出的血緣關(guān)系日志如下：

(1) ParallelCollectionRDD[0] at makeRDD at RDD_Dependence_01.scala:13 []
*************************
(1) MapPartitionsRDD[1] at flatMap at RDD_Dependence_01.scala:16 []
 |  ParallelCollectionRDD[0] at makeRDD at RDD_Dependence_01.scala:13 []
*************************
(1) MapPartitionsRDD[2] at map at RDD_Dependence_01.scala:19 []
 |  MapPartitionsRDD[1] at flatMap at RDD_Dependence_01.scala:16 []
 |  ParallelCollectionRDD[0] at makeRDD at RDD_Dependence_01.scala:13 []
*************************
(1) ShuffledRDD[3] at reduceByKey at RDD_Dependence_01.scala:22 []
 +-(1) MapPartitionsRDD[2] at map at RDD_Dependence_01.scala:19 []
    |  MapPartitionsRDD[1] at flatMap at RDD_Dependence_01.scala:16 []
    |  ParallelCollectionRDD[0] at makeRDD at RDD_Dependence_01.scala:13 []
*************************

寬依賴和窄依賴

窄依賴

窄依賴指的是父RDD的分區(qū)數(shù)據(jù)只提供給一個對應(yīng)的子RDD的分區(qū)

寬依賴

寬依賴指的是父RDD的分區(qū)數(shù)據(jù)提供給多個對應(yīng)的子RDD的分區(qū)，當父RDD有Shuffle操作的時候，父RDD與子RDD的依賴關(guān)系必定是寬依賴，因此其也被稱為Shuffle依賴。

階段劃分

DAG（Directed Acyclic Graph）有向無環(huán)圖是由點和線組成的拓撲圖形，該圖形具有方向， 不會閉環(huán)。例如，DAG 記錄了 RDD 的轉(zhuǎn)換過程和任務(wù)的階段。

DAGScheduler部分源碼解釋了任務(wù)的階段劃分過程：

1. 在handleJobSubmitted方法有一個傳入?yún)?shù)為finalRDD，通過 finalStage = createResultStage(finalRDD, func, partitions, jobId, callSite) 方法，可以看出無論有多少個RDD，都會默認通過最終的RDD去創(chuàng)建一個resultStage。

2. 之后createResultStage調(diào)用了getOrCreateParentStages(rdd: RDD[_], firstJobId: Int): List[Stage]方法，通過 getShuffleDependencies( rdd: RDD[_]) 返回依賴關(guān)系的鏈式結(jié)構(gòu)（ShuffleDependency的存儲map），如： A <-- B <-- C

3. 遍歷ShuffleDependency的存儲map，通過getOrCreateShuffleMapStage(shuffleDep, firstJobId) 去創(chuàng)建階段，這里通過firstJobId去做關(guān)聯(lián)，緩存的stage在shuffleIdToMapStage中。

  /**
   * Create a ResultStage associated with the provided jobId.
   */
  private def createResultStage(
      rdd: RDD[_],
      func: (TaskContext, Iterator[_]) => _,
      partitions: Array[Int],
      jobId: Int,
      callSite: CallSite): ResultStage = {
    checkBarrierStageWithDynamicAllocation(rdd)
    checkBarrierStageWithNumSlots(rdd)
    checkBarrierStageWithRDDChainPattern(rdd, partitions.toSet.size)
    val parents = getOrCreateParentStages(rdd, jobId) //這里調(diào)用
    val id = nextStageId.getAndIncrement()
    val stage = new ResultStage(id, rdd, func, partitions, parents, jobId, callSite)
    stageIdToStage(id) = stage
    updateJobIdStageIdMaps(jobId, stage)
    stage
  }

  /**
   * Get or create the list of parent stages for a given RDD.  The new Stages will be created with
   * the provided firstJobId.
   */
  private def getOrCreateParentStages(rdd: RDD[_], firstJobId: Int): List[Stage] = {
    getShuffleDependencies(rdd).map { shuffleDep =>
      getOrCreateShuffleMapStage(shuffleDep, firstJobId)
    }.toList
  }
  
    /**
   * Returns shuffle dependencies that are immediate parents of the given RDD.
   *
   * This function will not return more distant ancestors.  For example, if C has a shuffle
   * dependency on B which has a shuffle dependency on A:
   *
   * A <-- B <-- C
   *
   * calling this function with rdd C will only return the B <-- C dependency.
   *
   * This function is scheduler-visible for the purpose of unit testing.
   */
  private[scheduler] def getShuffleDependencies(
      rdd: RDD[_]): HashSet[ShuffleDependency[_, _, _]] = {
    val parents = new HashSet[ShuffleDependency[_, _, _]]
    val visited = new HashSet[RDD[_]]
    val waitingForVisit = new ListBuffer[RDD[_]]
    waitingForVisit += rdd
    while (waitingForVisit.nonEmpty) {
      val toVisit = waitingForVisit.remove(0)
      if (!visited(toVisit)) {
        visited += toVisit
        toVisit.dependencies.foreach {
          case shuffleDep: ShuffleDependency[_, _, _] =>
            parents += shuffleDep
          case dependency =>
            waitingForVisit.prepend(dependency.rdd)
        }
      }
    }
    parents
  }

任務(wù)劃分

RDD 任務(wù)切分為：Application、Job、Stage 和 Task

• Application：初始化一個 SparkContext 即生成一個 Application；

• Job：一個 Action 算子就會生成一個 Job；

• Stage：Stage 等于寬依賴(ShuffleDependency)的個數(shù)加 1；

• Task：一個 Stage 階段中，最后一個 RDD 的分區(qū)個數(shù)就是 Task 的個數(shù)。

注意：Application->Job->Stage->Task 每一層都是 1 對 n 的關(guān)系。

看完上述內(nèi)容，你們對spark RDD的依賴關(guān)系是什么有進一步的了解嗎？如果還想了解更多知識或者相關(guān)內(nèi)容，請關(guān)注億速云行業(yè)資訊頻道，感謝大家的支持。