MapReduce與Yarn 的詳細工作流程分析

MapReduce詳細工作流程之Map階段

如上圖所示

1. 首先有一個200M的待處理文件
2. 切片：在客戶端提交之前，根據參數配置，進行任務規划，將文件按128M每塊進行切片
3. 提交：提交可以提交到本地工作環境或者Yarn工作環境，本地只需要提交切片信息和xml配置文件，Yarn環境還需要提交jar包；本地環境一般只作為測試用
4. 提交時會將每個任務封裝為一個job交給Yarn來處理（詳細見后邊的Yarn工作流程介紹），計算出MapTask數量（等於切片數量），每個MapTask並行執行
5. MapTask中執行Mapper的map方法，此方法需要k和v作為輸入參數，所以會首先獲取kv值；
   • 首先調用InputFormat方法，默認為TextInputFormat方法，在此方法調用createRecoderReader方法，將每個塊文件封裝為k，v鍵值對，傳遞給map方法
6. map方法首先進行一系列的邏輯操作，執行完成后最后進行寫操作
7. map方法如果直接寫給reduce的話，相當於直接操作磁盤，太多的IO操作，使得效率太低，所以在map和reduce中間還有一個shuffle操作
   • map處理完成相關的邏輯操作之后，首先通過outputCollector向環形緩沖區寫入數據，環形緩沖區主要兩部分，一部分寫入文件的元數據信息，另一部分寫入文件的真實內容
   • 環形緩沖區的默認大小是100M，當緩沖的容量達到默認大小的80%時，進行反向溢寫
8. 在溢寫之前會將緩沖區的數據按照指定的分區規則進行分區和排序，之所以反向溢寫是因為這樣就可以邊接收數據邊往磁盤溢寫數據
9. 在分區和排序之后，溢寫到磁盤，可能發生多次溢寫，溢寫到多個文件
10. 對所有溢寫到磁盤的文件進行歸並排序
11. 在9到10步之間還可以有一個Combine合並操作，意義是對每個MapTask的輸出進行局部匯總，以減少網絡傳輸量
    • Map階段的進程數比Reduce階段要多，所以放在Map階段處理效率更高
    • Map階段合並之后，傳遞給Reduce的數據就會少很多
    • 但是Combiner能夠應用的前提是不能影響最終的業務邏輯，而且Combiner的輸出kv要和Reduce的輸入kv類型對應起來

整個MapTask分為Read階段，Map階段，Collect階段，溢寫（spill）階段和combine階段

• Read階段：MapTask通過用戶編寫的RecordReader，從輸入InputSplit中解析出一個個key/value
• Map階段：該節點主要是將解析出的key/value交給用戶編寫map()函數處理，並產生一系列新的key/value
• Collect收集階段：在用戶編寫map()函數中，當數據處理完成后，一般會調用OutputCollector.collect()輸出結果。在該函數內部，它會將生成的key/value分區（調用Partitioner），並寫入一個環形內存緩沖區中
• Spill階段：即“溢寫”，當環形緩沖區滿后，MapReduce會將數據寫到本地磁盤上，生成一個臨時文件。需要注意的是，將數據寫入本地磁盤之前，先要對數據進行一次本地排序，並在必要時對數據進行合並、壓縮等操作

MapReduce詳細工作流程之Reduce階段

如上圖所示

12. 所有的MapTask任務完成后，啟動相應數量的ReduceTask（和分區數量相同），並告知ReduceTask處理數據的范圍
13. ReduceTask會將MapTask處理完的數據拷貝一份到磁盤中，並合並文件和歸並排序
14. 最后將數據傳給reduce進行處理，一次讀取一組數據
15. 最后通過OutputFormat輸出

整個ReduceTask分為Copy階段，Merge階段，Sort階段（Merge和Sort可以合並為一個），Reduce階段。

• Copy階段：ReduceTask從各個MapTask上遠程拷貝一片數據，並針對某一片數據，如果其大小超過一定閾值，則寫到磁盤上，否則直接放到內存中
• Merge階段：在遠程拷貝數據的同時，ReduceTask啟動了兩個后台線程對內存和磁盤上的文件進行合並，以防止內存使用過多或磁盤上文件過多
• Sort階段：按照MapReduce語義，用戶編寫reduce()函數輸入數據是按key進行聚集的一組數據。為了將key相同的數據聚在一起，Hadoop采用了基於排序的策略。由於各個MapTask已經實現對自己的處理結果進行了局部排序，因此，ReduceTask只需對所有數據進行一次歸並排序即可
• Reduce階段：reduce()函數將計算結果寫到HDFS上

Shuffle機制

Map方法之后，Reduce方法之前的數據處理過程稱之為Shuffle。shuffle流程詳解如下：

1. MapTask收集map()方法輸出的kv對，放到環形緩沖區中
2. 從環形緩沖區不斷溢出到本地磁盤文件，可能會溢出多個文件
3. 多個溢出文件會被合並成大的溢出文件
4. 在溢出過程及合並的過程中，都要調用Partitioner進行分區和針對key進行排序
5. ReduceTask根據自己的分區號，去各個MapTask機器上取相應的結果分區數據
6. ReduceTask將取到的來自同一個分區不同MapTask的結果文件進行歸並排序
7. 合並成大文件后，shuffle過程也就結束了，進入reduce方法

Yarn工作機制

job提交全過程

1. MR程序提交到客戶端所在的節點，YarnRunner向ResourceManager申請一個Application
2. RM將該Application的資源路徑和作業id返回給YarnRunner
3. YarnRunner將運行job所需資源提交到HDFS上
4. 程序資源提交完畢后，申請運行mrAppMaster
5. RM將用戶的請求初始化成一個Task
6. 其中一個NodeManager領取到Task任務
7. 該NodeManager創建容器Container，並產生MRAppmaster
8. Container從HDFS上拷貝資源到本地
9. MRAppmaster向RM 申請運行MapTask資源
10. RM將運行MapTask任務分配給另外兩個NodeManager，另兩個NodeManager分別領取任務並創建容器
11. MR向兩個接收到任務的NodeManager發送程序啟動腳本，這兩個NodeManager分別啟動MapTask，MapTask對數據分區排序
12. MrAppMaster等待所有MapTask運行完畢后，向RM申請容器，運行ReduceTask
13. ReduceTask向MapTask獲取相應分區的數據
14. 程序運行完畢后，MR會向RM申請注銷自己

進度和狀態更新：

YARN中的任務將其進度和狀態(包括counter)返回給應用管理器, 客戶端每秒(通過mapreduce.client.progressmonitor.pollinterval設置)向應用管理器請求進度更新, 展示給用戶

作業完成：

除了向應用管理器請求作業進度外, 客戶端每5秒都會通過調用waitForCompletion()來檢查作業是否完成。時間間隔可以通過mapreduce.client.completion.pollinterval來設置。作業完成之后, 應用管理器和Container會清理工作狀態。作業的信息會被作業歷史服務器存儲以備之后用戶核查

歡迎關注下方公眾號，獲取更多文章信息