專業(yè)術語：

窄依賴和寬依賴

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

• 窄依賴

父RDD和子RDD partition之間的關系是一對一的。或者父RDD一個partition只對應一個子RDD的partition情況下的父RDD和子RDD partition關系是多對一的。不會有shuffle的產(chǎn)生。

• 寬依賴

父RDD與子RDD partition之間的關系是一對多。會有shuffle的產(chǎn)生。

寬窄依賴圖理解

Stage

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

?????stage是由一組并行的task組成。

• stage切割規(guī)則

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

?

• stage計算模式

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

?

• 數(shù)據(jù)一直在管道里面什么時候數(shù)據(jù)會落地？

對RDD進行持久化。

shuffle write的時候。

• Stage的task并行度是由stage的最后一個RDD的分區(qū)數(shù)來決定的 。
• 如何改變RDD的分區(qū)數(shù)？

例如：reduceByKey(XXX,3),GroupByKey(4)

• 測試驗證pipeline計算模式

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() package SparkRadioimport org.apache.spark.{SparkConf, SparkContext}/*** 驗證pipeline*/ object PipelineDemo {def main(args: Array[String]): Unit = {val conf = new SparkConf().setMaster("local").setAppName("pip")val sc = new SparkContext(conf)sc.setLogLevel("error")val rdd1 = sc.textFile("./data/words.txt")val rdd2 = rdd1.map(x => {println("*** map ***" + x)x + "~"})val rdd3 = rdd2.filter(one => {println("=== filter ===" + one)true})rdd3.count()} }

?

總結

以上是生活随笔為你收集整理的Spark _08窄依赖和宽依赖stage的全部內(nèi)容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網(wǎng)站內(nèi)容還不錯，歡迎將生活随笔推薦給好友。