Map Reduce & YARN

A、簡介

Apache Hadoop 是一個開源軟件框架，可安裝在一個商用機器集群中，使機器可彼此通信并協同工作，以高度分布式的方式共同存儲和處理大量數據。最初，Hadoop 包含以下兩個主要組件：Hadoop Distributed File System (HDFS) 和一個分布式計算引擎，該引擎支持以 MapReduce 作業的形式實現和運行程序。

MapReduce 是 Google 推廣的一個簡單的編程模型，它對以高度并行和可擴展的方式處理大數據集很有用。MapReduce 的靈感來源于函數式編程，用戶可將他們的計算表達為 map 和 reduce 函數，將數據作為鍵值對來處理。Hadoop 提供了一個高級 API 來在各種語言中實現自定義的 map 和 reduce 函數。

Hadoop 還提供了軟件基礎架構，以一系列 map 和 reduce 任務的形式運行 MapReduce 作業。Map 任務 在輸入數據的子集上調用 map 函數。在完成這些調用后，reduce 任務 開始在 map 函數所生成的中間數據上調用 reduce 任務，生成最終的輸出。 map 和 reduce 任務彼此單獨運行，這支持并行和容錯的計算。

最重要的是，Hadoop 基礎架構負責處理分布式處理的所有復雜方面：并行化、調度、資源管理、機器間通信、軟件和硬件故障處理，等等。得益于這種干凈的抽象，實現處理數百(或者甚至數千)個機器上的數 TB 數據的分布式應用程序從未像現在這么容易過，甚至對于之前沒有使用分布式系統的經驗的開發人員也是如此。

B、MR架構

將任務分割為 Map 端和 reduce 端。

一、JobClient JobTracker TaskTracker

• JobClient 向 JobTracker 請求一個新的 jobID
• 檢查作業輸出說明
• 計算作業輸出劃分split
• 將運行作業所需要的資源(作業的jar文件、配置文件、計算所得的輸入劃分)復制到一個以作業ID命名的目錄中JobTracker的文件系統。

通過調用JobTracker的submitJob()方法，告訴JobTracker作業準備執行

JobTracker接收到submitJob()方法調用后，把此調用放到一個內部隊列中，交由作業調度器進行調度，并對其進行初始化

創建運行任務列表，作業調度去首先從共享文件系統中獲取JobClient已經計算好的輸入劃分信息(圖中step6)，然后為每個劃分創建一個Map任務(一個split對應一個map，有多少split就有多少map)。

TaskTracker執行一個簡單的循環，定期發送心跳(heartbeat)調用JobTracker

二、shuffle combine

整體的Shuffle過程包含以下幾個部分：Map端Shuffle、Sort階段、Reduce端Shuffle。即是說：Shuffle 過程橫跨 map 和 reduce 兩端，中間包含 sort 階段，就是數據從 map task 輸出到reduce task輸入的這段過程。

sort、combine 是在 map 端的，combine 是提前的 reduce ，需要自己設置。

Hadoop 集群中，大部分 map task 與 reduce task 的執行是在不同的節點上。當然很多情況下 Reduce 執行時需要跨節點去拉取其它節點上的map task結果。如果集群正在運行的 job 有很多，那么 task 的正常執行對集群內部的網絡資源消耗會很嚴重。而對于必要的網絡資源消耗，最終的目的就是***化地減少不必要的消耗。還有在節點內，相比于內存，磁盤 IO 對 job 完成時間的影響也是可觀的。從最基本的要求來說，對于 MapReduce 的 job 性能調優的 Shuffle 過程，目標期望可以有：

• 完整地從map task端拉取數據到reduce 端。
• 在跨節點拉取數據時，盡可能地減少對帶寬的不必要消耗。
• 減少磁盤IO對task執行的影響。
• 總體來講這段Shuffle過程，能優化的地方主要在于減少拉取數據的量及盡量使用內存而不是磁盤。

三、Map Shuffle

1、輸入

在map task 執行時，其輸入來源 HDFS的 block ，map task 只讀取split 。Split 與 block 的對應關系可能是多對一，默認為一對一。

