Hadoop 和 Spark 的關系

Spark 運(yun)(yun)算(suan)(suan)(suan)比 Hadoop 的(de) MapReduce 框(kuang)架快(kuai)(kuai)的(de)原因(yin)是因(yin)為(wei) Hadoop 在(zai)(zai)(zai)一(yi)次(ci) MapReduce 運(yun)(yun)算(suan)(suan)(suan)之(zhi)后,會(hui)將(jiang)數據的(de)運(yun)(yun)算(suan)(suan)(suan)結果從(cong)內存寫入(ru)到(dao)磁(ci)盤(pan)中(zhong),第(di)二次(ci) Mapredue 運(yun)(yun)算(suan)(suan)(suan)時在(zai)(zai)(zai)從(cong)磁(ci)盤(pan)中(zhong)讀取數據,所(suo)以(yi)(yi)其(qi)瓶(ping)頸在(zai)(zai)(zai)2次(ci)運(yun)(yun)算(suan)(suan)(suan)間的(de)多余(yu) IO 消耗. Spark 則(ze)是將(jiang)數據一(yi)直緩存在(zai)(zai)(zai)內存中(zhong),直到(dao)計算(suan)(suan)(suan)得到(dao)最后的(de)結果,再將(jiang)結果寫入(ru)到(dao)磁(ci)盤(pan),所(suo)以(yi)(yi)多次(ci)運(yun)(yun)算(suan)(suan)(suan)的(de)情(qing)況下, Spark 是比較快(kuai)(kuai)的(de). 其(qi)優化了迭代(dai)式工(gong)作負(fu)載(zai)

Spark 的(de)主要特點(dian)還包括(kuo):

• • (1)提供 Cache 機制來支(zhi)持需(xu)要反復迭代(dai)計算或者多次數據共(gong)享,減少數據讀取的 IO 開銷;
  • (2)提供了一套支持 DAG 圖的分布式并行計算的編程(cheng)框架(jia),減少多次計算之間中(zhong)間結果寫到(dao) Hdfs 的開銷;
  • (3)使用多線程池模型減少(shao) Task 啟動開稍, shuffle 過(guo)程中避免不必要(yao)的 sort 操(cao)作并(bing)減少(shao)磁盤 IO 操(cao)作。(Hadoop 的 Map 和 reduce 之(zhi)間的 shuffle 需要(yao) sort)

Spark 系(xi)統(tong)架(jia)構(gou)

明確相(xiang)關術語

• Application: Appliction都是指用戶編寫的Spark應用程序，其中包括一個Driver功能的代(dai)碼和分布在集群中多個節點上運行的Executor代(dai)碼
• Driver:  Spark中的Driver即運行上述Application的main函數(shu)并創建(jian)SparkContext，創建SparkContext的目的是為了準備Spark應用程序的運行環境，在Spark中有SparkContext負(fu)責(ze)與ClusterManager通信，進行資源申請、任務的分配和監控等，當Executor部分運行完畢后，Driver同時負責將SparkContext關閉(bi)，通常用SparkContext代表Driver
• Executor:  某個Application運行在worker節點上的一個進(jin)程，  該進程負責運行某些Task， 并且負責將數據存到內存或磁盤上，每個Application都有各自獨立的一批Executor， 在Spark on Yarn模式下，其進程名稱為CoarseGrainedExecutor Backend。一個CoarseGrainedExecutor Backend有(you)且僅有(you)一個Executor對象， 負責將Task包裝成taskRunner,并從線程池中抽取一個空閑線程運行Task， 這個每一個oarseGrainedExecutor Backend能并行運行Task的數量取決與分配給它的cpu個數
• Cluter Manager：指的是在集群上獲取資源的外部服務。目前有三種類型

1. 1. 　Standalon : spark原生的資源管理，由Master負責資源的分配
   2. 　Apache Mesos:與hadoop MR兼容性良好的一種資源調度框架
   3. 　Hadoop Yarn: 主要是指Yarn中的ResourceManager

