WordCount的代碼

主要是從HDFS讀取文件后進(jìn)行單詞切割,然后進(jìn)行計(jì)數(shù),如果不懂RDD算子可以看RDD詳解

WordCount的各個(gè)算子

SparkRDD的運(yùn)行流程

SparkRDD寬依賴和窄依賴

SparkRDD之間的依賴主要有:

1.寬依賴

寬依賴指的是多個(gè)子RDD的Partition會(huì)依賴同一個(gè)父RDD的Partition

總結(jié)：窄依賴我們形象的比喻為超生

2.窄依賴

窄依賴指的是每一個(gè)父RDD的Partition最多被子RDD的一個(gè)Partition使用

總結(jié)：窄依賴我們形象的比喻為獨(dú)生子女

結(jié)合WordCount的源碼分析

WordCount算子內(nèi)部解析

在WordCount程序中,第一個(gè)使用的Spark方法是textFile()方法,主要的源碼是

這個(gè)方法的主要作用是從HDFS中讀取數(shù)據(jù), 這里創(chuàng)建一個(gè)HadoopRDD,在這個(gè)方法內(nèi)部還創(chuàng)建一個(gè)MapPartitionRDD,接下里的幾個(gè) RDD同樣是MapPartitionRDD,最主要的是看saveAsTextFile()方法。 下面是saveAsTextFile()方法,代碼在RDD類的1272行,具體內(nèi)容如下:

這個(gè)方法的主要作用是產(chǎn)生一個(gè)RDD,MapPartitionsRDD;然后將RDD轉(zhuǎn)化為PairRDDFuctions,接下來是saveAsHadoopFile()方法: 主要的代碼如下:

繼續(xù)查看saveAsHadoopDataset()方法源碼,主要的代碼如下:

代碼解析:

1.獲取寫入HDFS中的文件流

2.一個(gè)函數(shù)將分區(qū)數(shù)據(jù)迭代的寫入到HDFS中

3.開始提交作業(yè),Self表示Final RDD也就是作業(yè)最后的RDD在WordCount中也就是MapPartitionsRDD

這里我們將會(huì)追蹤到runJob()方法中,

這里我們繼續(xù)追蹤到runJob()的重載方法,夏滿是這個(gè)方法的核心代碼:

這里是非常重要的方法,主要做的工作是調(diào)用SparkContext類中創(chuàng)建的dagScheduler,使用dagScheduler劃分Stage,然后將Stage轉(zhuǎn)化為TaskSet交給TaskScheduler在交個(gè)Executor執(zhí)行

劃分Stage

在前面的分析中,我們已經(jīng)知道了dagScheduler調(diào)用了runJob()方法,這個(gè)方法的作用是劃分stage。

這里主要是劃分stage,然后調(diào)用submitJob()返回一個(gè)調(diào)度器,這里我們繼續(xù)查看submitJob()方法。

上面是submitJob()方法的核心代碼,主要的作用是eventProcessLoop對(duì)象內(nèi)部有一個(gè)阻塞隊(duì)列和線程，先將數(shù)據(jù)封裝到Case Class中將事件放入到阻塞隊(duì)列。

對(duì)于JobSubmitted類的模式匹配,主要的代碼如下:

這里調(diào)用dagScheduler的handleJobSubmitted()方法,這個(gè)方法是對(duì)stage劃分的主要方法,主要的核心代碼:

通過newStage()方法,根據(jù)這個(gè)方法在這里可以看出分區(qū)的數(shù)量決定Task數(shù)量。 通過追蹤newStage()方法,主要的代碼如下:

這個(gè)方法是遞歸的劃分Stage,主要的方法是getParentStages(rdd, jobId),具體的劃分代碼如下:

stage劃分算法如下:

涉及的數(shù)據(jù)結(jié)構(gòu):棧、HashSet

1.通過最后的RDD,獲取父RDD

2.將finalRDD放入棧中,然后出棧,進(jìn)行for循環(huán)的找到RDD的依賴,需要注意的是RDD可能有多個(gè)依賴

3.如果RDD依賴是ShuffleDependency,那么就可以劃分成為一個(gè)新的Stage,然后通過getShuffleMapStage()獲取這個(gè)stage的父stage;如果是一般的窄依賴,那么將會(huì)入棧

4.通過getShuffleMapStage()遞歸調(diào)用,得到父stage;一直到父stage是null

5.最后返回stage的集合

stage提交算法

在對(duì)于最后一個(gè)RDD劃stage后,進(jìn)行提交stage,主要的方法是:

這里和劃分stage的算法一樣,拿到最后的stage然后找到第一個(gè)stage開始從第一個(gè)stage開始提交。

stage提交

下面的代碼是submitMissingTasks(),主要是核心的代碼:

這里主要做的工作是根據(jù)分區(qū)數(shù)量決定Task數(shù)量,然后根據(jù)stage的類型創(chuàng)建Task,這里主要有ShuffleMapTask和ResultTask。

ShuffleMapTask:進(jìn)行分區(qū)局部聚合,從上游拉去數(shù)據(jù)。

ResultTask:將結(jié)果寫入持久化介質(zhì).比如HDFS等。

這里將Task進(jìn)行封裝成為TaskSet進(jìn)行提交給taskScheduler。

關(guān)于Stage劃分流程圖

總結(jié)

1.textFile()方法會(huì)產(chǎn)生兩個(gè)RDD,HadoopRDD和MapPartitionRDD

2.saveTextAsFile()方法會(huì)產(chǎn)生一個(gè)RDD,MapPartitionRDD

3.Task數(shù)量取決于HDFS分區(qū)數(shù)量

4.Stage劃分是通過最后的RDD,也就是final RDD根據(jù)依賴關(guān)系進(jìn)行遞歸劃分

5.stage提交主要是通過遞歸算法,根據(jù)最后一個(gè)Stage劃分然后遞歸找到第一個(gè)stage開始從第一個(gè)stage開始提交。

喜歡小編的文章可以關(guān)注喲!