專業(yè)術(shù)語(yǔ)：

窄依賴和寬依賴

RDD之間有一系列的依賴關(guān)系，依賴關(guān)系又分為窄依賴和寬依賴。

• 窄依賴

父RDD和子RDD partition之間的關(guān)系是一對(duì)一的。或者父RDD一個(gè)partition只對(duì)應(yīng)一個(gè)子RDD的partition情況下的父RDD和子RDD partition關(guān)系是多對(duì)一的。不會(huì)有shuffle的產(chǎn)生。

• 寬依賴

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

寬窄依賴圖理解

Stage

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

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

• stage切割規(guī)則

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

?

• stage計(jì)算模式

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

?

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

對(duì)RDD進(jìn)行持久化。

shuffle write的時(shí)候。

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

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

• 測(cè)試驗(yàn)證pipeline計(jì)算模式

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}/*** 驗(yàn)證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()} }

?

總結(jié)

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

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