. MapReduce基本編程模型和框架

1.1 MapReduce抽象模型

大數據計算的核心思想是：分而治之。如下圖1所示。把大量的數據劃分開來，分配給各個子任務來完成。再將結果合並到一起輸出。
註：如果數據的耦合性很高，不能分離，那麽這種並行計算就不適合了。

圖1：MapReduce抽象模型

1.2 Hadoop的MapReduce的並行編程模型

如下圖2所示，Hadoop的MapReduce先將數據劃分為多個key/value鍵值對。然後輸入Map框架來得到新的key/value對，這時候只是中間結果，這個時候的value值是個值集合。再通過同步障（為了等待所有的Map處理完），這個階段會把相同key的值收集整理（Aggregation&Shuffle）在一起，再交給Reduce框架做輸出組合，如圖2中每個Map輸出的結果，有k1，k2，k3，通過同步障後，k1收集到一起，k2收集到一起，k3收集到一起，再分別交給Reduce，通過Reduce組合結果。

1.3 Hadoop的MapReduce的完整編程模型和框架

圖3是MapReduce的完整編程模型和框架，比模型上多加入了Combiner和Partitioner。
1）Combiner
Combiner可以理解為一個小的Reduce，就是把每個Map的結果，先做一次整合。例如圖3中第三列的Map結果中有2個good，通過Combiner之後，先將本地的2個good組合到了一起（紅色的（good，2））。好處是大大減少需要傳輸的中間結果數量量，達到網絡數據傳輸優化，這也是Combiner的主要作用。
2）Partitioner
為了保證所有的主鍵相同的key值對能傳輸給同一個Reduce節點，如圖3中所有的good傳給第一個Reduce前，所有的is和has傳給第二個Reduce前，所有的weather，the和today傳到第三個Reduce前。MapReduce專門提供了一個Partitioner類來完成這個工作，主要目的就是消除數據傳入Reduce節點後帶來不必要的相關性。

2.Hadoop系統架構

圖4是Hadoop系統的基本組成框架。從邏輯上看，Hadoop系統的基本組成架構包括分布式存儲和並行計算兩部分。
1）分布式存儲框架（分布式文件系統HDFS）
Hadoop使用NameNode作為分布式存儲的主控節點，用以存儲和管理分布式文件系統的元數據，同時使用DataNode作為實際存儲大規模數據從節點。
2）並行計算框架（MapReduce）
Hadoop使用JobTracker作為MapReduce框架的主控節點，用來管理和調度作業的執行，用TaskTracker管理每個計算從節點上任務的執行。

為了實現Hadoop設計的本地化計算，數據節點DataNode和計算節點TaskTracker將放在同個節點，每個從節點也是同時運行DataNode和TaskTracker，從而讓每個TaskTracker盡量處理存儲在本地DataNode上的數據。
數據主控節點NameNode與作業執行節點JobTracker即可以設置在同一個節點上，也可以考慮負載較高時，而設置在兩個節點上。

3.Hadoop MapReduce程序執行過程

圖5：Hadoop MapReduce程序執行過程

MapReduce的整個工作過程如上圖所示，它包含如下4個獨立的實體：

　　1）客戶端，用來提交MapReduce作業。

　　2）jobtracker，用來協調作業的運行。

　　3）tasktracker，用來處理作業劃分後的任務。

　　4）HDFS，用來在其它實體間共享作業文件。

　　MapReduce整個工作過程有序地包含如下工作環節：

　　1）作業的提交

　　2）作業的初始化

　　3）任務的分配

　　4）任務的執行

　　5）進程和狀態的更新

　　6）作業的完成

　　有關MapReduce的詳細工作細節，請見：《Hadoop權威指南（第二版）》第六章MapReduce工作機制。　

4.MapReduce執行框架的組件和執行流程

圖6是MapReduce執行框架的組件和執行流程，下面足一做解釋。

圖6：MapReduce執行框架的組件和執行流程

4.1 文件輸入格式InputFormat

1）定義了數據文件如何分割和讀取
2）InputFile提供了以下一些功能

• 選擇文件或者其它對象，用來作為輸入
• 定義InputSplits，將一個文件分開成為任務
• 為RecordReader提供一個工廠，用來讀取這個文件

3）有一個抽象的類FileInputFormat，所有的輸入格式類都從這個類繼承這個類的功能以及特性。當啟動一個Hadoop任務的時候，一個輸入文件所在的目錄被輸入到FileInputFormat對象中。FileInputFormat從這個目錄中讀取所有文件。然後FileInputFormat將這些文件分割為一個或者多個InputSplits。
4）通過在JobConf對象上設置JobConf.setInputFormat設置文件輸入的格式
2）接口定義

圖7：InputFormat接口定義

4.2 輸入數據分塊InputSplits

1）InputSplit定義了輸入到單個Map任務的輸入數據
2）一個MapReduce程序被統稱為一個Job，可能有上百個任務構成
3）InputSplit將文件分為64MB的大小

• 配置文件hadoop-site.xml中的mapred.min.split.size參數控制這個大小

4）mapred.tasktracker.map.taks.maximum用來控制某一個節點上所有map任務的最大數目

4.3 數據記錄讀入RecordReader（RR）

1）InputSplit定義了一項工作的大小，但是沒有定義如何讀取數據
2）RecordReader實際上定義了如何從數據上轉化為一個(key,value)對的詳細方法，並將數據輸出到Mapper類中
3）TextInputFormat提供了LineRecordReader

4.4 Mapper

1）每一個Mapper類的實例生成了一個Java進程（在某一個InputSplit上執行）
2）有兩個額外的參數OutputCollector以及Reporter，前者用來收集中間結果，後者用來獲得環境參數以及設置當前執行的狀態。
3）現在用Mapper.Context提供給每一個Mapper函數，用來提供上面兩個對象的功能

4.5 Combiner

1）合並相同key的鍵值對，減少partitioner時候的數據通信開銷
2）conf.setCombinerClass(Reduce.class)

4.6 Partitioner & Shuffle

在Map工作完成之後，每一個 Map函數會將結果傳到對應的Reducer所在的節點，此時，用戶可以提供一個Partitioner類，用來決定一個給定的(key,value)對傳輸的具體位置

4.7 Sort

傳輸到每一個節點上的所有的Reduce函數接收到得Key,value對會被Hadoop自動排序（即Map生成的結果傳送到某一個節點的時候，會被自動排序）

4.8 Reducer

1）做用戶定義的Reduce操作
2）最新的編程接口是Reducer.Context

4.9 文件輸出格式OutputFormat

1）說明
寫入到HDFS的所有OutputFormat都繼承自FileOutputFormat
每一個Reducer都寫一個文件到一個共同的輸出目錄，文件名是part-nnnnn，其中nnnnn是與每一個reducer相關的一個號（partition id）
FileOutputFormat.setOutputPath()
JobConf.setOutputFormat()
2）接口定義

圖8：OutputFormat接口定義

4.10 RecordWriter

TextOutputFormat實現了缺省的LineRecordWriter，以”key／value”形式輸出一行結果。

5. 容錯處理與計算性能優化

1）主要方法是將失敗的任務進行再次執行
2）TaskTracker會把狀態信息匯報給JobTracker，最終由JobTracker決定重新執行哪一個任務
3）為了加快執行的速度，Hadoop也會自動重復執行同一個任務，以最先執行成功的為準（投機執行）
4）mapred.map.tasks.speculative.execution
5）mapred.reduce.tasks.speculative.execution

hadoop入門筆記MapReduce簡介（三）