2.2 分布式計算編程模型MapReduce

MapReduce是Google提出的一個軟件架構，是一種處理海量數據的并行編程模型，用于大規模數據集（通常大于1TB）的并行運算。Map（映射）、Reduce（化簡）的概念和主要思想，都是從函數式編程語言和矢量編程語言借鑒來的。正是由于MapReduce有函數式編程語言和矢量編程語言的共性，這種編程模型特別適合非結構化和結構化的海量數據的搜索、挖掘、分析與機器智能學習等。

2.2.1 產生背景

MapReduce這種并行編程模型思想最早是在1995年被提出的，Darlington等人首次提出了“map”和“fold”的概念，與Google現在所使用的“Map”和“Reduce”的思想相吻合。

與傳統的分布式程序設計相比，MapReduce封裝了并行處理、容錯處理、本地化計算、負載均衡等細節，還提供了一個簡單而強大的接口。通過這個接口，可以把大尺度的計算自動地并發和分布執行，使編程變得非常容易。另外，MapReduce也具有較好的通用性，大量不同的問題都可以簡單地通過MapReduce來解決。

MapReduce把對數據集的大規模操作，分發給一個主節點管理下的各分節點共同完成，通過這種方式實現任務的可靠執行與容錯機制。在每個時間周期，主節點都會對分節點的工作狀態進行標記。一旦將分節點狀態標記為死亡狀態，則這個節點的所有任務都將分配給其他分節點重新執行。

據相關統計，每使用一次Google搜索引擎，Google的后臺服務器就要進行1011次運算。這么龐大的運算量，如果沒有好的負載均衡機制，有些服務器的利用率會很低，有些則會負荷太重，有些甚至可能死機，這些都會影響系統對用戶的服務質量。而使用MapReduce這種編程模型，就保持了服務器之間的均衡，提高了整體效率。

2.2.2 編程模型

MapReduce的運行模型如圖2-2所示。圖中有M個Map操作和R個Reduce操作。

簡單地說，一個Map函數就是對一部分原始數據進行指定的操作。每個Map操作都針對不同的原始數據，因此Map與Map之間是互相獨立的，這使它們可以充分并行化。一個Reduce操作就是對每個Map產生的一部分中間結果進行合并操作，每個Reduce所處理的Map中間結果是互不交叉的，所有Reduce產生的最終結果經過簡單連接就形成了完整的結果集，因此Reduce也可以在并行環境下執行。

在編程的時候，開發者需要編寫兩個主要函數：

Map和Reduce的輸入參數及輸出結果根據應用的不同而有所不同。Map的輸入參數是in_key和in_value，它指明了Map需要處理的原始數據是哪些。Map的輸出結果是一組<key，value>對，這是經過Map操作后所產生的中間結果。在進行Reduce操作之前，系統已經將所有Map產生的中間結果進行了歸類處理，使得相同key對應的一系列value能夠集結在一起提供給一個Reduce進行歸并處理，也就是說，Reduce的輸入參數是（key，[value1，…，valuem]）。Reduce的工作是對這些對應相同key的value值進行歸并處理，最終形成（key，final_value）的結果。這樣，一個Reduce處理了一個key，所有Reduce的結果并在一起就是最終結果。

例如，假設我們想用MapReduce來計算一個大型文本文件中各單詞出現的次數，Map的輸入參數指明了需要處理哪部分數據，以“（在文本中的起始位置，需要處理的數據長度）”表示，經過Map處理，形成一批中間結果“<單詞，出現次數>”。而Reduce處理這些中間結果，將相同單詞出現的次數進行累加，得到每個單詞總的出現次數。

2.2.3 實現機制

MapReduce操作的執行流程如圖2-3所示。

用戶程序調用MapReduce函數后，會引起下面的操作過程（圖2-3中的數字標示和下面的數字標示相同）。

（1）MapReduce函數首先把輸入文件分成M塊，每塊大概16～64MB（可以通過參數決定），接著在集群的機器上執行分派處理程序。

（2）分派處理程序中有一個程序比較特別，它是主控程序Master。剩下的分派處理程序都作為Master分派工作的Worker（工作機）。總共有M個Map任務和R個Reduce任務需要分派，Master選擇空閑的Worker來分配這些Map或Reduce任務。

（3）一個被分配了Map任務的Worker讀取并處理相關的輸入塊。它處理輸入的數據，并且將分析出的<key，value>對傳遞給用戶定義的Map函數。Map函數產生的中間結果<key，value>對暫時緩沖到內存。

（4）這些緩沖到內存的中間結果將被定時寫到本地硬盤，這些數據通過分區函數分成R個區。中間結果在本地硬盤的位置信息將被發送回Master，然后Master負責把這些位置信息傳送給Reduce Worker。

（5）當Master通知Reduce Worker關于中間<key，value>對的位置時，它調用遠程過程，從Map Worker的本地硬盤上讀取緩沖的中間數據。當Reduce Worker讀到所有的中間數據時，它就使用中間key進行排序，這樣可使有相同key的值都在一起。因為有許多不同key的Map都對應相同的Reduce任務，所以，排序是必需的。如果中間結果集過于龐大，那么就需要使用外排序。

（6）Reduce Worker根據每個唯一中間key來遍歷所有排序后的中間數據，并且把key和相關的中間結果值集傳遞給用戶定義的Reduce函數。Reduce函數的結果將被寫入一個最終的輸出文件。

最終，當所有的Map任務和Reduce任務都完成的時候，Master激活用戶程序。此時MapReduce返回用戶程序的調用點。

由于MapReduce在成百上千臺機器上處理海量數據，因此容錯機制是不可或缺的。總的來說，MapReduce通過重新執行失效的地方來實現容錯。

1.Master失效

Master會周期性地設置檢查點（Checkpoint），并導出Master的數據。一旦某個任務失效，系統就從最近的一個檢查點恢復并重新執行。由于只有一個Master在運行，如果Master失效了，則只能終止整個MapReduce程序的運行并重新開始。

2.Worker失效

相對于Master失效而言，Worker失效算是一種常見的狀態。Master會周期性地給Worker發送ping命令，如果有Worker沒有應答，則Master認為該Worker失效，終止對這個Worker的任務調度，把失效Worker的任務調度到其他Worker上重新執行。

2.2.4 案例分析

排序通常用于衡量分布式數據處理框架的數據處理能力，下面介紹如何利用MapReduce進行數據排序。假設有一批海量的數據，每個數據都是由26個字母組成的字符串，原始的數據集是完全無序的，怎樣通過MapReduce完成排序工作，使其有序（字典序）呢？可通過以下三個步驟來完成。

（1）對原始的數據進行分割（Split），得到N個不同的數據分塊，如圖2-4所示。

（2）對每個數據分塊都啟動一個Map進行處理。采用桶排序的方法，按照首字母將每個Map中的字符串分配到26個不同的桶中。圖2-5是Map的過程及其得到的中間結果。

（3）對于Map得到的中間結果，啟動26個Reduce。按照首字母將Map中不同桶中的字符串集放置到相應的Reduce中進行處理。具體來說就是首字母為a的字符串全部放在Reduce1中處理，首字母為b的字符串全部放在Reduce2中處理，以此類推。每個Reduce對于其中的字符串進行排序，結果直接輸出。由于Map過程中已經做到了首字母有序，Reduce輸出的結果就是最終的排序結果。這一過程如圖2-6所示。

從上述過程中可以看出，由于能夠實現處理過程的完全并行化，因此利用MapReduce處理海量數據是非常合適的。