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

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

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 過程包含以下幾個部分: 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 過程，能最佳化的地方主要在於減少拉取資料的量及儘量使用記憶體而不是磁碟。

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

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

      經典 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 外，越來越需要為可透過高效的、公平的方式在同一個叢集上執行並共享資源的其他程式設計模型提供支援。

      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 的雙重職責(叢集資源管理和任務協調)分開為兩種不同型別的行程來反映。

• 1、更快地 MapReduce 計算

• 2、對多框架支援

• 3、框架升級更容易

• 1、ResourceManager代替叢集管理器

• 2、ApplicationMaster代替一個專用且短暫的JobTracker

• 3、NodeManager代替TaskTracker

• 4、一個分散式應用程式代替一個MapReduce作業

      一個全域性 ResourceManager以主要後臺行程的形式執行，它通常在專用機器上執行，在各種競爭的應用程式之間仲裁可用的叢集資源。

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

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

連結: http://www.jianshu.com/p/c97ff0ab5f49