Spark核心原理初探

本文轉載自查看原文 2019-09-02 16:00 364 BigData

一、運行架構概覽

Spark架構是主從模型，分為兩層，一層管理集群資源，另一層管理具體的作業，兩層是解耦的。第一層可以使用yarn等實現。

Master是管理者進程，Worker是被管理者進程，每個Worker節點啟動一個Worker進程，了解每台機器的資源有多少，並將這些信息匯報各Master進程。

每個提交的作業程序對應一個Driver和多個Executor，每個Executor執行具體的任務。

圖 Spark基本運行架構

二、運行模式

Local
偽分布式
Standalone
Yarn/K8S

三、作業執行流程

1.提交Spark應用到機器上

程序jar包提交到機器上，程序在服務器上叫Application

通過Spark-submit執行提交的Application

Application提交到公用的集群上，有兩種資源分配方式：

FIFO，先提交的先執行，后提交的等待
FAIR，提交的作業都分配一些資源

2. 提交后會在本地客戶端啟動Driver進程

standalonde會通過反射的方式，創建和構造一個DriverActor進程出來

Driver進程會執行我們的Application應用程序，也就是我們編寫的代碼。

3.構造SparkContext

代碼首先構造SparkConf，再構造SparkContext

1)SparkContext初始化

SpakrContext初始化時，最重要的兩件事是構造出DAGSchedule和TaskSchedule。

2)TaskScheduler會通過對應的后台進程去連接Master

TaskScheduler有自己的后台進程

會向Master注冊Application

3)Master接收到Application的注冊請求后，會使用自己的資源調度算法，在Spark集群的Worker上，為這個Application啟動多個Executor

4)Worker會為Application啟動Executor

5)Executor啟動之后會自己反向注冊到TaskScheduler上去

所有Executor都反向注冊到Driver上之后，Driver結束Context初始化，會繼續執行我們自己編寫的代碼。

4.每執行一個action就會創建一個Job，Job會提交給DAGScheduler

5.DAGScheduler會將Job划分為多個stage，然后每個stage創建TaskSet

stage划分算法非常重要

6.TaskScheduler會把TaskSet每一個task提交到Executor上執行

task分配算法非常重要

7.Executor每接收一個task，都會用TaskRunner來封裝task，然后從線程池里取出一個線程，執行這個task

8.TaskRunner將我們編寫的代碼，也就是要執行的算子以及函數，拷貝，反序列化，然后執行Task

Task有兩種，ShuffleMapTask和ResultTask，只有最后一個stage是ResultTask，之前的stage，都是ShuffleMapTask。

所以，最后整個Spark應用程序的執行，就是stage划分批次為TaskSet，提交到executor執行，每個task針對RDD的一個partition，執行我們定義的算子和函數。以此類推，直到所有操作都執行完為止。

四、作業流程再探討

簡單來說，是將spark程序翻譯成spark core可執行的Task的過程

1.BSP(Bulk synchronous parallel)並行模型

比如，wordcount程序，首先是並行的本地的過濾操作，將字符串轉為單詞，該過程可以拆成多個同質的Task，這些Task之間沒有依賴

單詞聚合的時候，就產生了依賴，會等待前一個階段所有任務都執行完，屏障同步

分布式環境中，前后階段可能在不同節點上，會產生通信

同步體現在上一階段任務全執行完，下一階段任務才可以執行

異步模型，上一階段有些任務沒執行完，有些任務執行完，下一階段就可以啟動

2.Word Count

提交作業后，作業會像master請求一些資源，master會幫忙啟動driver進程和Executor進程，服務於word count程序，這個程序打包為jar包，分發到driver上，driver會啟動二進制打好的包，包啟動之后，會將作業編譯解析成細粒度的Task。Task執行順序，由driver決定，編譯成一個個stage，每個stage有具體的Task，這樣一步步按順序並行執行，直到最后順利完成job，將作業狀態匯報給管理者，可以通過日志等查詢。

創建邏輯查詢計划

創建物理查詢計划

flatMap、map的操作只是將單條記錄，將輸入是一行的字符串轉成了(String, int)類型，這一過程完全可以在本地運算，和集群上其他節點是沒有關聯的，可以通過操作符的合並將前三步合並為一個stage，不能合並的聚合操作成為了另一個stage

寬窄依賴是划分stage的依據

這就是創建物理查詢計划，防止出現大量的RDD，減少任務的調度開銷

在同一個stage內，根據數據划分規則，或數據塊的數量等拆分成多個並行的任務，下面拆分成了四個並行的任務，每個Task都按順序執行了textFile、flatMap、map

因此，同一個stage是同質同樣的Task的集合

只有當前一個stage所有任務完成后，下一個stage才執行

Driver中的TaskScheduler會調度Task，根據executor匯報的資源情況和stage中Task的執行情況，調度到executor上執行，executor會具體分配線程執行Task，執行完后會將Task執行正確完成狀態分發返回給driver，driver再根據情況去調度，直到該stage所有Task執行完畢，才執行下一個stage的Task

DAGScheduler和TaskScheduler

DAGScheduler將邏輯查詢計划轉為物理查詢計划，切分為stage，stage內部會產生TaskSet

TaskScheduler調度TaskSet中具體的Task

Executor上由Block manager管理它能對應執行的數據塊，即相應的partition，分配給Task，分配線程執行具體任務，執行過程組件會匯報狀態給相應調度器

Spark作業層級

job : A job is triggered by an action, like count() or saveAsTextFile(). Click on a job to see information about the stages of tasks inside it. 所謂一個 job，就是由一個 rdd 的 action 觸發的動作，可以簡單的理解為，當你需要執行一個 rdd 的 action 的時候，會生成一個 job。
stage : stage 是一個 job 的組成單位，就是說，一個 job 會被切分成 1 個或 1 個以上的 stage，然后各個 stage 會按照執行順序依次執行。
task : A unit of work within a stage, corresponding to one RDD partition。即 stage 下的一個任務執行單元，一般來說，一個 rdd 有多少個 partition，就會有多少個 task，因為每一個 task 只是處理一個 partition 上的數據。