一、應用執行機制

一個應用的生命週期即，使用者提交自定義的作業之後，Spark框架進行處理的一系列過程。

在這個過程中，不同的時間段裡，應用會被拆分為不同的形態來執行。

1、應用執行過程中的基本元件和形態

Driver：
執行在客戶端或者叢集中，執行Application的main方法並建立SparkContext，調控整個應用的執行。

Application：
使用者自定義並提交的Spark程式。

Job：
一個Application可以包含多個Job，每個Job由Action操作觸發。

Stage：
比Job更小的單位，一個Job會根據RDD之間的依賴關係被劃分為多個Stage，每個Stage中只存有RDD之間的窄依賴，即Transformation運算元。

TaskSet：
每個Stage中包含的一組相同的Task。

Task：
最後被分發到Executor中執行的具體任務，執行Stage中包含的運算元。

明確了一個應用的生命週期中會有哪些元件參與之後，再來看看使用者是怎麼提交Spark程式的。

2、應用的兩種提交方式

Driver程序執行在客戶端（Client模式）：

即使用者在客戶端直接執行程式。
程式的提交過程大致會經過以下階段：

1. 使用者執行程式。
2. 啟動Driver進行（包括DriverRunner和SchedulerBackend），並向叢集的Master註冊。
3. Driver在客戶端初始化DAGScheduler等元件。
4. Woker節點向Master節點註冊並啟動Executor（包括ExecutorRunner和ExecutorBackend）。
5. ExecutorBackend啟動後，向Driver內部的SchedulerBackend註冊，使得Driver可以找到計算節點。
6. Driver中的DAGScheduler解析RDD生成Stage等操作。
7. Driver將Task分配到各個Executor中並行執行。

Driver程序執行在叢集中（某個Worker節點，Cluster模式）：

即使用者將Spark程式提交給Master分配執行。
大致會經過一下流程：

1. 使用者啟動客戶端，提交Spark程式給Master。
2. Master針對每個應用分發給指定的Worker啟動Driver進行。
3. Worker收到命令之後啟動Driver程序（即DriverRunner和其中的SchedulerBackend），並向Master註冊。
4. Master指定其他Worker啟動Executor（即ExecutorRunner和其內部的ExecutorBackend）。
5. ExecutorBackend向Driver中的SchedulerBackend註冊。
6. Driver中的DAGScheduler解析RDD生產Stage等。
7. Executor內部啟動執行緒池並行化執行Task。

可以看到，兩種程式的提交方式在處理過程中，僅僅是在哪個地方啟動Driver程序的區別而已。
為Client模式中時（使用Spark Shell直接執行的程式），Driver就在客戶端上。
為Cluster模式時（提交Spark程式到Master），Driver執行與叢集中的某個Worker節點。

二、排程與任務分配模組

Spark框架就像一個作業系統一樣，有著自己的作業排程策略，當叢集執行在不同的模式下，排程不同級別的單位，使用的策略也是有所不同的。

1、Application之間的排程

當有多個使用者提交多個Spark程式時，Spark是如何排程這些應用併合理地分配資源呢？

Standalone模式下，預設使用FIFO，每個app會獨佔所有資源

可以通過以下幾個引數調整叢集相關的資源：

• spark.cores.max：調整app可以在整個叢集中申請的CPU core數量
• spark.deploy.defaultCores：預設的CPU core數量
• spark.executor.memory：限制每個Executor可用的記憶體

在Mesos模式下，可以使用

• spark.mesos.coarse=true設定靜態配置資源的策略
• 使用mesos://URL且不配置spark.mesos.coarse=true（每個app會有獨立固定的記憶體分配，空閒時其他機器可以使用其資源）

在Yarn模式下，提交作業時可以使用

• 通過–num-executors控制分配多少個Executor給app
• –executor-memory和–executor-cores分別控制Executor的記憶體和CPU core

2、Application內部的Job排程機制

一個Application中，由各個Action觸發的多個Job之間也是存在排程關係的。

Action操作實現上是呼叫了SparkContext的runJob方法提交Job。

Spark中排程Job有兩種策略

FIFO：

• 第一個Job分配其所需的所有資源
• 第二個Job如果還有剩餘資源的話就分配，否則等待

FAIR：

• 使用輪詢的方式排程Job

可以通過配置spark.scheduler.mode調整Job的排程方式

另外也可以配置排程池，具體參考官方文件
或者參考conf/fairscheduler.xml.template檔案。

3、Job中的Stage排程

Stage是由DAGScheduler元件生產的，在原始碼中，有三個比較特殊的變數：