• Worker: 集群中任何可以運行Application代碼的節點，在Standalone模式中指的是通過slave文件配置的Worker節點，在Spark on Yarn模式下就是NoteManager節點
• Task: 被送到某個Executor上的工作單元，但hadoopMR中的MapTask和ReduceTask概念一樣，是運行Application的基本單位，多(duo)個(ge)Task組成一個(ge)Stage，而Task的調度和管理等是由TaskScheduler負責
• Job: 包含多個Task組成的并行計算，往往由Spark Action觸(chu)發生成， 一個Application中往往會產生多個Job
• Stage: 每個Job會被拆分成多組Task， 作為一個TaskSet， 其名稱為Stage，Stage的劃分和調度是有DAGScheduler來負責的，Stage有非最終的Stage（Shuffle Map Stage）和最終的Stage（Result Stage）兩種，Stage的邊界就是(shi)發生shuffle的地方
• DAGScheduler: 根據Job構建基于Stage的DAG（Directed Acyclic Graph有向無環圖)，并提交Stage給TASkScheduler。 其劃分Stage的依據是RDD之間的依賴的關系找出開銷最小的調度方法，如下圖
• 
• TASKSedulter: 將TaskSET提交給worker運行，每個Executor運行什么Task就是在此處分配的. TaskScheduler維護所有TaskSet，當Executor向Driver發生心跳時，TaskScheduler會根據資源剩余情況分配相應的Task。另外TaskScheduler還維護著所有Task的運行標簽，重試失敗的Task。下圖展示了TaskScheduler的作用
• 
• 在不同運行模式中任務調度器具體為：

1. 1. 　　Spark on Standalone模式為TaskScheduler
   2. 　　YARN-Client模式為YarnClientClusterScheduler
   3. 　　YARN-Cluster模式為YarnClusterScheduler

• 將這些術語串起來的運行層次圖如下：
• 
• Job=多(duo)個stage，Stage=多(duo)個同種(zhong)task, Task分(fen)為ShuffleMapTask和ResultTask，Dependency分(fen)為ShuffleDependency和NarrowDependency

　　整個 Spark 集群中,分為 Master 節點與 worker 節點,,其中 Master 節點負責將串行任務變成可并行執行的任務集Tasks, 同時還負責出錯問題處理等,而 Worker 節點負責執行任務. 
　　Driver 的(de)功能是創(chuang)建 SparkContext, 負責(ze)執行用戶(hu)寫的(de) Application 的(de) main 函數進程(cheng),Application 就是用戶(hu)寫的(de)程(cheng)序(xu). 
　　不同的模式可能會將 Driver 調度到不同的節點上執行.集群管理模式里, local 一般用于本地調試. 
　　每個 Worker 上存在一個或多個 Executor 進程,該對象擁有一個線程池,每個線程負責一個 Task 任務的執行.根據 Executor 上 CPU-core 的數量,其每個時間可以并行多個 跟 core 一樣數量的 Task.Task 任務即為具體執行的 Spark 程序的任務. 

• spark運行流程圖如下：
• 

1. 構建Spark Application的運行環境，啟動SparkContext
2. SparkContext向資源管理器（可以是Standalone，Mesos，Yarn）申請運行Executor資源，并啟動StandaloneExecutorbackend，
3. Executor向SparkContext申請Task
4. SparkContext將應用程序分發給Executor
5. SparkContext構建成DAG圖，將DAG圖分解成Stage、將Taskset發送給Task Scheduler，最后由Task Scheduler將Task發送給Executor運行
6. Task在Executor上運行，運行完釋放所有資源

     Spark運行特點：

1. 每個Application獲取專屬的executor進程，該進程在Application期間一直駐留，并以多(duo)線(xian)程方式運(yun)行Task。這種Application隔離機制是有優勢的，無論是從調度角度看（每(mei)個Driver調(diao)度他自己的任務），還是從運行角度看（來自不同Application的Task運行在不同JVM中），當然這樣意味著Spark Application不能跨應用程序共享數據，除非將數據寫入外部存儲系統
2. Spark與資源管理器無關(guan)，只要能夠獲取executor進程，并能保持相互通信就可以了
3. 提(ti)交SparkContext的Client應該(gai)靠近Worker節點（運行Executor的節點），最好是在同一個Rack里，因為Spark Application運行過程中SparkContext和Executor之間有大量的信息交換
4. Task采用了數據(ju)本地性和推測(ce)執行的(de)優化機制

Spark作業基本(ben)運(yun)行原理

詳細原理見上圖(tu)。

