MapReduce與Yarn 的詳細工作流程分析
阿新 • • 發佈：2019-10-09
MapReduce詳細工作流程之Map階段
如上圖所示
- 首先有一個200M的待處理檔案
- 切片：在客戶端提交之前，根據引數配置，進行任務規劃，將檔案按128M每塊進行切片
- 提交：提交可以提交到本地工作環境或者Yarn工作環境，本地只需要提交切片資訊和xml配置檔案，Yarn環境還需要提交jar包；本地環境一般只作為測試用
- 提交時會將每個任務封裝為一個job交給Yarn來處理（詳細見後邊的Yarn工作流程介紹），計算出MapTask數量（等於切片數量），每個MapTask並行執行
- MapTask中執行Mapper的map方法，此方法需要k和v作為輸入引數，所以會首先獲取kv值；
- 首先呼叫InputFormat方法，預設為TextInputFormat方法，在此方法呼叫createRecoderReader方法，將每個塊檔案封裝為k，v鍵值對，傳遞給map方法
- map方法首先進行一系列的邏輯操作，執行完成後最後進行寫操作
- map方法如果直接寫給reduce的話，相當於直接操作磁碟，太多的IO操作，使得效率太低，所以在map和reduce中間還有一個shuffle操作
- map處理完成相關的邏輯操作之後，首先通過outputCollector向環形緩衝區寫入資料，環形緩衝區主要兩部分，一部分寫入檔案的元資料資訊，另一部分寫入檔案的真實內容
- 環形緩衝區的預設大小是100M，當緩衝的容量達到預設大小的80%時，進行反向溢寫
- 在溢寫之前會將緩衝區的資料按照指定的分割槽規則進行分割槽和排序，之所以反向溢寫是因為這樣就可以邊接收資料邊往磁碟溢寫資料
- 在分割槽和排序之後，溢寫到磁碟，可能發生多次溢寫，溢寫到多個檔案
- 對所有溢寫到磁碟的檔案進行歸併排序
- 在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階段
如上圖所示
- 所有的MapTask任務完成後，啟動相應數量的ReduceTask（和分割槽數量相同），並告知ReduceTask處理資料的範圍
- ReduceTask會將MapTask處理完的資料拷貝一份到磁碟中，併合並檔案和歸併排序
- 最後將資料傳給reduce進行處理，一次讀取一組資料
- 最後通過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流程詳解如下：
- MapTask收集map()方法輸出的kv對，放到環形緩衝區中
- 從環形緩衝區不斷溢位到本地磁碟檔案，可能會溢位多個檔案
- 多個溢位檔案會被合併成大的溢位檔案
- 在溢位過程及合併的過程中，都要呼叫Partitioner進行分割槽和針對key進行排序
- ReduceTask根據自己的分割槽號，去各個MapTask機器上取相應的結果分割槽資料
- ReduceTask將取到的來自同一個分割槽不同MapTask的結果檔案進行歸併排序
- 合併成大檔案後，shuffle過程也就結束了，進入reduce方法
Yarn工作機制
job提交全過程
- MR程式提交到客戶端所在的節點，YarnRunner向ResourceManager申請一個Application
- RM將該Application的資源路徑和作業id返回給YarnRunner
- YarnRunner將執行job所需資源提交到HDFS上
- 程式資源提交完畢後，申請執行mrAppMaster
- RM將使用者的請求初始化成一個Task
- 其中一個NodeManager領取到Task任務
- 該NodeManager建立容器Container，併產生MRAppmaster
- Container從HDFS上拷貝資源到本地
- MRAppmaster向RM 申請執行MapTask資源
- RM將執行MapTask任務分配給另外兩個NodeManager，另兩個NodeManager分別領取任務並建立容器
- MR向兩個接收到任務的NodeManager傳送程式啟動指令碼，這兩個NodeManager分別啟動MapTask，MapTask對資料分割槽排序
- MrAppMaster等待所有MapTask執行完畢後，向RM申請容器，執行ReduceTask
- ReduceTask向MapTask獲取相應分割槽的資料
- 程式執行完畢後，MR會向RM申請登出自己
進度和狀態更新：
YARN中的任務將其進度和狀態(包括counter)返回給應用管理器, 客戶端每秒(通過mapreduce.client.progressmonitor.pollinterval設定)嚮應用管理器請求進度更新, 展示給使用者
作業完成：
除了嚮應用管理器請求作業進度外, 客戶端每5秒都會通過呼叫waitForCompletion()來檢查作業是否完成。時間間隔可以通過mapreduce.client.completion.pollinterval來設定。作業完成之後, 應用管理器和Container會清理工作狀態。作業的資訊會被作業歷史伺服器儲存以備之後使用者核查
歡迎關注下方公眾號，獲取更多文章資訊
