【Spark篇】--Spark中的寬窄依賴和Stage的划分

一、前述

RDD之間有一系列的依賴關系，依賴關系又分為窄依賴和寬依賴。

Spark中的Stage其實就是一組並行的任務，任務是一個個的task 。

二、具體細節

 

• 窄依賴

父RDD和子RDD partition之間的關系是一對一的。或者父RDD一個partition只對應一個子RDD的partition情況下的父RDD和子RDD partition關系是多對一的。不會有shuffle的產生。父RDD的一個分區去到子RDD的一個分區。

 

• 寬依賴

父RDD與子RDD partition之間的關系是一對多。會有shuffle的產生。父RDD的一個分區的數據去到子RDD的不同分區里面。

其實區分寬窄依賴主要就是看父RDD的一個Partition的流向，要是流向一個的話就是窄依賴，流向多個的話就是寬依賴。看圖理解：

 

• Stage概念

Spark任務會根據RDD之間的依賴關系，形成一個DAG有向無環圖，DAG會提交給DAGScheduler，DAGScheduler會把DAG划分相互依賴的多個stage，划分stage的依據就是RDD之間的寬窄依賴。遇到寬依賴就划分stage,每個stage包含一個或多個task任務。然后將這些task以taskSet的形式提交給TaskScheduler運行。     stage是由一組並行的task組成。

 

• stage切割規則

 切割規則：從后往前，遇到寬依賴就切割stage。

 

 

 

• stage計算模式

    pipeline管道計算模式,pipeline只是一種計算思想，模式。

 

備注：圖中幾個理解點：

   1、Spark的pipeLine的計算模式，相當於執行了一個高階函數f3(f2(f1(textFile))) !+!+!=3 也就是來一條數據然后計算一條數據，把所有的邏輯走完，然后落地，准確的說一個task處理遺傳分區的數據 因為跨過了不同的邏輯的分區。而MapReduce是 1+1=2,2+1=3的模式，也就是計算完落地，然后在計算，然后再落地到磁盤或內存，最后數據是落在計算節點上，按reduce的hash分區落地。所以這也是比Mapreduce快的原因，完全基於內存計算。

   2、管道中的數據何時落地：shuffle write的時候，對RDD進行持久化的時候。

   3.   Stage的task並行度是由stage的最后一個RDD的分區數來決定的 。一般來說，一個partiotion對應一個task,但最后reduce的時候可以手動改變reduce的個數，也就是分區數，即改變了並行度。例如reduceByKey(XXX,3),GroupByKey(4)，union由的分區數由前面的相加。

   4.、如何提高stage的並行度：reduceBykey(xxx,numpartiotion),join(xxx,numpartiotion)

 

• 測試驗證pipeline計算模式

  import org.apache.spark.SparkConf
  import org.apache.spark.SparkContext
  import java.util.Arrays
  
  object PipelineTest {
    def main(args: Array[String]): Unit = {
      val conf = new SparkConf()
      conf.setMaster("local").setAppName("pipeline");
      val sc = new SparkContext(conf)
      val rdd = sc.parallelize(Array(1,2,3,4))
      val rdd1 = rdd.map { x => {
        println("map--------"+x)
        x
      }}
      val rdd2 = rdd1.filter { x => {
        println("fliter********"+x)
        true
      } }
      rdd2.collect()
      sc.stop()
    }
  }
  

   

  可見是按照所有的邏輯將數據一條條的執行。！！！