1 總體框架結構圖

由上圖我們可以看到Spark應用程式架構主要由Driver Program和Executor構成，Driver負責執行main()和建立SparkContext，Executor主要負責執行Task任務。

2 各大重要元件和概念

2.1 重要概念

2.2 相關元件

3 應用的提交執行流程

3.1 執行模式

1）Spark應用的執行模式有以下幾種：

• local[n]：本地模式
• standalone：使用spark內部的資源管理器，master-slaves架構
• mesos : Apache下的一個資源管理框架
• yarn : Hadoop下的一個資源管理框架

2）根據Driver的執行位置來劃分有以下兩種執行模式：

• Cluster：Driver和Executor執行在叢集中的Worker上
• Client：Driver執行在客戶端上，Executor執行在Worker上

3）命令列

• spark-shell：bin/spark-shell –master yarn –deploy-mode client（yarn的client模式）
• spark-submit：spark-shell底層也是呼叫spark-submit，如果你想知道它有什麼引數，可以直接在spark目錄下輸入 spark-submit來檢視

3.2 流程

這裡我們就以 Spark On Yarn 為例來說明

3.2.1 Yarn-Client 模式

流程如下：
1. 在客戶端通過spark-submit向Yarn提交Application
2. 應用在Client啟動Driver，建立SparkContext並進行初始化
3. ResourceManger為應用申請一個Container來啟動Application Master，用來與Client中的SparkContext進行通訊。
4. Application Master即承擔起了為應用申請資源執行Executor的責任，一旦申請到資源，那麼就會在指定的NodeManager的Container上建立Executor程序，Executor會向SparkContext註冊，保持通訊
5. SparkContext會根據任務所需資源，資料本地性等條件指定Task的執行在哪個Executor上，將Task分發到指定Executor並開始執行，Executor會週期向SparkContext彙報任務執行情況
6. 一旦所有任務執行完畢，SparkContext會向ResourceManger申請登出自己並關閉

3.2.2 Yarn-Cluster 模式

流程如下：
1. 在客戶端通過spark-submit向Yarn提交Application
2. ResourceManger為應用申請一個Container來啟動Application Master，並且在Application Master進行SparkContext等初始化。
3. Application Master即承擔起了為應用申請資源執行Executor的責任，一旦申請到資源，那麼就會在指定的NodeManager的Container上建立Executor程序，Executor會向SparkContext註冊，保持通訊
4. Applicaiton Master會根據任務所需資源，資料本地性等條件指定Task的執行在哪個Executor上，將Task分發到指定Executor並開始執行，Executor會週期向Applicaiton Master彙報任務執行情況
5. 一旦所有任務執行完畢，Applicaiton Master會向ResourceManger申請登出自己並關閉

4 Spark的讀寫流程

4.1 寫入流程

1. RDD呼叫compute( )方法進行指定分割槽的寫入。
2. CacheManager中呼叫BlockManater判斷資料是否已經寫入，如果未寫則寫入。
3. BlockManager中資料與其他節點同步。（與副本節點通訊）
4. BlockManager根據儲存級別寫入指定的儲存層。（儲存級別決定了資料是儲存在記憶體還是磁碟，或者是兩者結合）
5. BlockManager向主節點彙報儲存狀態。

4.2 讀取流程

1. RDD呼叫compute( )方法進行指定分割槽的讀取。
2. 呼叫BlockManager的get()進行讀取
   
   • 在本地同步讀取資料塊，首先看能否在記憶體讀取資料塊，如果不能讀取，則看能否從Tachyon讀取資料塊，如果仍不能讀取，則看能否從磁碟讀取資料塊。
   • 如果本地沒有資料，就進行遠端讀取：遠端獲取呼叫路徑，然後getRemote呼叫doGetRemote，通過BlockManagerWorker.syncGetBlock從遠端獲取資料。

5 Spark各元件之間的通訊方式和流程

5.1 通訊框架

1.6之前

• Spark在模組間通訊使用的是AKKA框架。AKKA基於Scala開發，用於編寫Actor應用。
• Actors是一些包含狀態和行為的物件。它們通過顯式傳遞訊息來進行通訊，訊息會被髮送到它們的訊息佇列中。對訊息佇列中的訊息進行處理作出響應
• RPC是單獨通過AKKA實現，資料以及檔案傳輸是採用netty實現，而AKKA底層也是使用Netty來實現的

1.6 後

• 通過netty封裝了一套簡潔的類似於AKKA Actor模式的RPC介面，並逐步拋棄AKKA這個重量級框架
• 2.0 之後，所有的網路功能模組都是通過Netty來實現的

