?一、什么是MapReduce

二、MapReduce 的設計思想

?2.1 分而治之

?2.2 構建抽象模型：Map和Reduce

?2.3 隱藏系統層細節

三、MapReduce 的框架原理

?3.1 MRv1工作原理

?3.1.1 MRv1架構工作原理圖

?3.1.1.1 流程說明

?3.1.1.1.1 作業的提交

?3.1.1.1.2 作業的初始化

?3.1.1.1.3 任務的分配

?3.1.1.1.4 任務的執行

?3.1.1.1.5 進度和狀態的更新

?3.1.1.1.6 作業的完成

?3.1.1.2 組件說明

?3.1.1.2.1 Mapper和Reducer

?3.1.1.2.2 JobTracker

?3.1.1.2.3 TaskTracker

?3.1.1.2.4 JobClient

?3.1.1.2.5 MapTask和ReduceTask

?3.1.1.2.5.1 MapTask工作機制

?3.1.1.2.5.2 ReduceTask工作機制

3.1.2 MapReduce工作原理圖（邏輯實體角度）

?3.1.2.1 流程說明

?3.1.2.1.1 split 階段

?3.1.2.1.2 map 階段

?3.1.2.1.3 Shuffle 階段

?3.1.2.1.4 Reduce 階段

?四、MapReduce 的基本使用案例

?4.1 MapReduce編程規范

?4.1.1 編寫 Mapper 類

?4.1.2 編寫 Reducer 類

?4.1.3 Driver 階段

?4.2 案例說明（wordcount案例）

? ? ? ?? 4.2.1 split（分割）

4.2.2 執行Map方法?

4.2.3 排序及Combine

4.2.4 執行Reduce方法?

?五、性能優化

?5.1 Mapreduce 性能影響因素分析

?5.1.1 計算機性能

?5.1.2 I/O 操作優化

?5.1.2.1 數據傾斜

?5.1.2.2 map 和 reduce 數設置不合理

?5.1.2.3 map 運行時間太長，導致 reduce 等待過久

?5.1.2.4 小文件過多

?5.1.2.5 大量的不可分塊的超大文件

?5.1.2.6 spill 次數過多

?5.1.2.7 merge 次數過多等

?5.2 優化方法

?5.2.1 數據輸入

?5.1.2 Map 階段

?5.1.3 Reduce 階段

?5.1.4 數據傾斜問題

?3.1.5 常用的調優參數

?一、什么是MapReduce

MapReduce是一個用于大規模數據處理的分布式計算模型，最初由Google工程師設計并實現的，Google已經將完整的MapReduce論文公開發布了。其中的定義是，MapReduce是一個編程模型，是一個用于處理和生成大規模數據集的相關的實現。用戶定義一個map函數來處理一個Key-Value對以生成一批中間的Key-Value對，再定義一個reduce函數將所有這些中間的有相同Key的Value合并起來。很多現實世界中的任務都可用這個模型來表達。

二、MapReduce 的設計思想

?2.1 分而治之

簡化并行計算的編程模型

?2.2 構建抽象模型：Map和Reduce

開發人員專注于實現Mapper和Reducer函數

?2.3 隱藏系統層細節

開發人員專注于業務邏輯實現

三、MapReduce 的框架原理

?3.1 MRv1工作原理

?3.1.1 MRv1架構工作原理圖

?3.1.1.1 流程說明

?3.1.1.1.1 作業的提交

JobClient的submitJob()方法實現的作業提交過程，如下所示：

1）通過JobTracker的getNewJobId()方法，向jobtracker請求一個新的作業ID。參見步驟2。

2）檢查作業的輸出說明，也就是說要指定輸出目錄的路徑，但是輸出目錄還不能存在(防止覆蓋輸出結果)，如果不滿足條件，就會將錯誤拋給MapReduce程序。

3）檢查作業的輸入說明，也就是說如果輸入路徑不存在，作業也沒法提交，如果不滿足條件，就會將錯誤拋給MapReduce程序。