2、切分

決定于當前的 mapper的 part交給哪個 reduce的方法是：mapreduce 提供的Partitioner接口，對key 進行 hash 后，再以 reducetask 數量取模，然后到指定的 job 上。

然后將數據寫入內存緩沖區中，緩沖區的作用是批量收集map結果，減少磁盤IO的影響。key/value對以及 Partition 的結果都會被寫入緩沖區。寫入之前，key 與value 值都會被序列化成字節數組。

3、溢寫

由于內存緩沖區的大小限制(默認100MB)，當map task輸出結果很多時就可能發生內存溢出，所以需要在一定條件下將緩沖區的數據臨時寫入磁盤，然后重新利用這塊緩沖區。這個從內存往磁盤寫數據的過程被稱為Spill，中文可譯為溢寫。

這個溢寫是由另外單獨線程來完成，不影響往緩沖區寫map結果的線程。

整個緩沖區有個溢寫的比例spill.percent。這個比例默認是0.8，

Combiner 將有相同key的 key/value 對加起來，減少溢寫spill到磁盤的數據量。Combiner的適用場景：由于Combiner的輸出是Reducer的輸入，Combiner絕不能改變最終的計算結果。故大多數情況下，combiner適用于輸入輸出的key/value類型完全一致，且不影響最終結果的場景(比如累加、***值等……)。

4、Merge

map 很大時，每次溢寫會產生一個 spill_file，這樣會有多個 spill_file，而最終的輸出只有一個文件，在最終輸出之前會對多個中間過程多次產生的溢寫文件 spill_file 進行合并，此過程就是 merge。

merge 就是把相同 key 的結果加起來。(當然，如果設置過combiner，也會使用combiner來合并相同的key)

四、Reduce Shuffle

1、reduce shuffle

在 reduce task 之前，不斷拉取當前 job 里每個 maptask 的最終結果，然后對從不同地方拉取過來的數據不斷地做 merge ，也最終形成一個文件作為 reduce task 的輸入文件。

2、copy

Reduce進程啟動一些數據copy線程(Fetcher)，通過HTTP方式請求map task所在的TaskTracker獲取map task的輸出文件。因為maptask早已結束，這些文件就歸TaskTracker管理在本地磁盤中。

3、merge

Copy 過來的數據會先放入內存緩沖區中，這里的緩沖區大小要比 map 端的更為靈活，它基于 JVM 的 heap size 設置，因為 Shuffle 階段 Reducer 不運行，所以應該把絕大部分的內存都給 Shuffle 用。這里需要強調的是，merge 有三種形式：1)內存到內存 2)內存到磁盤 3)磁盤到磁盤。默認情況下***種形式不啟用，讓人比較困惑，是吧。當內存中的數據量到達一定閾值，就啟動內存到磁盤的 merge 。與 map 端類似，這也是溢寫的過程，這個過程中如果你設置有Combiner，也是會啟用的，然后在磁盤中生成了眾多的溢寫文件。第二種merge方式一直在運行，直到沒有 map 端的數據時才結束，然后啟動第三種磁盤到磁盤的 merge 方式生成最終的那個文件。

4、reducer的輸入

merge 的***會生成一個文件，大多數情況下存在于磁盤中，但是需要將其放入內存中。當reducer 輸入文件已定，整個 Shuffle 階段才算結束。然后就是 Reducer 執行，把結果放到 HDFS 上。

C、YARN

YARN(Yet Another Resource Negotiator),下一代MapReduce框架的名稱，為了容易記憶，一般稱為MRv2(MapReduce version 2)。該框架已經不再是一個傳統的MapReduce框架，甚至與MapReduce無關，她是一個通用的運行時框架，用戶可以編寫自己的計算框架，在該運行環境中運行。用于自己編寫的框架作為客戶端的一個lib，在運用提交作業時打包即可。

五、why YARN instead of MR

MR 的缺點

經典 MapReduce 的最嚴重的限制主要關系到可伸縮性、資源利用和對與 MapReduce 不同的工作負載的支持。在 MapReduce 框架中，作業執行受兩種類型的進程控制：