　　我們使用(yong)spark-submit提(ti)交一(yi)個(ge)Spark作(zuo)(zuo)業之后，這個(ge)作(zuo)(zuo)業就(jiu)會(hui)(hui)啟(qi)(qi)動(dong)一(yi)個(ge)對應的(de)(de)Driver進(jin)程(cheng)。根據(ju)(ju)你使用(yong)的(de)(de)部(bu)署模式（deploy-mode）不(bu)同，Driver進(jin)程(cheng)可(ke)能(neng)在(zai)本地啟(qi)(qi)動(dong)，也可(ke)能(neng)在(zai)集(ji)(ji)群(qun)(qun)中(zhong)某個(ge)工(gong)作(zuo)(zuo)節點上啟(qi)(qi)動(dong)。Driver進(jin)程(cheng)本身會(hui)(hui)根據(ju)(ju)我們設(she)置的(de)(de)參數(shu)(shu)，占(zhan)有一(yi)定數(shu)(shu)量(liang)的(de)(de)內(nei)存(cun)和CPU core。而(er)Driver進(jin)程(cheng)要(yao)做的(de)(de)第一(yi)件事情，就(jiu)是向集(ji)(ji)群(qun)(qun)管(guan)理器(YARN或者其(qi)他資(zi)(zi)源(yuan)管(guan)理集(ji)(ji)群(qun)(qun)）申請(qing)運(yun)行Spark作(zuo)(zuo)業需要(yao)使用(yong)的(de)(de)資(zi)(zi)源(yuan)，這里的(de)(de)資(zi)(zi)源(yuan)指的(de)(de)就(jiu)是Executor進(jin)程(cheng)。YARN集(ji)(ji)群(qun)(qun)管(guan)理器會(hui)(hui)根據(ju)(ju)我們為Spark作(zuo)(zuo)業設(she)置的(de)(de)資(zi)(zi)源(yuan)參數(shu)(shu)，在(zai)各個(ge)工(gong)作(zuo)(zuo)節點上，啟(qi)(qi)動(dong)一(yi)定數(shu)(shu)量(liang)的(de)(de)Executor進(jin)程(cheng)，每(mei)個(ge)Executor進(jin)程(cheng)都占(zhan)有一(yi)定數(shu)(shu)量(liang)的(de)(de)內(nei)存(cun)和CPU core。

　　在申請到了作業(ye)執(zhi)行(xing)(xing)(xing)所需(xu)的(de)(de)(de)(de)資源之后(hou)，Driver進(jin)程就(jiu)會開始調度(du)和執(zhi)行(xing)(xing)(xing)我(wo)們編寫(xie)(xie)的(de)(de)(de)(de)作業(ye)代碼(ma)了。Driver進(jin)程會將我(wo)們編寫(xie)(xie)的(de)(de)(de)(de)Spark作業(ye)代碼(ma)分(fen)拆為(wei)多個(ge)(ge)(ge)stage，每(mei)個(ge)(ge)(ge)stage執(zhi)行(xing)(xing)(xing)一(yi)(yi)部(bu)分(fen)代碼(ma)片段(duan)，并(bing)(bing)為(wei)每(mei)個(ge)(ge)(ge)stage創建一(yi)(yi)批task，然后(hou)將這些(xie)task分(fen)配到各(ge)個(ge)(ge)(ge)Executor進(jin)程中執(zhi)行(xing)(xing)(xing)。task是(shi)最小的(de)(de)(de)(de)計(ji)算(suan)(suan)單元，負責執(zhi)行(xing)(xing)(xing)一(yi)(yi)模一(yi)(yi)樣的(de)(de)(de)(de)計(ji)算(suan)(suan)邏(luo)輯(ji)(ji)（也(ye)就(jiu)是(shi)我(wo)們自己編寫(xie)(xie)的(de)(de)(de)(de)某個(ge)(ge)(ge)代碼(ma)片段(duan)），只(zhi)是(shi)每(mei)個(ge)(ge)(ge)task處理的(de)(de)(de)(de)數據(ju)不同而已。一(yi)(yi)個(ge)(ge)(ge)stage的(de)(de)(de)(de)所有(you)task都執(zhi)行(xing)(xing)(xing)完畢之后(hou)，會在各(ge)個(ge)(ge)(ge)節點本(ben)地的(de)(de)(de)(de)磁盤文(wen)件中寫(xie)(xie)入計(ji)算(suan)(suan)中間(jian)結(jie)果(guo)(guo)，然后(hou)Driver就(jiu)會調度(du)運行(xing)(xing)(xing)下一(yi)(yi)個(ge)(ge)(ge)stage。下一(yi)(yi)個(ge)(ge)(ge)stage的(de)(de)(de)(de)task的(de)(de)(de)(de)輸入數據(ju)就(jiu)是(shi)上一(yi)(yi)個(ge)(ge)(ge)stage輸出的(de)(de)(de)(de)中間(jian)結(jie)果(guo)(guo)。如此(ci)循(xun)環往復，直到將我(wo)們自己編寫(xie)(xie)的(de)(de)(de)(de)代碼(ma)邏(luo)輯(ji)(ji)全部(bu)執(zhi)行(xing)(xing)(xing)完，并(bing)(bing)且計(ji)算(suan)(suan)完所有(you)的(de)(de)(de)(de)數據(ju)，得(de)到我(wo)們想(xiang)要的(de)(de)(de)(de)結(jie)果(guo)(guo)為(wei)止。

　　Spark是根據shuffle類算(suan)(suan)子(zi)(zi)(zi)來進(jin)行stage的劃(hua)分(fen)。如(ru)果我們的代(dai)碼中執(zhi)行了某個(ge)(ge)shuffle類算(suan)(suan)子(zi)(zi)(zi)（比如(ru)reduceByKey、join等(deng)），那么就(jiu)會(hui)(hui)在(zai)該算(suan)(suan)子(zi)(zi)(zi)處，劃(hua)分(fen)出一個(ge)(ge)stage界(jie)限來。可以大致理解為，shuffle算(suan)(suan)子(zi)(zi)(zi)執(zhi)行之前的代(dai)碼會(hui)(hui)被劃(hua)分(fen)為一個(ge)(ge)stage，shuffle算(suan)(suan)子(zi)(zi)(zi)執(zhi)行以及之后的代(dai)碼會(hui)(hui)被劃(hua)分(fen)為下一個(ge)(ge)stage。因此一個(ge)(ge)stage剛開始執(zhi)行的時候，它的每個(ge)(ge)task可能都會(hui)(hui)從上一個(ge)(ge)stage的task所(suo)在(zai)的節點，去通過(guo)(guo)網絡傳輸拉(la)取需要自己處理的所(suo)有(you)key，然后對拉(la)取到的所(suo)有(you)相(xiang)同(tong)的key使用我們自己編寫的算(suan)(suan)子(zi)(zi)(zi)函數執(zhi)行聚合操(cao)作（比如(ru)reduceByKey()算(suan)(suan)子(zi)(zi)(zi)接收的函數）。這個(ge)(ge)過(guo)(guo)程(cheng)就(jiu)是shuffle。

　　當我們(men)在代(dai)碼中執行了cache/persist等(deng)持久化(hua)操(cao)作時(shi)，根據(ju)我們(men)選(xuan)擇的持久化(hua)級別的不同(tong)，每個task計算出來(lai)的數據(ju)也會(hui)保存(cun)到Executor進程的內(nei)存(cun)或者(zhe)所在節點(dian)的磁盤文件中。

　　因此Executor的內(nei)存主要分(fen)為(wei)三塊(kuai)(kuai)(kuai)：第一(yi)塊(kuai)(kuai)(kuai)是(shi)讓(rang)task執行(xing)我(wo)們自(zi)己編寫(xie)的代(dai)碼時(shi)使(shi)用(yong)，默(mo)認(ren)(ren)是(shi)占(zhan)(zhan)Executor總(zong)內(nei)存的20%；第二塊(kuai)(kuai)(kuai)是(shi)讓(rang)task通過(guo)shuffle過(guo)程拉取了上一(yi)個stage的task的輸(shu)出后，進行(xing)聚合等(deng)操(cao)作時(shi)使(shi)用(yong)，默(mo)認(ren)(ren)也是(shi)占(zhan)(zhan)Executor總(zong)內(nei)存的20%；第三塊(kuai)(kuai)(kuai)是(shi)讓(rang)RDD持久化時(shi)使(shi)用(yong)，默(mo)認(ren)(ren)占(zhan)(zhan)Executor總(zong)內(nei)存的60%。

　　task的(de)(de)執(zhi)行(xing)(xing)(xing)速度是跟(gen)每(mei)個(ge)Executor進(jin)(jin)程(cheng)(cheng)的(de)(de)CPU core數(shu)量有直接關系的(de)(de)。一(yi)(yi)個(ge)CPU core同一(yi)(yi)時(shi)間只能執(zhi)行(xing)(xing)(xing)一(yi)(yi)個(ge)線(xian)(xian)程(cheng)(cheng)。而每(mei)個(ge)Executor進(jin)(jin)程(cheng)(cheng)上分(fen)配到的(de)(de)多個(ge)task，都(dou)是以(yi)每(mei)個(ge)task一(yi)(yi)條(tiao)線(xian)(xian)程(cheng)(cheng)的(de)(de)方式(shi)，多線(xian)(xian)程(cheng)(cheng)并(bing)發運行(xing)(xing)(xing)的(de)(de)。如果(guo)CPU core數(shu)量比(bi)(bi)較(jiao)(jiao)(jiao)充(chong)足(zu)，而且(qie)分(fen)配到的(de)(de)task數(shu)量比(bi)(bi)較(jiao)(jiao)(jiao)合(he)理，那么通(tong)常來說，可以(yi)比(bi)(bi)較(jiao)(jiao)(jiao)快速和高效(xiao)地執(zhi)行(xing)(xing)(xing)完這些task線(xian)(xian)程(cheng)(cheng)。

以上就是Spark作業的(de)基本(ben)運(yun)行原理的(de)說明

Refer

Spark(一): 基本架構及原理