4）將作業運行所需的資源，比如作業JAR文件、配置文件等復制到HDFS中。參見步驟3。

5）通過JobTracker的submitJob()方法，告訴jobtracker作業準備執行。參見步驟4。

?3.1.1.1.2 作業的初始化

1）JobTracker接收到對其submitJob()方法調用之后，就會把此調用放入一個內部隊列當中，交由作業調度器進行調度。(說明：Hadoop作業的調度器常見的有3個：先進先出調度器；容量調度器；公平調度器。Hadoop作業調度器采用的是插件機制，即作業調度器是動態加載的、可插拔的，同時第三方可以開發自己的作業調度器。參見步驟5。

2）初始化包括創建一個表示正在運行作業的對象——封裝任務的記錄信息，以便跟蹤任務的狀態和進程。參見步驟5。

3）接下來要創建運行任務列表，作業調度器首先從共享文件系統中獲取JobClient已計算好的輸入分片信息，然后為每個分片創建一個map任務(也就是說mapper的個數與分片的數目相同)。參見步驟6。(創建reduce任務的數量由JobConf的mapred.reduce.task屬性決定，它是用setNumReduceTasks()方法來設置的，然后調度器創建相應數量的要運行的reduce任務，默認情況只有一個reducer)

?3.1.1.1.3 任務的分配

1）tasktracker本身運行一個簡單的循環來定期發送”心跳(heartbeat)”給jobtracker。什么是心跳呢？就是tasktracker告訴jobtracker它是否還活著，同時心跳也充當兩者之間的消息通信，比如tasktracker會指明它是否已經做好準備來運行新的任務了，如果是，管理者jobtracker就會給執行者tasktracker分配一個任務。參見步驟7。

2）當然，在管理者jobtracker為執行者tasktracker選擇任務之前，jobtracker必須先選定任務所在的作業。一旦選擇好作業，jobtracker就可以給tasktracker選定一個任務。如何選擇一個作業呢？當然是Hadoop作業的調度器了，它就像是Hadoop的中樞神經系統一樣，默認的方法是簡單維護一個作業優先級列表。(對于調度算法的更深理解可以學習操作系統的作業調度算法，進程調度算法，比如先來先服務(FCFS)調度算法，短作業優先(SJF)調度算法，優先級調度算法，高響應比優先調度算法，時間片輪轉調度算法，多級反饋隊列調度算法等。如果從更高的角度來看調度算法，其實是一種控制和決策的策略選擇。)

?3.1.1.1.4 任務的執行

1）作業選擇好了，任務也選擇好了，接下來要做的事情就是任務的運行了。首先，從HDFS中把作業的JAR文件復制到tasktracker所在的文件系統，同時，tasktracker將應用程序所需要的全部文件從分布式緩存復制到本地磁盤，也就是從HDFS文件系統復制到ext4等文件系統之中。參見步驟8。

2）tasktracker為任務新建一個本地工作目錄，并把JAR文件中的內容解壓到這個文件夾中，新建一個TaskRunner實例來運行該任務。

3）TaskRunner啟動一個新的JVM(參見步驟9)來運行每個任務(參見步驟10)，以便用戶定義的map和reduce函數的任何缺陷都不會影響TaskTracker守護進程(比如導致它崩潰或者掛起)。需要說明一點的是，對于map和reduce任務，tasktracker有固定數量的任務槽，準確數量由tasktracker核的數量和內存大小來決定，比如一個tasktracker可能同時運行兩個map任務和reduce任務。map任務和reduce任務中關于數據本地化部分不再講解，因為DRCP沒有用到，只要理解本地數據級別就可以了，比如node-local，rack-local，off-switch。

4）子進程通過umbilical接口與父進程進行通信，任務的子進程每隔幾秒便告訴父進程它的進度，直到任務完成。

?3.1.1.1.5 進度和狀態的更新

1）MapReduce是Hadoop的一個離線計算框架，運行時間范圍從數秒到數小時，因此，對于我們而言直到作業進展是很重要的。

2）一個作業和每個任務都有一個狀態信息，包括作業或任務的運行狀態(比如，運行狀態，成功完成，失敗狀態)、Map和Reduce的進度、計數器值、狀態消息和描述(可以由用戶代碼來設置)等。

3）這些消息通過一定的時間間隔由Child JVM—>TaskTracker—>JobTracker匯聚。JobTracker將產生一個表明所有運行作業及其任務狀態的全局視圖。可以通過Web UI查看。同時JobClient通過每秒查詢JobTracker來獲得最新狀態，輸出到控制臺上。

4）現在可能會有一個疑問，這些狀態信息在作業執行期間不斷變化，它們是如何與客戶端進行通信的呢？詳細細節不在講解，參考資料《Hadoop權威指南》。

?3.1.1.1.6 作業的完成

1）當jobtracker收到作業最后一個任務已完成的通知后，便把作業的狀態設置為”成功”。然后，在JobClient查詢狀態時，便知道作業已成功完成，于是JobClient打印一條消息告知用戶，最后從runJob()方法返回。

說明：

MapReduce容錯，即作業失敗情況不再講解，參考資料《Hadoop權威指南》。

?3.1.1.2 組件說明

?3.1.1.2.1 Mapper和Reducer

運行在Hadoop上的MapReduce應用程序最基本的組成部分包括：一是Mapper抽象類，一是Reducer抽象類，一是創建JobConf的執行程序。

?3.1.1.2.2 JobTracker

JobTracker是一個master服務，軟件啟動之后JobTracker接收Job，負責調度Job的每一個子任務Task運行于TaskTracker上，并且監控它們的運行，如果發現有失敗的Task就重新運行它，一般情況下應該把JobTracker部署在單獨的機器上。

?3.1.1.2.3 TaskTracker

TaskTracker是運行在多個節點上的slaver服務。TaskTracker主動與JobTracker通信(與DataNode和NameNode相似，通過心跳來實現)接收作業，并負責直接執行每一個任務。

?3.1.1.2.4 JobClient

每一個Job都會在用戶端通過JobClient類將應用程序以及配置參數Configuration打包成JAR文件存儲在HDFS中，并把路徑提交到JobTracker的master服務，然后由master創建每一個Task(即MapTask和ReduceTask)將它們分發到各個TaskTracker服務中去執行。

?3.1.1.2.5 MapTask和ReduceTask

一個完整的Job會自動依次執行Mapper、Combiner(在JobConf指定Combiner時執行)和Reducer，其中Mapper和Combiner是由MapTask調用執行，Reduce則由ReduceTask調用，Combiner實際也是Reducer接口類的實現。Mapper會根據Job JAR中定義的輸入數據集<key1, value1>對讀入，處理完成生成臨時的<key2, value2>對，如果定義了Combiner，MapTask會在Mapper完成調用該Combiner將相同Key的值做合并處理，以減少輸出結果集。MapTask的任務全部完成后，交給ReduceTask進程調用Reducer處理，生成最終結果<Key3, value3>對。

?3.1.1.2.5.1 MapTask工作機制

1. 并行度決定機制

1）問題引出
maptask 的并行度決定 map 階段的任務處理并發度，進而影響到整個 job的處理速度。
那么，mapTask 并行任務是否越多越好呢？

2）MapTask 并行度決定機制
一個 job 的 map 階段 MapTask 并行度（個數），由客戶端提交 job 時的切片個數決定。

2. MapTask工作機制

1）Read 階段：Map Task 通過用戶編寫的 RecordReader，按照 InputSplit 記錄的位置信息讀取數據，從中解析出一個個 `<Key,Value>`。

2）Map 階段：將解析出的 key/value 交給用戶編寫 map()函數處理，并產生一系列新的 key/value。

3）Collect 收集階段：在用戶編寫 map()函數中，當數據處理完成后，一般會調用 OutputCollector.collect()輸出結果。在該函數內部，它會將生成的 key/value
分區（調用 Partitioner），并寫入一個環形內存緩沖區中。

4）Spill 階段：即 **溢寫** ，當環形緩沖區滿后，MapReduce 會將數據寫到本地磁盤上，生成一個臨時文件。需要注意的是，將數據寫入本地磁盤之前，先要對數據進行一次本地排序，并在必要時對數據進行合并、壓縮等操作。
溢寫階段詳情：
步驟 1：利用快速排序算法對緩存區內的數據進行排序，排序方式是，先按照分區編號 partition 進行排序，然后按照 key 進行排序。這樣，經過排序后，數據以分區為單位聚集在一起，且同一分區內所有數據按照 key 有序。
步驟 2：按照分區編號由小到大依次將每個分區中的數據寫入任務工作目錄下的臨時文件 output/spillN.out（N 表示當前溢寫次數）中。如果用戶設置了Combiner，則寫入文件之前，對每個分區中的數據進行一次聚集操作。
步驟 3：將分區數據的元信息寫到內存索引數據結構 SpillRecord 中，其中每個分區的元信息包括在臨時文件中的偏移量、壓縮前數據大小和壓縮后數據大小。如果當前內存索引大小超過 1MB ，則將內存索引寫到文件output/spillN.out.index 中。

5）Combine 階段：當所有數據處理完成后，MapTask 對所有臨時文件進行一次合并，以確保最終只會生成一個數據文件。
當所有數據處理完后，MapTask 會將所有臨時文件合并成一個大文件，并保存到文件 output/file.out 中，同時生成相應的索引文件 output/file.out.index。

在進行文件合并過程中，MapTask 以分區為單位進行合并。對于某個分區，它將采用多輪遞歸合并的方式。每輪合并 io.sort.factor（默認 100）個文件，并將產生的文件重新加入待合并列表中，對文件排序后，重復以上過程，直到最終得到一個大文件。

讓每個 MapTask 最終只生成一個數據文件，可避免同時打開大量文件和同時讀取大量小文件產生的隨機讀取帶來的開銷。

?3.1.1.2.5.2 ReduceTask工作機制

1.設置 ReduceTask 并行度（個數）
reducetask 的并行度同樣影響整個 job 的執行并發度和執行效率，但與maptask 的并發數由切片數決定不同，Reducetask 數量的決定是可以直接手動設置：

//默認值是 1，手動設置為 5
job.setNumReduceTasks(5);

2.注意
（1）reducetask=0 ，表示沒有 reduce 階段，輸出文件個數和 map 個數一致。
（2）reducetask 默認值就是 1，所以輸出文件個數為一個。
（3）如果數據分布不均勻，就有可能在 reduce 階段產生數據傾斜
（4）reducetask 數量并不是任意設置，還要考慮業務邏輯需求，有些情況下，需要計算全局匯總結果，就只能有 1 個 reducetask。
（5）具體多少個 reducetask，需要根據集群性能而定。
（6）如果分區數不是 1，但是 reducetask 為1，是否執行分區過程。答案是：不執行分區過程。因為在maptask的源碼中，執行分區的前提是先判斷reduceNum個數是否大于 1。不大于 1 肯定不執行。

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

3.1.2 MapReduce工作原理圖（邏輯實體角度）

?3.1.2.1 流程說明

?3.1.2.1.1 split 階段

首先 mapreduce 會根據要運行的大文件來進行 split，每個輸入分片(input split)針對一個 map 任務，輸入分片(InputSplit)存儲的并非數據本身，而是一個分片長度和一個記錄數據位置的數組。輸入分片(InputSplit)通常和 HDFS 的 block(塊)關系很密切，假如我們設定 HDFS 的塊的大小是 128MB，我們運行的大文件是128x10MB，MapReduce 會分為 10 個 MapTask，每個 MapTask 都盡可能運行在block(塊)所在的 DataNode 上，體現了移動計算不移動數據的思想。

?3.1.2.1.2 map 階段

map 階段就是執行自己編寫的 Mapper 類中的 map 函數，Map 過程開始處理，MapTask 會接受輸入分片，通過不斷的調用 map()方法對數據進行處理。處理完畢后，轉換為新的 `<KEY,VALUE>`鍵值對輸出。

?3.1.2.1.3 Shuffle 階段

shuffle 階段主要負責將 map 端生成的數據傳遞給 reduce 端，因此 shuffle 分為在 map 端的過程和在 reduce 端的執行過程。具體過程如下：
（1）MapTask 收集 map()方法的輸出<KEY,VALUE>對，放到內存緩沖區（稱為環形緩沖區）中，其中環形緩沖區的大小默認是 100MB。
（2）環形緩沖區到達一定閾值（環形緩沖區大小的 80%）時，會將緩沖區中的數據溢出本地磁盤文件，這個過程中可能會溢出多個文件。
（3）多個溢出文件會被合并成大的溢出文件。
（4）在溢出過程及合并的過程中，都要調用 Partitioner 進行分區和針對 key進行排序 sort。
（5）合并成大文件后，Map 端 shuffle 的過程也就結束了，后面進入 reduce端 shuffle 的過程。
（6）在 Reduce 端，shuffle 主要分為復制 Map 輸出(copy)、排序合并（Merge Sort）兩個階段。

Reduce 任務通過 HTTP 向各個 Map 任務拖取它所需要的數據。
Copy 過來的數據會先放入內存緩沖區中，如果內存緩沖區中能放得下這次數據的話就直接把數據寫到內存中，即內存到內存 merge。Reduce 要向每個 Map去拖取數據，在內存中每個 Map 對應一塊數據，當內存緩存區中存儲的 Map 數據占用空間達到一定程度的時候，開始啟動內存中 merge，把內存中的數據 merge輸出到磁盤上一個文件中，即內存到磁盤 merge。

當屬于該 reducer 的 map 輸出全部拷貝完成，則會在 reducer 上生成多個文件（如果拖取的所有 map 數據總量都沒有超出內存緩沖區，則數據就只存在于內存中），這時開始執行合并操作，即磁盤到磁盤 merge。

?3.1.2.1.4 Reduce 階段

Reduce 從合并的文件中取出一個一個的鍵值對 group，調用用戶自定義的 `reduce()`方法，生成最終的輸出文件。

注意:?Shuffle中的緩沖區大小會影響到MapReduce程序的執行效率，原則上說，緩沖區越大，磁盤io的次數越少，執行速度就越快。
緩沖區的大小可以通過參數調整，參數：`io.sort.mb` 默認 100M。

?四、MapReduce 的基本使用案例

?4.1 MapReduce編程規范

需要重點明確兩點：

1. 一個記錄調用一次 map()方法。
2. 相同的 key 調用一次 reduce()方法。

?4.1.1 編寫 Mapper 類

（1）用戶自定義的 Mapper 要繼承框架提供的 Mapper 類。

（2）Mapper 的輸入數據是 KV 鍵值對的形式（KV 的類型可自定義）。

（3）對數據的處理邏輯寫在 Mapper 類中 map()方法中。

（4）Mapper 的輸出數據是 KV 鍵值對的形式（KV 的類型可自定義）。

（5）map()方法（maptask 進程）每一個<K,V>數據執行一次。

?4.1.2 編寫 Reducer 類

（1）用戶自定義的 Reducer 要繼承框架提供的 Reducer 父類。

（2）Reducer 的輸入數據類型對應 Mapper 的輸出數據類型，也是 KV。

（3）Reducer 的業務邏輯寫在 reduce()方法中。

（4）每一組相同 k 的<k,Iterator>組調用一次 reduce()方法。

?4.1.3 Driver 階段

整個程序需要編寫一個 Driver 來進行提交，將自定義 Mapper 和 Reducer 類

組合成一個 job，并提交 job 對象

?4.2 案例說明（wordcount案例）

?4.2.1 split（分割）

首先，Map階段框架會將用戶輸入分割成固定大小的片段，隨后將每個片段進一步分解成一批鍵值對作為map函數的輸入

4.2.2 執行Map方法?

?4.2.3 排序及Combine

4.2.4 執行Reduce方法?

?五、性能優化

?5.1 Mapreduce 性能影響因素分析

?5.1.1 計算機性能

CPU、內存、磁盤健康、網絡

?5.1.2 I/O 操作優化

?5.1.2.1 數據傾斜

?5.1.2.2 map 和 reduce 數設置不合理

?5.1.2.3 map 運行時間太長，導致 reduce 等待過久

?5.1.2.4 小文件過多

?5.1.2.5 大量的不可分塊的超大文件

?5.1.2.6 spill 次數過多

?5.1.2.7 merge 次數過多等

?5.2 優化方法

?5.2.1 數據輸入

合并小文件：在執行 mr 任務前將小文件進行合并，大量的小文件會產生大量的 map 任務，增大 map 任務裝載次數，而任務的裝載比較耗時，從而導致mr 運行較慢。

?5.1.2 Map 階段

1）減少溢寫（spill）次數：通過調整 `io.sort.mb` 及 `sort.spill.percent`參數值，增大觸發 spill 的內存上限，減少 spill 次數，從而減少磁盤 IO。

2）減少合并（merge）次數：通過調整 `io.sort.factor` 參數，增大 merge 的文件數目，減少 merge 的次數，從而縮短 mr 處理時間。

3）在 map 之后，不影響業務邏輯前提下，先進行 combine 處理，減少 I/O。

?5.1.3 Reduce 階段

1）合理設置 map 和 reduce 數：兩個都不能設置太少，也不能設置太多。太少，會導致 task 等待，延長處理時間；太多，會導致 map、reduce 任務間競爭資源，造成處理超時等錯誤。

2）設置 map、reduce 共存：調整 `slowstart.completedmaps` 參數，使 map 運行到一定程度后，reduce 也開始運行，減少 reduce 的等待時間。

3）使用 reduce：因為 reduce 在用于搜集數據集的時候將會產生大量的網絡消耗。

4）合理設置 reduce 端的 buffer：默認情況下，數據達到一個閾值的時候，buffer中的數據就會寫入磁盤，然后 reduce 會從磁盤中獲得所有的數據。也就是說，buffer 和reduce 是沒有直接關聯的，中間多個一個寫磁盤->讀磁盤的過程，既然有這個弊端，那么就可以通過參數來配置，使得 buffer 中的一部分數據可以直接輸送到reduce，從而減少IO開銷：`mapred.job.reduce.input.buffer.percent`，默認為0.0。當值大于0的時候，會保留指定比例的內存讀buffer中的數據直接拿給reduce使用。這樣一來，設置 buffer 需要內存，讀取數據需要內存，reduce 計算也要內存，所以要根據作業的運行情況進行調整。

?5.1.4 數據傾斜問題

1）數據傾斜現象
數據頻率傾斜——某一個區域的數據量要遠遠大于其他區域。
數據大小傾斜——部分記錄的大小遠遠大于平均值。

2）如何收集傾斜數據
在 reduce 方法中加入記錄 map 輸出鍵的詳細情況的功能。

3）減少數據傾斜的方法
方法 1：抽樣和范圍分區
可以通過對原始數據進行抽樣得到的結果集來預設分區邊界值。
方法 2：自定義分區
基于輸出鍵的背景知識進行自定義分區。例如，如果 map 輸出鍵的單詞來源于一本書。且其中某幾個專業詞匯較多。那么就可以自定義分區將這這些專業詞匯發送給固定的一部分 reduce 實例。而將其他的都發送給剩余的 reduce 實例。
方法 3：Combine
使用 Combine 可以大量地減小數據傾斜。在可能的情況下，combine 的目的就是提前聚合并精簡數據。
方法 4：采用 Map Join，盡量避免 Reduce Join。