2020-12-21
MapReduce
MapReduce概念
2004年，Google發表一篇MapReduce論文，向全世界介紹了MapReduce，MapReduce主要應用於日誌分析，海量資料的排序等應用場景。Doug cutting根據谷歌的論文MapReduce編寫一個框架並將原始碼貢獻出來。
MapReduce是一種分散式計算模型，由Google提出，主要用於搜尋領域，解決離線海量資料的計算問題，不能實現對實時資料的分析和處理。
MapReduce由兩個階段組成：
map()：任務分解，reduce()：結果彙總，這兩個函式的形參是key、value對，表示函式的輸入資訊。
map、reduce鍵值對格式：
MapReduce的總執行過程
為了更簡單的理解MapReducer的執行過程，先對它來個總體的過程圖如下：
MapReduce執行的時候，通過Mapper執行的任務讀取HDFS中的資料檔案，然後呼叫自己的方法，處理資料，最後輸出。Reducer任務會接收Mapper任務輸出的資料，作為自己的輸入資料，呼叫自己的方法，最後輸出到HDFS的檔案中。
Mapper任務的執行過程
每個Mapper任務是一個java程序，它會讀取HDFS中的檔案，解析成很多的鍵值對，經過我們覆蓋的map方法處理後，轉換為很多的鍵值對再輸出。整個Mapper任務的處理過程又可以分為以下幾個階段，如圖所示：
1）第一階段是讀取HDF中的檔案。每一行解析成一個<k,v>。每一個鍵值對呼叫一次Map函式。把輸入檔案按照一定的標準分片(InputSplit)，每個輸入片的大小是固定的。預設情況下，輸入片(InputSplit)的大小與資料塊(Block)的大小是相同的。如果資料塊(Block)的大小是預設值64MB，輸入檔案有兩個，一個是32MB，一個是72MB，那麼小的檔案是一個輸入片，大檔案會分為兩個資料塊64MB和8MB，一共產生三個輸入片。每一個輸入片由一個Mapper程序處理。這裡的三個輸入片，會有三個Mapper程序處理。
2）第二階段是對輸入片中的記錄按照一定的規則解析成鍵值對。有個預設規則是把每一行文字內容解析成鍵值對。“鍵”是每一行的起始位置(單位是位元組)，“值”是本行的文字內容。
3）第三階段是呼叫Mapper類中的map方法。第二階段中解析出來的每一個鍵值對，呼叫一次map方法。如果有1000個鍵值對，就會呼叫1000次map方法。每一次呼叫map方法會輸出零個或者多個鍵值對。
4）第四階段是按照一定的規則對第三階段輸出的鍵值對進行分割槽。比較是基於鍵進行的。預設是隻有一個區。分割槽的數量就是Reducer任務執行的數量。
5）第五階段是對每個分割槽中的鍵值對進行排序。首先，按照鍵進行排序，對於鍵相同的鍵值對，按照值進行排序。比如三個鍵值對<2,2>、<1,3>、<2,1>，鍵和值分別是整數。那麼排序後的結果是<1,3>、<2,1>、<2,2>。如果有第六階段，那麼進入第六階段；如果沒有，直接輸出到本地的linux檔案中。
6）第六階段是對資料進行歸約處理，也就是reduce處理。鍵相等的鍵值對會呼叫一次reduce方法。經過這一階段，資料量會減少。歸約後的資料輸出到本地的linux檔案中。本階段預設是沒有的，需要使用者自己增加這一階段的程式碼。
Reducer任務的執行過程
每個Reducer任務是一個java程序。Reducer任務接收Mapper任務的輸出，歸約處理後寫入到HDFS中，可以分為如下圖所示的幾個階段。
1）第一階段是Reducer任務會主動從Mapper任務複製其輸出的鍵值對。Mapper任務可能會有很多，因此Reducer會複製多個Mapper的輸出。
2）第二階段是把複製到Reducer本地資料，全部進行合併，即把分散的資料合併成一個大的資料。再對合並後的資料排序。
3）第三階段是對排序後的鍵值對呼叫reduce方法,鍵相等的鍵值對呼叫一次reduce方法，每次呼叫會產生零個或者多個鍵值對。最後把這些輸出的鍵值對寫入到HDFS檔案中。
MapReduce資料型別與格式
Writable介面與序列化機制
1、Writable介面
MapReduce的任意Key和Value必須實現Writable介面。
public interface Writable {
void write(DataOutput out) throwas IOException;
void readFilelds(DataInput in) throws IOException;
}
MapReduce的任意key必須實現WritableComparable介面
public interface WritableComparable<T> extends Writable,Comparable<T>{
}
常用的Writable實現類
Text一般認為它等價於java.lang.String的Writable。針對UTF-8序列。例:
Text test = new Text("test");
IntWritable one = new IntWritable(1);
2、序列化概念
序列化（Serialization）是指把結構化物件轉化為位元組流，便於在網路上傳輸或寫到磁碟進行永久儲存。反序列化（Deserialization）是序列化的逆過程。即把位元組流轉回結構化物件。
序列化在分散式資料處理的領域經常使用：程序間通訊和永久儲存
Hadoop的序列化格式：Writable
序列化格式特點：
Ø 緊湊：高效使用儲存空間。
Ø 快速：讀寫資料的額外開銷小
Ø 可擴充套件：可透明地讀取老格式的資料
Ø 互操作：支援多語言的互動
Hadoop序列化的作用
Ø 序列化在分散式環境的兩大作用：程序間通訊，永久儲存。
Ø Hadoop節點間通訊。
補充
- TaskTracker負責執行任務、JobClient負責提交作業的，負責啟動、跟蹤任務執行、訪問任務狀態和日誌
- 執行MapReduce任務時，Map函式會呼叫，而reduce函式不一定執行
- 分片數目在numSplits中限定，分片大小必須大於mapred.min.size個位元組，但小於檔案系統的塊。
- TextInputFormat是預設的InputFormat。