架構精講：Hadoop技術框架和架構演進方向

科技 05-15

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 數據的分散式應用程序從未像現在這麼容易過，甚至對於之前沒有使用分散式系統的經驗的開發人員也是如此。

MR 架構

Map Reduce 過程圖

將任務分割為Map端和Reduce 端，下圖是JobClient、JobTracker和TaskTracker架構。

1、JobClient 向 JobTracker 請求一個新的JobID

2、檢查作業輸出說明

3、計算作業輸出劃分split

4、將運行作業所需要的資源(作業的 jar 文件、配置文件、計算所得的輸入劃分)複製到一個以作業 ID 命名的目錄中JobTracker 的文件系統。

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

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

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

8、TaskTracker 執行一個簡單的循環，定期發送心跳調用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、切分

決定於當前的Mpper的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 類型完全一致，且不影響最終結果的場景(比如累加、最大值等)。

Merge

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

Merge 就是把相同Key 的結果加起來(當然，如果設置過Combiner，也會使用 combiner 來合併相同的Key)。

Reduce Shuffle

在Reduce Task 之前，不斷拉取當前Job 里每個Maptask的最終結果，然後對從不同地方拉取過來的數據不斷地做Merge ，也最終形成一個文件作為Reduce Task 的輸入文件。

1、Copy

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

2、Merge

Copy 過來的數據會先放入內存緩衝區中，這裡的緩衝區大小要比 map 端的更為靈活，它基於JVM的 heap size設置，因為Shuffle 階段 Reducer不運行，所以應該把絕大部分的內存都給 Shuffle 用。這裡需要強調的是，Merge有三種形式:

1) 內存到內存

2) 內存到磁碟

3) 磁碟到磁碟

默認情況下第一種形式不啟用，讓人比較困惑，是吧。當內存中的數據量到達一定閾值，就啟動內存到磁碟的Merge 。與Map端類似，這也是溢寫的過程，這個過程中如果你設置有Combiner，也是會啟用的，然後在磁碟中生成了眾多的溢寫文件。第二種Merge 方式一直在運行，直到沒有Map 端的數據時才結束，然後啟動第三種磁碟到磁碟的Merge 方式生成最終的那個文件。

3、Reducer的輸入

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

YARN

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

MR的缺點(YARN取代MR)

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

一個稱為JobTracker 的主要進程，它協調在集群上運行的所有作業，分配要在 TaskTracker上運行的Map和Reduce 任務。

許多稱為TaskTracker 的下級進程，它們運行分配的任務並定期向 JobTracker報告進度。

大型的Hadoop 集群顯現出了由單個JobTracker 導致的可伸縮性瓶頸。

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

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的優點

1、更快地 MapReduce 計算

2、對多框架支持

3、框架升級更容易