剖析spark作業(yè)運(yùn)行機(jī)制

剖析spark作業(yè)運(yùn)行機(jī)制主要內(nèi)容2Spark運(yùn)行機(jī)制Spark運(yùn)行機(jī)制特點(diǎn)3一、spark運(yùn)行機(jī)制4Spark運(yùn)行機(jī)制流程圖5Spark運(yùn)行機(jī)制構(gòu)建SparkApplication的運(yùn)行環(huán)境，啟動(dòng)SparkContextSparkContext向資源管理器（可以是Standalone，Mesos，Yarn）申請(qǐng)運(yùn)行Executor資源，并啟動(dòng)StandaloneExecutorbackend，Executor向SparkContext申請(qǐng)TaskSparkContext將應(yīng)用程序分發(fā)給ExecutorSparkContext構(gòu)建成DAG圖，將DAG圖分解成Stage、將Taskset發(fā)送給TaskScheduler，最后由TaskScheduler將Task發(fā)送給Executor運(yùn)行Task在Executor上運(yùn)行，運(yùn)行完釋放所有資源

6Spark運(yùn)行機(jī)制1）第一步：生成RDD

Spark本身對(duì)RDD的操作模式是惰性計(jì)算。在惰性計(jì)算機(jī)制中，盡管每一次算子操作都會(huì)將一個(gè)RDD轉(zhuǎn)換成一個(gè)新的RDD，并且邏輯上會(huì)順序的執(zhí)行這一系列計(jì)算，但是這些RDD的操作并不是立即執(zhí)行的，而是會(huì)等到出現(xiàn)行動(dòng)算子時(shí)才觸發(fā)整個(gè)RDD操作序列，將之前的所有算子操作形成一個(gè)有向無(wú)環(huán)圖（DirectedAcyclicGraph，DAG），每個(gè)有向無(wú)環(huán)圖再觸發(fā)執(zhí)行一個(gè)作業(yè)（Job）。例如，WordCount程序中，只有程序執(zhí)行到saveAsTextFile時(shí)，Spark才對(duì)RDD進(jìn)行真正的處理，將之前的flatMap、map、reduceByKey和saveAsTextFile這些算子連成一個(gè)有向無(wú)環(huán)圖，并向Spark提交該作業(yè)。

采用惰性計(jì)算的優(yōu)勢(shì)：

1.相關(guān)的操作序列可以進(jìn)行連續(xù)計(jì)算，不用為存儲(chǔ)的中間結(jié)果離散的獨(dú)立分配內(nèi)存空間；

2.節(jié)省存儲(chǔ)空間，為之后對(duì)RDD變換操作的優(yōu)惠提供了條件；

7Spark運(yùn)行機(jī)制（2）第二步：生成Stage

Driver節(jié)點(diǎn)中的DAGSchedule實(shí)例會(huì)對(duì)有向無(wú)環(huán)圖中節(jié)點(diǎn)間的依賴(lài)關(guān)系進(jìn)行遍歷，將所有操作切分為多個(gè)調(diào)度階段（Stage）；

（3）第三步：生成Task

每個(gè)Stage需要轉(zhuǎn)換成任務(wù)在集群中的Worker節(jié)點(diǎn)上執(zhí)行，因此，由Driver節(jié)點(diǎn)中的TaskSchedule實(shí)例將Stage轉(zhuǎn)換為T(mén)ask，并提交到Worker節(jié)點(diǎn)的Executor進(jìn)程中執(zhí)行；8二、RDD特點(diǎn)9RDD特點(diǎn)1.Alistofpartitions2.Afunctionforcomputingeachsplit3.AlistofdependenciesonotherRDDs4.Optionally,aPartitionerforkey-valueRDDs5.Optionally,alistofpreferredlocationstocomputeeachspliton101.Alistofpartitions

RDD由很多partition構(gòu)成，在spark中，計(jì)算式，有多少partition就對(duì)應(yīng)有多少個(gè)task來(lái)執(zhí)行

一組分片（partition），即數(shù)據(jù)集的基本組成單位，對(duì)于RDD來(lái)說(shuō)，每個(gè)分片都會(huì)被一個(gè)計(jì)算任務(wù)處理，并決定并行計(jì)算的粒度。用戶(hù)可以在創(chuàng)建RDD時(shí)指定RDD的分片個(gè)數(shù)，如果沒(méi)有指定，那么就會(huì)采用默認(rèn)值。默認(rèn)值就是程序所分配到的CPUCore的數(shù)目。每個(gè)分配的存儲(chǔ)是由BlockManager實(shí)現(xiàn)的，每個(gè)分區(qū)都會(huì)被邏輯映射成BlockManager的一個(gè)Block，而這個(gè)Block會(huì)被一個(gè)Task負(fù)責(zé)計(jì)算。112.Afunctionforcomputingeachsplit

對(duì)RDD做計(jì)算，相當(dāng)于對(duì)RDD的每個(gè)split或partition做計(jì)算

Spark中的RDD的計(jì)算是以分片為單位的，每個(gè)RDD都會(huì)實(shí)現(xiàn)compute函數(shù)以達(dá)到這個(gè)目的。compute函數(shù)會(huì)對(duì)迭代器進(jìn)行復(fù)合，不需要保存每次計(jì)算的結(jié)果123.AlistofdependenciesonotherRDDs

RDD之間有依賴(lài)關(guān)系，可溯源

RDD的每次轉(zhuǎn)換都會(huì)生成一個(gè)新的RDD，所以RDD之間就會(huì)形成類(lèi)似于流水線(xiàn)一樣的前后依賴(lài)關(guān)系。在部分分區(qū)數(shù)據(jù)丟失時(shí)，Spark可以通過(guò)這個(gè)依賴(lài)關(guān)系重新計(jì)算丟失的分區(qū)數(shù)據(jù)，而不是對(duì)RDD的所有分區(qū)進(jìn)行重新計(jì)算134.Optionally,aPartitionerforkey-valueRDDs

如果RDD里面存的數(shù)據(jù)是key-value形式，則可以傳遞一個(gè)自定義的Partitioner進(jìn)行重新分區(qū)，比如可以按key的hash值分區(qū)

一個(gè)partitioner，即RDD的分片函數(shù)。當(dāng)前Spark中實(shí)現(xiàn)了兩種類(lèi)型的分片函數(shù)，一個(gè)是基于哈希的HashPartitioner，另外一個(gè)基于范圍的RangePartitioner。只有對(duì)于key-value的RDD，才會(huì)有Partitioner，非key-value的RDD的Partitioner的值是None。Partitioner函數(shù)不但決定了RDD本身的分片數(shù)量，也決定了parentRDDShuffle輸出時(shí)的分片數(shù)量145.Optionally,alistofpreferredlocationstocomputeeachspliton

最優(yōu)的位置去計(jì)算，也就是數(shù)據(jù)的本地性

計(jì)算每個(gè)split時(shí)，在split所在機(jī)器的本地上運(yùn)行task是最好的，避免了數(shù)據(jù)的移動(dòng)