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

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

依賴關系

基本概念

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

RDD不會保存數(shù)據(jù)，因此當一個算子出錯的時候，為了能夠提高容錯性，需要通過算子間的依賴關系找到數(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()
  }

輸出的血緣關系日志如下：

(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ù)只提供給一個對應的子RDD的分區(qū)

寬依賴

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

階段劃分

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

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

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[_]) 返回依賴關系的鏈式結構（ShuffleDependency的存儲map），如： A <-- B <-- C

3. 遍歷ShuffleDependency的存儲map，通過getOrCreateShuffleMapStage(shuffleDep, firstJobId) 去創(chuàng)建階段，這里通過firstJobId去做關聯(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
  }

任務劃分

RDD 任務切分為：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 的關系。

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

當前標題：sparkRDD的依賴關系是什么
網(wǎng)站URL：http://www.yijiale78.com/article10/jjpggo.html

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

廣告

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