Mapreduce shuffle的示例分析

這篇文章給大家分享的是有關Mapreduce shuffle的示例分析的內容。小編覺得挺實用的，因此分享給大家做個參考，一起跟隨小編過來看看吧。

Mapreduce shuffle詳解

Mapreduce確保每個reducer的的輸入都是按鍵排序的。系統執行排序的過程(即將map輸出作為輸入 傳給reducer)成為shuffle。從多個方面來看shuffle是mapreduce的心臟，是奇跡發生的地方。

上圖展示了，mapreduce的詳細過程。

1  輸入分片

對于數據的輸入分片，要根據不同的存儲格式有不同的介紹。對于，hdfs存儲的文件，數據的分片就可分為兩種，文件可切分(不壓縮或者壓縮格式bzip2)的按照一定大小進行分片有既定算法，默認是block的大小，具體算法不在這里細講，前面hive調優的文章又說到，而且浪尖也會在后續的文章提到這個內容；

分片的時候計算公式計算過程舉例

文件不可切分則一個文件一個分片。

2  Map端

從上圖我們可以看到map端的處理過程。Map會讀取輸入分片數據。但是map函數開始產生輸出時并不是簡單的將數據寫入磁盤。這個過程很復雜，他利用了緩沖的方式寫到內存并出于效率的考慮進行排序。

每個map任務都是有一個環形緩沖區的用于存儲任務的輸出。在默認情況下，緩沖區的大小為100MB，辭職可以通過改變io.sort.mb來調整。一旦緩沖內容達到閾值(io.sort,spill,percent，默認是0.8)，一個后臺線程會將內容spill到磁盤。在spill到磁盤的過程中，map輸出并不會停止往緩沖區寫入數據，但如果在此期間緩沖區被寫滿，map會被阻塞知道寫磁盤過程完成。

溢出寫過程安裝輪詢方式將緩沖區的內容寫到mapred.local.dir指定的作業特定子目錄中的目錄中。

寫磁盤之前，線程首先根據數據最終要傳的reducer把數據劃分成相應的分區。在每個分區中后臺線程按鍵進行內排序，如果有一個combiner，它就在排序后的輸出上運行。運行combinner使得map輸出結果更緊湊，因此可以減少寫到磁盤的數據和傳遞給reducer的數據。

每次內存緩沖區達到溢出閾值，就會新建一個溢出文件(spill file)，因此在map任務寫完其最后一個輸出記錄之后，會有幾個溢寫文件。在任務完成之前，溢出文件被合并成一個已分區且已排序的輸出文件。配置屬性io.sort.factor控制著一次最多能合并多少流，默認是10。

如果至少存在3個溢出文件(通過min.num.spills.for.combine屬性設置)時，則combiner就會在輸出文件寫到磁盤之前再次運行。前面曾說過，combiner可以在輸入上反復運行，單不影響最終的結果。如果只有一兩個溢出文件，那么對map輸出的減少不值得調用combiner，就不會為map輸出再次運行combiner。

在將壓縮map輸出寫到磁盤的過程中對它進行壓縮往往是個好主意，因為這樣就會寫磁盤的速度更快，更加節約時間，并且減少傳給reducer的數據量。在默認情況下，輸出是不壓縮的，但是只要將mapred.compress.map.output設置為true，就可以啟用這個功能。使用的壓縮庫由mapred.map.output.compression.codec指定。

Reducer是通過HTTP的方式得到輸出文件的分區。在MRV2中使用netty進行數據傳輸，默認情況下netty的工作線程數是處理器數的2倍。MRV1中，默認值是40，由tracker.http.threads來在tasktracker端設定。

3   Reducer端

集群中往往一個mr任務會有若干map任務和reduce任務，map任務運行有快有慢，reduce不可能等到所有的map任務都運行結束再啟動，因此只要有一個任務完成，reduce任務就開始復制器輸出。復制線程的數量由mapred.reduce.parallel.copies屬性來改變，默認是 5。

Reducer如何知道map輸出的呢？對于MRv2 map運行結束之后直接就通知了appmaster，對于給定的job appmaster是知道map的輸出和host之間的關系。在reduce端獲取所有的map輸出之前，Reduce端的線程會周期性的詢問master 關于map的輸出。Reduce并不會在獲取到map輸出之后就立即刪除hosts，因為reduce有肯能運行失敗。相反，是等待appmaster的刪除消息來決定刪除host。

Reduce對map輸出的不同大小也有相應的調優處理。如果map輸出相當小，會被復制到reduce任務JVM的內存(緩沖區大小由mapred.job.shuffle.input.buffer.percent屬性控制，指定用于此用途的堆空間的百分比)，否則，map輸出會被復制到磁盤。一旦內存緩沖區達到閾值(由mapred.job.shuffle.merge.percent決定)或達到map的輸出閾值(mapred.inmem.merge,threshold控制)，則合并后溢出寫到磁盤中。如果指定combiner，則在合并期間運行它已降低寫入磁盤的數據量。

隨著磁盤上副本的增多，后臺線程會將它們合并為更大的，排序好的文件。這會為后面的合并節省一些時間。注意，為了合并，壓縮的map輸出(通過map任務)都必須在內存中解壓縮。

復制完所有的map輸出后，reduce任務進入排序階段(更加恰當的說法是合并階段，因為排序是在map端進行的)，這個階段將合并map的輸出，維持其順序排序。這是循環進行的。比如，有50個map輸出，而合并因子是10(默認值是10，由io.sort.factor屬性設置，與map的合并類似)，合并將進行5趟。每趟將10個文件合并成一個文件，因此最后有5個中間文件。

在最后階段，即reduce階段，直接把數據輸入reduce函數，從而省略了一次磁盤往返行程，并沒有將這5個文件合并成一個已排序的文件最為最后一趟。最后的合并可以來自內存和磁盤片段。

在reduce階段，對已排序輸出中的每個鍵調用reduce函數。此階段的輸出直接寫到輸出文件系統，一般為hdfs。

注意：

每趟合并的文件數實際上比上面例子中展示的有所不同的。目標是合并最小數據量的文件以便滿足最后一趟的合并系數。因此，如果有40個文件，我們不會再四趟中每趟合并10個文件而得到4個文件。相反，第一趟只合并4個文件，隨后的三塘合并10個文件。最后一趟中，4個已經合并的文件和剩余的6個文件合計是個文件進行合并。如下圖所述：

注意這并沒有改變合并的次數，它只是一個優化措施，目的是盡量減少寫到磁盤的數據量，因為最后一趟總是直接合并到reduce。

感謝各位的閱讀！關于“Mapreduce shuffle的示例分析”這篇文章就分享到這里了，希望以上內容可以對大家有一定的幫助，讓大家可以學到更多知識，如果覺得文章不錯，可以把它分享出去讓更多的人看到吧！