5.2 通訊的簡單流程

• 客戶端傳送請求訊息，經過Encoder加上頭資訊，通過網路傳送給服務端
• 服務端收到訊息之後，經過TransportFrameDecoder進行處理，得到訊息型別和訊息體，在解析得到具體的請求資訊
• 最後由TransportChannelHandler處理具體的請求資訊，根據訊息型別判斷是否迴應

6 Spark的容錯機制

分散式系統常用的容錯機制

• 資料檢查點：成本高，耗費資源
• 記錄資料的更新：耗費資源較少

Spark所採用的容錯機制是記錄資料更新和資料檢查點相結合的形式。原因如下：

• RDD只支援粗粒度轉換，即在大量記錄上執行的單個操作。將建立RDD的一系列Lineage（即血統）記錄下來，以便恢復丟失的分割槽。也就是說當子RDD分割槽丟失時，只要重新計算它對應的父RDD分割槽就可以進行恢復，這對於窄依賴來說不存在冗餘計算。
• 寬依賴也就是Shuffle Dependency，也就是發生在shuffle階段的依賴，如果子RDD分割槽丟失了，那麼重新計算的代價就比較大，因為有多個父RDD分割槽對應該子RDD分割槽，那麼重新計算時所有的父RDD分割槽都得重新計算，造成了計算冗餘

以下情況需要新增檢查點

• DAG中依賴線過長，如果重算，則開銷太大
• 在Shuffle Dependency上做CheckPoint(檢查點)獲得的收益更大

在RDD計算中，通過檢查點機制進行容錯，傳統做檢查點有兩種方式：通過冗餘資料和日誌記錄更新操作。RDD中所採用的就是通過冗餘資料來快取資料，然後對已進行冗餘操作的RDD執行刪除該祖先的RDD依賴

官方建議，做檢查點的RDD做好先快取在記憶體中，否則需要進行重新計算

7 Spark的Shuffle機制

Shuffle中文意思就是混洗，跟MapReduce中的Shuffle的思想是相同，就是資料的重新分割槽和組合

Spark的Shuffle是在stage的承接階段完成的，前面的stage會根據後面stage的分割槽數來將資料按照一定的規則分成相應的bucket，然後寫到磁碟上。後續的stage會從元資料那裡獲得指定資料的所在節點，將資料拉取本地做進一步的操作

Shuffle分為兩個階段

• Shuffle Write ：Shuffle的第一步驟，就是將前一個stage中的資料寫到磁碟，用於第二步的Fetch
  
  • 主要是在ShuffleMapTask中執行(runTask)方法
  • 如果設定了map端進行聚合的話，那麼會先執行資料在map的合併，減少網路傳輸
  • Consolidate Shuffle相比普通Shuffle的優勢在：普通shuffle產生的Shuffle檔案數為map任務數*reduce任務數，而Consolidate Shuffle的理論Shuffle檔案數為Spark Core數*Reducer數，而概念上C_Shuffle引入了檔案陣列，一個Bucket不再對應一個檔案，而是對應檔案中的一個segment
• Shuffle Fetch：通過本地或者遠端抓取需要的資料到本節點進行計算，完成操作如儲存資料到指定路徑或者作為下一個Shuffle的Shuffle Write

Shuffle Aggregator並不會對所有情況下的資料進行排序，所以Aggregator分為不需要外排和需要外排兩種方式

• 不需要外排：資料全部儲存在記憶體當中，使用AppendOnlyMap來進行資料的儲存，資料是來一個處理一個，更新到map中，所以reducer的記憶體必須足夠大，能夠儲存該分割槽的所有key和count的值
• 需要外排：reduce型別的操作，記憶體沒辦法存放所有的key-value對，必須藉助外部磁碟

在Reduce端，各個Task會併發啟動多個執行緒同時從多個Map Task端拉取資料。由於Reduce階段的主要任務是對資料進行按組規約。也就是說，需要將資料分成若干組，以便以組為單位進行處理。大家知道，分組的方式非常多，常見的有：Map/HashTable（key相同的，放到同一個value list中）和Sort（按key進行排序，key相同的一組，經排序後會挨在一起），這兩種方式各有優缺點，第一種複雜度低，效率高，但是需要將資料全部放到記憶體中，第二種方案複雜度高，但能夠藉助磁碟（外部排序）處理龐大的資料集。Spark前期採用了第一種方案，而在最新的版本中加入了第二種方案， MapReduce則從一開始就選用了基於sort的方案。
摘抄自董西成的文章連結如下http://dongxicheng.org/framework-on-yarn/apache-spark-shuffle-details/