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

創(chuàng)新互聯(lián)公司是一家集網(wǎng)站建設(shè),湘西土家族企業(yè)網(wǎng)站建設(shè),湘西土家族品牌網(wǎng)站建設(shè),網(wǎng)站定制,湘西土家族網(wǎng)站建設(shè)報(bào)價(jià),網(wǎng)絡(luò)營(yíng)銷,網(wǎng)絡(luò)優(yōu)化,湘西土家族網(wǎng)站推廣為一體的創(chuàng)新建站企業(yè)，幫助傳統(tǒng)企業(yè)提升企業(yè)形象加強(qiáng)企業(yè)競(jìng)爭(zhēng)力?？沙浞譂M足這一群體相比中小企業(yè)更為豐富、高端、多元的互聯(lián)網(wǎng)需求。同時(shí)我們時(shí)刻保持專業(yè)、時(shí)尚、前沿，時(shí)刻以成就客戶成長(zhǎng)自我，堅(jiān)持不斷學(xué)習(xí)、思考、沉淀、凈化自己，讓我們?yōu)楦嗟钠髽I(yè)打造出實(shí)用型網(wǎng)站。

依賴關(guān)系

基本概念

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

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

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ù)只提供給一個(gè)對(duì)應(yīng)的子RDD的分區(qū)

寬依賴

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

階段劃分

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

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

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

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

3. 遍歷ShuffleDependency的存儲(chǔ)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：初始化一個(gè) SparkContext 即生成一個(gè) Application；

• Job：一個(gè) Action 算子就會(huì)生成一個(gè) Job；

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

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

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

看完上述內(nèi)容，你們對(duì)spark RDD的依賴關(guān)系是什么有進(jìn)一步的了解嗎？如果還想了解更多知識(shí)或者相關(guān)內(nèi)容，請(qǐng)關(guān)注創(chuàng)新互聯(lián)行業(yè)資訊頻道，感謝大家的支持。

當(dāng)前標(biāo)題：sparkRDD的依賴關(guān)系是什么
網(wǎng)站URL：http://jinyejixie.com/article10/jjpggo.html

成都網(wǎng)站建設(shè)公司_創(chuàng)新互聯(lián)，為您提供營(yíng)銷型網(wǎng)站建設(shè)、網(wǎng)站制作、品牌網(wǎng)站設(shè)計(jì)、靜態(tài)網(wǎng)站、微信公眾號(hào)、標(biāo)簽優(yōu)化

廣告

聲明：本網(wǎng)站發(fā)布的內(nèi)容（圖片、視頻和文字）以用戶投稿、用戶轉(zhuǎn)載內(nèi)容為主，如果涉及侵權(quán)請(qǐng)盡快告知，我們將會(huì)在第一時(shí)間刪除。文章觀點(diǎn)不代表本網(wǎng)站立場(chǎng)，如需處理請(qǐng)聯(lián)系客服。電話：028-86922220；郵箱：631063699@qq.com。內(nèi)容未經(jīng)允許不得轉(zhuǎn)載，或轉(zhuǎn)載時(shí)需注明來(lái)源： 創(chuàng)新互聯(lián)