五分鐘讀懂大數據核心MapReduce架構及原理

什麼是MapReduce

Hadoop中的MapReduce是一個簡單的軟件框架，基於它寫出的應用程序可以運行在由上千個商用機器組成的大型集群上，並以一種可靠容錯式並行處理TB級數據

MapReduce的起源

源於谷歌在2004年發表的一篇MapReduce的論文，而Hadoop Reduce實際上就是谷歌MapReduce的克隆版本

MapReduce具有的特點

眾所周知MapReduce是一種很受歡迎的軟件框架，尤其是我們國家發展到現在互聯網的浪潮愈演愈烈，那麼它都有什麼特點呢？

1.易於編程：MapReduce通過相應的接口，程序員只需要簡單的調用就可以完成對一個複雜的分佈式程序的編寫。

2.易擴展性：在計算資源不足時可以通過增加機器來增加計算能力

3.高容錯性：要知道MapReduce的提出就是為了運行在廉價的商用pc中，而商用pc得到問題也是頗多，經常會出現pc掛掉的情況，這時候就需要可以迅速的把計算任何和資源轉移到另外的一個節點上運行，從而保證任務、作業的順利運行。

4.海量PB級數據的離線處理，所謂離線處理即為它不具體毫秒級別的迅速反饋能力，在對反饋要求非常及時的場景下，自然是不可用的

那麼MapReduce有哪些不適合的場景呢？

1.實時計算。沒有mysql等數據庫的毫秒級反饋能力，儘管可以用過hive等數據庫進行高速的計算但是速度還是不及

2.流式計算。流式計算的輸入數據時動態的，而 MapReduce 的輸入數據集是靜態的，不能動態變化。這是因為 MapReduce 自身的設計特點決定了數據源必須是靜態的

3.DAG(有向圖)模式：即每個作業或者任務之間都有很強的連接性，下一個作業的運行需要另外一個作業的運行結果的數據，這種情況下MapReduce的性能非常低，因為每個MapReduce的作業都會把計算寫入到磁盤中，若如此做則會造成大量的磁盤IO,性能低下。

MapReduce的編程模型

MapReduce 實例

為了分析 MapReduce 的編程模型，這裡我們以 WordCount 為實例。就像 Java、C++等編程語言的入門程序 hello word 一樣，WordCount 是 MapReduce 最簡單的入門程序。下面我們就來逐步分析。

1、場景：假如有大量的文件，裡面存儲的都是單詞。

類似應用場景：WordCount 雖然很簡單，但它是很多重要應用的模型。

1) 搜索引擎中，統計最流行的 K 個搜索詞。

2) 統計搜索詞頻率，幫助優化搜索詞提示。

2、任務：我們該如何統計每個單詞出現的次數？

3、將問題規範為：有一批文件（規模為 TB 級或者 PB 級），如何統計這些文件中所有單詞出現的次數。

4、解決方案：首先，分別統計每個文件中單詞出現的次數；然後，累加不同文件中同一個單詞出現次數。

通過上面的分析可知，它其實就是一個典型的 MapReduce 過程。下面我們通過示意圖來分析 MapReduce 過程。

上圖的流程大概分為以下幾步。

第一步：假設一個文件有三行英文單詞作為 MapReduce 的Input（輸入），這裡經過 Splitting 過程把文件分割為3塊。分割後的3塊數據就可以並行處理，每一塊交給一個 map 線程處理。

第二步：每個 map 線程中，以每個單詞為key，以1作為詞頻數value，然後輸出。

第三步：每個 map 的輸出要經過 shuffling（混洗），將相同的單詞key放在一個桶裡面，然後交給 reduce 處理。

第四步：reduce 接受到 shuffling 後的數據， 會將相同的單詞進行合併，得到每個單詞的詞頻數，最後將統計好的每個單詞的詞頻數作為輸出結果。

上述就是 MapReduce 的大致流程，前兩步可以看做 map 階段，後兩步可以看做 reduce 階段。下面我們來看看 MapReduce 大致實現。

