Hadoop mapreduce原理學習

MapReduce模式結構圖：

細化圖：

最近整了很長一段時間才瞭解了map reduce的工做原理，shuffle是mapreduce的心臟，瞭解了這個過程，有助於編寫效率更高的mapreduce程序和hadoop調優。本身畫了一幅流程圖（點擊查看全圖）：

另外，還找到一篇文章，很好，引用一下。

Hadoop

是Apache 下的一個項目，由HDFS、MapReduce、HBase、Hive 和ZooKeeper等成員組成。其中，HDFS 和MapReduce 是兩個最基礎最重要的成員。

HDFS是Google GFS 的開源版本，一個高度容錯的分佈式文件系統，它可以提供高吞吐量的數據訪問，適合存儲海量（PB 級）的大文件（一般超過64M），其原理以下圖所示：

 

採用Master/Slave 結構。NameNode 維護集羣內的元數據，對外提供建立、打開、刪除和重命名文件或目錄的功能。DatanNode 存儲數據，並提負責處理數據的讀寫請求。DataNode按期向NameNode 上報心跳，NameNode 經過響應心跳來控制DataNode。

 

InfoWord將MapReduce 評爲2009 年十大新興技術的冠軍。MapReduce 是大規模數據（TB 級）計算的利器，Map 和Reduce 是它的主要思想，來源於函數式編程語言，它的原理以下圖所示：Map負責將數據打散，Reduce負責對數據進行彙集，用戶只須要實現map 和reduce 兩個接口，便可完成TB級數據的計算，常見的應用包括：日誌分析和數據挖掘等數據分析應用。另外，還可用於科學數據計算，如圓周率PI 的計算等。Hadoop MapReduce的實現也採用了Master/Slave 結構。Master 叫作JobTracker，而Slave 叫作TaskTracker。用戶提交的計算叫作Job，每個Job會被劃分紅若干個Tasks。JobTracker負責Job 和Tasks 的調度，而TaskTracker負責執行Tasks。

 

 

 

MapReduce中的Shuffle和Sort分析

MapReduce 是現今一個很是流行的分佈式計算框架，它被設計用於並行計算海量數據。第一個提出該技術框架的是Google 公司，而Google 的靈感則來自於函數式編程語言，如LISP，Scheme，ML 等。MapReduce 框架的核心步驟主要分兩部分：Map 和Reduce。當你向MapReduce 框架提交一個計算做業時，它會首先把計算做業拆分紅若干個Map 任務，而後分配到不一樣的節點上去執行，每個Map 任務處理輸入數據中的一部分，當Map 任務完成後，它會生成一些中間文件，這些中間文件將會做爲Reduce 任務的輸入數據。Reduce 任務的主要目標就是把前面若干個Map 的輸出彙總到一塊兒並輸出。從高層抽象來看，MapReduce的數據流圖如圖1 所示：

 

本文的重點是剖析MapReduce的核心過程----Shuffle和Sort。在本文中，Shuffle是指從Map產生輸出開始，包括系統執行排序 以及傳送Map輸出到Reducer做爲輸入的過程。在這裏咱們將去探究Shuffle是如何工做的，由於對基礎的理解有助於對MapReduce程序進 行調優。

 

首先從Map端開始分析，當Map開始產生輸出的時候，他並非簡單的把數據寫到磁盤，由於頻繁的操做會導致性能嚴重降低，他的處理更加複雜，數據首先是寫到內存中的一個緩衝區，並做一些預排序，以提高效率，如圖：

每一個Map任務都有一個用來寫入輸出數據的循環內存緩衝區，這個緩衝區默認大小是100M，能夠經過io.sort.mb屬性來設置具體的大小，當緩衝區中的數據量達到一個特定的閥值(io.sort.mb * io.sort.spill.percent，其中io.sort.spill.percent 默認是0.80)時，系統將會啓動一個後臺線程把緩衝區中的內容spill 到磁盤。在spill過程當中，Map的輸出將會繼續寫入到緩衝區，但若是緩衝區已經滿了，Map就會被阻塞直道spill完成。spill線程在把緩衝區的數據寫到磁盤前，會對他進行一個二次排序，首先根據數據所屬的partition排序，而後每一個partition中再按Key排序。輸出包括一個索引文件和數據文件，若是設定了Combiner，將在排序輸出的基礎上進行。Combiner就是一個Mini Reducer，它在執行Map任務的節點自己運行，先對Map的輸出做一次簡單的Reduce，使得Map的輸出更緊湊，更少的數據會被寫入磁盤和傳送到Reducer。Spill文件保存在由mapred.local.dir指定的目錄中，Map任務結束後刪除。

