【原創(chuàng)】Spark Standalong模式運(yùn)行原理剖析

　　一、概述

　　Apache Spark是一種快速和通用的集群計(jì)算系統(tǒng)。它提供Java，Scala，Python和R中的高級(jí)API，以及支持一般執(zhí)行圖的優(yōu)化引擎。它還支持一組豐富的更高級(jí)別的工具，包括Spark SQL用于SQL和結(jié)構(gòu)化數(shù)據(jù)的處理，MLlib機(jī)器學(xué)習(xí)，GraphX用于圖形處理和Spark Streaming。

　　Spark除了在Mesos或YARN群集管理器上運(yùn)行，它還提供了一種簡(jiǎn)單的獨(dú)立部署模式Standalone模式。接下來(lái)我們就以下面的WordCount代碼為例剖析Spark Standalone模式的運(yùn)行原理。理解并掌握Spark Standalone模式的運(yùn)行原理對(duì)后期進(jìn)一步學(xué)習(xí)Spark相關(guān)技術(shù)有很大的幫助，同時(shí)也是Spark開(kāi)發(fā)工程師崗位面試經(jīng)常被提問(wèn)的地方。

　　WordCount代碼如下：

　　Standalone運(yùn)行模式原理概要如下圖所示：

　　二、Standalong模式運(yùn)式原理剖析之天龍八“步”

　　1、第一步：

　　通過(guò)spark-submit指令將打好的Spark jar包提交到Spark集群中運(yùn)行。先從Driver進(jìn)程開(kāi)始運(yùn)行，Driver中包含了我們所編寫(xiě)的代碼。

　　首先執(zhí)行代碼中的前兩行代碼，

　　//創(chuàng)建SparkConf對(duì)象　　

　　val conf = new SparkConf().setAppName("WordCount")　　

　　//創(chuàng)建SparkContext對(duì)象　　

　　val sc = new SparkContext(conf)

　　這兩行代碼分別創(chuàng)建了SparkConf和SparkContext對(duì)象，在創(chuàng)建SparkContext對(duì)象的過(guò)程中，Spark會(huì)去做兩件很重要的事，就是創(chuàng)建DAGScheduler和TaskScheduler這兩個(gè)對(duì)象。然后，TaskScheduler會(huì)通過(guò)一個(gè)后臺(tái)進(jìn)程負(fù)責(zé)與Master進(jìn)行注冊(cè)通信，告訴Master有一個(gè)新的Application應(yīng)用程序要運(yùn)行，需要Master管理分配調(diào)度集群的資源。

　　2、第二步：

　　Master接收到TaskScheduler的注冊(cè)請(qǐng)求之后，會(huì)通過(guò)資源調(diào)度算法對(duì)集群資源進(jìn)行調(diào)度，并且與Worker進(jìn)行通信，請(qǐng)求Worker啟動(dòng)相應(yīng)的Executor。

　　3、第三步：

　　Worker接收到Master的請(qǐng)求之后，會(huì)在本節(jié)點(diǎn)中啟動(dòng)Executor。因?yàn)榧褐杏卸鄠€(gè)Worker節(jié)點(diǎn)，那么也意味著會(huì)啟動(dòng)多個(gè)Executor。一個(gè)Application對(duì)應(yīng)著Worker中的一個(gè)Executor。

　　4、第四步：

　　Executor啟動(dòng)完成之后，會(huì)向Driver中的TaskScheduler進(jìn)行反注冊(cè)，反注冊(cè)的目的就是讓Driver知道新提交的Application應(yīng)用將由哪些Executor負(fù)責(zé)執(zhí)行。

　　5、第五步：

　　Executor向Driver中的TaskScheduler反注冊(cè)完成之后，就意味著SparkContext的初始化過(guò)程已經(jīng)完成，接下來(lái)去執(zhí)行SparkContext下面的代碼。

　　//從linux或者HDFS中獲取數(shù)據(jù)

　　val lines = sc.textFile("hdfs://tgmaster:9000/in/resws")

　　//進(jìn)行單詞統(tǒng)計(jì)計(jì)數(shù)

　　val result = lines.flatMap(_.split(" ")).map((_, 1))

　　//將計(jì)算結(jié)果保存到HDFS中

　　result.saveAsTextFile("hdfs://tgmaster:9000/out/res3")

　　sc.stop()

　　6、第六步：

　　在SparkContext下面的代碼中，創(chuàng)建了初始RDD，并對(duì)初始RDD進(jìn)行了Transformation類型的算子操作，但是系統(tǒng)只是記錄下了這些操作行為，這些操作行并沒(méi)有真正的被執(zhí)行，直到遇到Action類型的算子，觸發(fā)提交job之后，Action類型的算子之前所有的Transformation類型的算子才會(huì)被執(zhí)行。job會(huì)被提交給DAGScheduler，DAGScheduler根據(jù)stage劃分算法將job劃分為多個(gè)stage(階段),并將其封裝成TaskSet(任務(wù)集合)，然后將TaskSet提交給TaskScheduler。

　　7、第七步：

　　TaskScheduler根據(jù)task分配算法，將TaskSet中的每一個(gè)小task分配給Executor去執(zhí)行。

　　8、第八步：

　　Executor接受到task任務(wù)之后，通過(guò)taskrunner來(lái)封裝一個(gè)task,并從線程池中取出相應(yīng)的一個(gè)線程來(lái)執(zhí)行task。

　　task線程針對(duì)RDD partition分區(qū)中的數(shù)據(jù)進(jìn)行指定的算子操作，這些算子操作包括Transformation和Action類型的操作。

　　補(bǔ)充說(shuō)明：

　　1、taskrunner(任務(wù)運(yùn)行器),會(huì)對(duì)我們編寫(xiě)代碼進(jìn)行復(fù)制、反序列化操作，進(jìn)行執(zhí)行task任務(wù)。

　　2、task分為兩大類：ShuffleMapTask和ResultTask。最后一個(gè)stage階段中的task稱為ResultTask，在這之前所有的Task稱為ShuffleMapTask。