• waitingStages：儲存等待執行的Stages
• runningStages：儲存正在執行的Stages
• failedStages：儲存執行失敗的Stage

Spark會通過廣度優先遍歷找到最開始的Stage執行，若有父Stage沒有執行完則等待。

4、Stage中的Task排程

暫未了解。。。

三、I/O制度

Spark雖然是基於記憶體計算的框架，但是不可避免的也會接觸到一些儲存層，那麼在和儲存層互動的時候，Spark做了哪些工作？

1、序列化

序列化的本質就是將物件轉換為位元組流，可以理解為將連結串列中儲存的非連續空間的資料儲存轉化為連續空間儲存的陣列中

Spark為什麼要做序列化操作？

記憶體或磁碟中RDD會含有物件的儲存，而在節點間資料的傳輸時，序列化之後的資料可以節約空間和提高效率。

2、壓縮

壓縮是日常生活中的一個常見操作，好處顯而易見，節約空間，從而就可以獲得時間上的效率。

Spark中序列化之後的資料可以進行壓縮以減少空間開銷。
Spark支援兩種壓縮演算法

• Snappy演算法：高壓縮速度
• LZF演算法：高壓縮比

在不同的場景中選擇不同的壓縮演算法可以有效的提高程式執行的效率。

壓縮配置方式：

1. 啟動前在spark-env.sh中設定：export SPARK_JAVA_OPTS=”-Dspark.broadcast.compress”
2. 在應用程式中配置
   conf.getBoolean(“spark.broadcast.compress,true”)
   conf.set(“spark.broadcast.compress”,true)

3、塊管理

RDD從物理上看是一個元資料結構，記錄著Block和Node之間的對映關係。

儲存RDD是以Block塊為單位的，每個分割槽對應一個塊，PartitionID通過元資料資訊可以對映到Block。

BlockManager管理和介面、塊讀寫流程、資料塊讀寫管理等細節待繼續深入瞭解。

四、通訊模組

Spark中使用Akka作為通訊框架

• Actors是一組包含狀態和行為的物件
• 一個Actor接收到其他Actor的資訊之後可以根據需求做出各種反應
• Client、Master、Worker等都是一個Actor

Spark各個元件的之間協調工作都是基於Akka機制來的，待深入瞭解的有：

• Client Actor通訊程式碼邏輯
• Master Actor通訊程式碼邏輯
• Worker Actor訊息處理邏輯

五、容錯機制

之前講過，RDD之間的運算元操作會形成DAG圖，RDD之間的依賴關係會形成Lineage。

要理解Lineage機制首先要明確兩種依賴的概念：

• Shuffle Dependencies（寬依賴）
  父分割槽可以被多個子分割槽所用
  即多對多的關係

• Narrow Dependencies（窄依賴）
  父分割槽最多被一個子分割槽所用
  即一對一或者多對一的關係

當出現某個節點計算錯誤的時候，會順著RDD的操作順序往回走

一旦是Narrow Dependencies錯誤，重新計算父RDD分割槽即可，因為其不依賴其他節點

而如果Shuffle Dependencies錯誤，重算代價較高，因為一旦重新計算其依賴的父RDD分割槽，會造成冗餘計算

這時候就需要人為的新增檢查點來提高容錯機制的執行效率

什麼情況下需要加CheckPoint

• DAG中的Lineage過長，如果重算開銷太大，故在特定幾個Shuffle Dependencies上做CheckPoint是有價值的。
• Checkpoint會產生磁碟開銷，因為其就是將資料持久化到磁碟中，所以做檢查點的RDD最好是已經在記憶體中快取了。

六、Shuffle機制

Shuffle的定義：對無規則的資料進行重組排序等過程

為什麼要Shuffle：分散式計算中資料是分佈在各個節點上計算的，而彙總統計等操作需要在所有資料上執行

Spark中Shuffle經歷的階段：

Shuffle Write
    將各個節點資料寫入到指定分割槽
        1、根據下一個Stage分割槽數分成相應的Bucket
        2、將Bucket寫入磁碟
Shuffle Fetch
    獲取各個分割槽傳送的資料
        1、在儲存有Shuffle資料節點的磁碟Fetch需要的資料
        2、Fetch到本地之後進行自定義的聚集函式操作

最後記錄一下提交Spark作業的方法
在spark的bin目錄下
執行spark-submit指令碼
./spark-submit \
–class 入口函式所在的類名全稱 \
–master spark master節點的地址（預設埠7077）\
–executor-memory 指定worker中Executor的記憶體 \
–total-executor-cores 100 \
jar檔案所在的目錄 \