基本流程

用戶(hù)編寫(xiě)的Spark應(yīng)用程序最開(kāi)始都要初始化SparkContext。

用戶(hù)編寫(xiě)的應(yīng)用程序中，每執(zhí)行一個(gè)action操作，就會(huì)觸發(fā)一個(gè)job的執(zhí)行，一個(gè)應(yīng)用程序中可能會(huì)生成多個(gè)job執(zhí)行。一個(gè)job如果存在寬依賴(lài)，會(huì)將shuffle前后劃分成兩個(gè)stage，前一個(gè)stage會(huì)將計(jì)算結(jié)果臨時(shí)進(jìn)行存儲(chǔ)，后一個(gè)stage則進(jìn)行讀取，完成數(shù)據(jù)交換。

每個(gè)stage中，需要執(zhí)行的計(jì)算過(guò)程會(huì)被劃分成多個(gè)邏輯相同的一組Task，每個(gè)Task會(huì)被提交到Executor中運(yùn)行。當(dāng)Task運(yùn)行完成后，會(huì)將運(yùn)行結(jié)果返回至Driver中。?

主要組件

Driver端

1. DAGScheduler：負(fù)責(zé)將Job劃分為Stage，再將Stage劃分為T(mén)askSet；
2. TaskScheduler：負(fù)責(zé)任務(wù)的調(diào)度；
3. SchedulerBackend：負(fù)責(zé)資源的分配，并把Task提交給Executor中執(zhí)行。

Executor端

1. BlockManager：緩存RDD、緩存Task運(yùn)行結(jié)果。

Job提交執(zhí)行流程

Task提交

SparkContext將RDD的action操作轉(zhuǎn)化為Job，并將Job交給DAGScheduler做進(jìn)一步處理。

DAGScheduler首選根據(jù)shuffle劃分stage，根據(jù)stage中分區(qū)的數(shù)量，生成一組Task（即TaskSet），生成Task時(shí)還會(huì)計(jì)算Task的最佳執(zhí)行位置。DAGScheduler會(huì)根據(jù)RDD是否進(jìn)行了緩存來(lái)確定是否具有最佳運(yùn)行位置。

DAGScheduler將Stage生成TaskSet以后，會(huì)將TaskSet交給TaskScheduler進(jìn)行處理，TaskScheduler負(fù)責(zé)將Task提交到集群中運(yùn)行，并負(fù)責(zé)失敗重試，為DAGScheduler返回事件信息等。

當(dāng)有任務(wù)提交至TaskScheduler中時(shí)，TaskScheduler會(huì)通知SchedulerBackend分配計(jì)算資源。SchedulerBackend將所有可用的Executor的資源信息轉(zhuǎn)換為WorkerOffer交給TaskScheduler。TaskScheduler負(fù)責(zé)根據(jù)這些WorkerOffer在相應(yīng)的Executor分配TaskSet中的Task。

SchedulerBackend中通過(guò)使用Map結(jié)構(gòu)記錄每一個(gè)ExecutorData的映射，即可管理所有Executor的CPU使用的情況。為計(jì)算任務(wù)分配計(jì)算資源時(shí)，只需要遍歷所有的ExecutorData，分配可用的資源即可。

TaskScheduler在接受到DAGScheduler提交的TaskSet以后，會(huì)為每個(gè)TaskSet創(chuàng)建一個(gè)TaskSetManager，用于管理該TaskSet中所有任務(wù)的運(yùn)行。TaskSetManager會(huì)根據(jù)Task中的最佳運(yùn)行位置計(jì)算TaskSet的所有本地運(yùn)行級(jí)別，本地運(yùn)行的級(jí)別決定了Task最終在哪個(gè)Executor中運(yùn)行。Spark中本地運(yùn)行級(jí)別從小到大可分為進(jìn)程本地化、節(jié)點(diǎn)本地化、無(wú)優(yōu)先位置、機(jī)架本地化、任意節(jié)點(diǎn)。

Task執(zhí)行

Executor接收到SchedulerBackend提交的LaunchTask消息后，即可運(yùn)行該消息中包含的Task。Executor將接收到的Task封裝到TaskRunner中，TaskRunner是一個(gè)Runnable接口，從而可以將該任務(wù)提交到線(xiàn)程池中運(yùn)行。

當(dāng)在一個(gè)Executor中同時(shí)運(yùn)行多個(gè)Task時(shí)，多個(gè)Task共享Executor中SparkEnv的所有組件，共用Executor中分配的內(nèi)存。如使用Spark廣播變量時(shí)，每個(gè)Executor中會(huì)存在一份，Executor中所有的任務(wù)會(huì)共享這一份變量。當(dāng)Executor中的BlockManager緩存了某rdd某分區(qū)的數(shù)據(jù)時(shí)，在該Executor上調(diào)度使用這個(gè)RDD的這個(gè)分區(qū)的數(shù)據(jù)的Task執(zhí)行，可以有效減少網(wǎng)絡(luò)加載數(shù)據(jù)的過(guò)程，減少網(wǎng)絡(luò)傳輸。

當(dāng)Executor中Task運(yùn)行完成時(shí)，需要將Task的運(yùn)行結(jié)果返回Driver程序，Driver程序根據(jù)結(jié)果判斷該Stage是否計(jì)算完成，或者該Job是否計(jì)算完成。