1、Input：首先 MapReduce 輸入的是一系列key/value對。key表示每行偏移量，value代表每行輸入的單詞。

2、用戶提供了 map 函數和 reduce 函數的實現：

map(k,v) ——> list(k1,v1)

reduce(k1,list(v1)) ——>(k2,v2)

map 函數將每個單詞轉化為key/value對輸出，這裡key為每個單詞，value為詞頻1。(k1,v1)是 map 輸出的中間key/value結果對。reduce 將相同單詞的所有詞頻進行合併，比如將單詞k1，詞頻為list(v1)，合併為(k2,v2)。reduce 合併完之後，最終輸出一系列(k2,v2)鍵值對。

下面我們來看一下 MapReduce 的偽代碼。

map(key,value)://map 函數，key代表偏移量，value代表每行單詞

for each word w in value: //循環每行數據，輸出每個單詞和詞頻的鍵值對(w,1)

 emit(w,1)

reduce(key,values)://reduce 函數，key代表一個單詞，value代表這個單詞的所有詞頻數集合

 result=0

 for each count v in values: //循環詞頻集合，求出該單詞的總詞頻數，然後輸出(key,result)

 result+=v

 emit(key,result)

講到這裡，我們可以對 MapReduce 做一個總結。MapReduce 將 作業的整個運行過程分為兩個階段：Map 階段和Reduce 階段。

1、Map 階段

Map 階段是由一定數量的 Map Task 組成。這些 Map Task 可以同時運行，每個 Map Task又是由以下三個部分組成。

1) 對輸入數據格式進行解析的一個組件：InputFormat。因為不同的數據可能存儲的數據格式不一樣，這就需要有一個 InputFormat 組件來解析這些數據的存放格式。默認情況下，它提供了一個 TextInputFormat 來解釋數據格式。TextInputFormat 就是我們前面提到的文本文件輸入格式，它會將文件的每一行解釋成(key,value)，key代表每行偏移量，value代表每行數據內容。 通常情況我們不需要自定義 InputFormat，因為 MapReduce 提供了很多種InputFormat的實現，我們根據不同的數據格式，選擇不同的 InputFormat 來解釋就可以了。這一點我們後面會講到。

2)輸入數據處理：Mapper。這個 Mapper 是必須要實現的，因為根據不同的業務對數據有不同的處理。

3)數據分組：Partitioner。Mapper 數據處理之後輸出之前，輸出key會經過 Partitioner 分組或者分桶選擇不同的reduce。默認的情況下，Partitioner 會對 map 輸出的key進行hash取模，比如有6個Reduce Task，它就是模（mod）6，如果key的hash值為0，就選擇第0個 Reduce Task，如果key的hash值為1，就選擇第一個 Reduce Task。這樣不同的 map 對相同單詞key，它的 hash 值取模是一樣的，所以會交給同一個 reduce 來處理。

2、Reduce

Reduce 階段由一定數量的 Reduce Task 組成。這些 Reduce Task 可以同時運行，每個 Reduce Task又是由以下四個部分組成。

1) 數據運程拷貝。Reduce Task 要運程拷貝每個 map 處理的結果，從每個 map 中讀取一部分結果。每個 Reduce Task 拷貝哪些數據，是由上面 Partitioner 決定的。

2) 數據按照key排序。Reduce Task 讀取完數據後，要按照key進行排序。按照key排序後，相同的key被分到一組，交給同一個 Reduce Task 處理。

3) 數據處理：Reducer。以WordCount為例，相同的單詞key分到一組，交個同一個Reducer處理，這樣就實現了對每個單詞的詞頻統計。

4) 數據輸出格式：OutputFormat。Reducer 統計的結果，將按照 OutputFormat 格式輸出。默認情況下的輸出格式為 TextOutputFormat，以WordCount為例，這裡的key為單詞，value為詞頻數。

InputFormat、Mapper、Partitioner、Reducer和OutputFormat 都是用戶可以實現的。通常情況下，用戶只需要實現 Mapper和Reducer，其他的使用默認實現就可以了。