• 一個稱為 JobTracker 的主要進程，它協調在集群上運行的所有作業，分配要在 TaskTracker 上運行的 map 和 reduce 任務。
• 許多稱為 TaskTracker 的下級進程，它們運行分配的任務并定期向 JobTracker 報告進度。
• 大型的 Hadoop 集群顯現出了由單個 JobTracker 導致的可伸縮性瓶頸。

此外，較小和較大的 Hadoop 集群都從未***效地使用他們的計算資源。在 Hadoop MapReduce 中，每個從屬節點上的計算資源由集群管理員分解為固定數量的 map 和 reduce slot，這些 slot 不可替代。設定 map slot 和 reduce slot 的數量后，節點在任何時刻都不能運行比 map slot 更多的 map 任務，即使沒有 reduce 任務在運行。這影響了集群的利用率，因為在所有 map slot 都被使用(而且我們還需要更多)時，我們無法使用任何 reduce slot，即使它們可用，反之亦然。

Hadoop 設計為僅運行 MapReduce 作業。隨著替代性的編程模型(比如 Apache Giraph 所提供的圖形處理)的到來，除 MapReduce 外，越來越需要為可通過高效的、公平的方式在同一個集群上運行并共享資源的其他編程模型提供支持。

• 原MapReduce框架的不足
• JobTracker是集群事務的集中處理點，存在單點故障
• JobTracker需要完成的任務太多，既要維護job的狀態又要維護job的task的狀態，造成過多的資源消耗
• 在taskTracker端，用map/reduce task作為資源的表示過于簡單，沒有考慮到CPU、內存等資源情況，當把兩個需要消耗大內存的task調度到一起，很容易出現OOM
• 把資源強制劃分為map/reduce slot,當只有map task時，reduce slot不能用;當只有reduce task時，map slot不能用，容易造成資源利用不足。
• 解決可伸縮性問題

在 Hadoop MapReduce 中，JobTracker 具有兩種不同的職責：

管理集群中的計算資源，這涉及到維護活動節點列表、可用和占用的 map 和 reduce slots 列表，以及依據所選的調度策略將可用 slots 分配給合適的作業和任務

協調在集群上運行的所有任務，這涉及到指導 TaskTracker 啟動 map 和 reduce 任務，監視任務的執行，重新啟動失敗的任務，推測性地運行緩慢的任務，計算作業計數器值的總和，等等

為單個進程安排大量職責會導致重大的可伸縮性問題，尤其是在較大的集群上，JobTracker 必須不斷跟蹤數千個 TaskTracker、數百個作業，以及數萬個 map 和 reduce 任務。相反，TaskTracker 通常近運行十來個任務，這些任務由勤勉的 JobTracker 分配給它們。

為了解決可伸縮性問題，一個簡單而又絕妙的想法應運而生：我們減少了單個 JobTracker 的職責，將部分職責委派給 TaskTracker，因為集群中有許多 TaskTracker。在新設計中，這個概念通過將 JobTracker 的雙重職責(集群資源管理和任務協調)分開為兩種不同類型的進程來反映。

六、YARN 的優點

• 更快地MapReduce計算
• 對多框架支持
• 框架升級更容易

YARN

• ResourceManager 代替集群管理器
• ApplicationMaster 代替一個專用且短暫的 JobTracker
• NodeManager 代替 TaskTracker
• 一個分布式應用程序代替一個 MapReduce 作業

一個全局 ResourceManager 以主要后臺進程的形式運行，它通常在專用機器上運行，在各種競爭的應用程序之間仲裁可用的集群資源。

在用戶提交一個應用程序時，一個稱為 ApplicationMaster 的輕量型進程實例會啟動來協調應用程序內的所有任務的執行。這包括監視任務，重新啟動失敗的任務，推測性地運行緩慢的任務，以及計算應用程序計數器值的總和。有趣的是，ApplicationMaster 可在容器內運行任何類型的任務。

NodeManager 是 TaskTracker 的一種更加普通和高效的版本。沒有固定數量的 map 和 reduce slots，NodeManager 擁有許多動態創建的資源容器。