每當內存中的數據達到spill閥值的時候，都會產生一個新的spill文件，因此在Map任務寫完他的最後一個輸出記錄的時候，可能會有多個spill文件，在Map任務完成前，全部的spill文件將會被歸併排序爲一個索引文件和數據文件。如圖3 所示。這是一個多路歸併過程，最大歸併路數由io.sort.factor 控制(默認是10)。若是設定了Combiner，而且spill文件的數量至少是3（由min.num.spills.for.combine 屬性控制），那麼Combiner 將在輸出文件被寫入磁盤前運行以壓縮數據。

對寫入到磁盤的數據進行壓縮（這種壓縮同Combiner 的壓縮不同）一般是一個很好的方法，由於這樣作使得數據寫入磁盤的速度更快，節省磁盤空間，並減小須要傳送到Reducer 的數據量。默認輸出是不被壓縮的， 但能夠很簡單的設置mapred.compress.map.output爲true 啓用該功能。壓縮所使用的庫由mapred.map.output.compression.codec來設定

 

當spill 文件歸並完畢後，Map 將刪除全部的臨時spill 文件，並告知TaskTracker 任務已完成。Reducers 經過HTTP 來獲取對應的數據。用來傳輸partitions 數據的工做線程個數由tasktracker.http.threads 控制，這個設定是針對每個TaskTracker 的，並非單個Map，默認值爲40，在運行大做業的大集羣上能夠增大以提高數據傳輸速率。

 

如今讓咱們轉到Shuffle的Reduce部分。Map的輸出文件放置在運行Map任務的TaskTracker的本地磁盤上（注意：Map輸出老是寫到本地磁盤，可是Reduce輸出不是，通常是寫到HDFS），它是運行Reduce任務的TaskTracker所須要的輸入數據。Reduce任務的輸入數據分佈在集羣內的多個Map任務的輸出中，Map任務可能會在不一樣的時間內完成，只要有其中一個Map任務完成，Reduce任務就開始拷貝他的輸出。這個階段稱爲拷貝階段，Reduce任務擁有多個拷貝線程，能夠並行的獲取Map輸出。能夠經過設定mapred.reduce.parallel.copies來改變線程數。

Reduce是怎麼知道從哪些TaskTrackers中獲取Map的輸出呢？當Map任務完成以後，會通知他們的父TaskTracker，告知狀態更新，而後TaskTracker再轉告JobTracker，這些通知信息是經過心跳通訊機制傳輸的，所以針對以一個特定的做業，jobtracker知道Map輸出與tasktrackers的映射關係。Reducer中有一個線程會間歇的向JobTracker詢問Map輸出的地址，直到把全部的數據都取到。在Reducer取走了Map輸出以後，TaskTracker不會當即刪除這些數據，由於Reducer可能會失敗，他們會在整個做業完成以後，JobTracker告知他們要刪除的時候纔去刪除。

若是Map輸出足夠小，他們會被拷貝到Reduce TaskTracker的內存中（緩衝區的大小由mapred.job.shuffle.input.buffer.percnet控制），或者達到了 Map輸出的閥值的大小(由mapred.inmem.merge.threshold控制)，緩衝區中的數據將會被歸併而後spill到磁盤。

拷貝來的數據疊加在磁盤上，有一個後臺線程會將它們歸併爲更大的排序文件，這樣作節省了後期歸併的時間。對於通過壓縮的Map 輸出，系統會自動把它們解壓到內存方便對其執行歸併。

當全部的Map 輸出都被拷貝後，Reduce 任務進入排序階段（更恰當的說應該是歸併階段，由於排序在Map 端就已經完成），這個階段會對全部的Map 輸出進行歸併排序，這個工做會重複屢次才能完成。

 

假設這裏有50 個Map 輸出（可能有保存在內存中的），而且歸併因子是10（由io.sort.factor控制，就像Map 端的merge 同樣），那最終須要5 次歸併。每次歸併會把10個 文件歸併爲一個，最終生成5 箇中間文件。在這一步以後，系統再也不把5 箇中間文件歸併成一個，而是排序後直接「喂」給Reduce 函數，省去向磁盤寫數據這一步。最終歸併的數據能夠是混合數據，既有內存上的也有磁盤上的。因爲歸併的目的是歸併最少的文件數目，使得在最後一次歸併時總 文件個數達到歸併因子的數目，因此每次操做所涉及的文件個數在實際中會更微妙些。譬如，若是有40 個文件，並非每次都歸併10 個最終獲得4 個文件，相反第一次只歸併4 個文件，而後再實現三次歸併，每次10 個，最終獲得4 個歸併好的文件和6 個未歸併的文件。要注意，這種作法並無改變歸併的次數，只是最小化寫入磁盤的數據優化措施，由於最後一次歸併的數據老是直接送到Reduce 函數那裏。在Reduce 階段，Reduce 函數會做用在排序輸出的每個key 上。這個階段的輸出被直接寫到輸出文件系統，通常是HDFS。在HDFS 中，由於TaskTracker 節點也運行着一個DataNode 進程，因此第一個塊備份會直接寫到本地磁盤。到此，MapReduce 的Shuffle 和Sort 分析完畢。