大數據技術 - MapReduce的Shuffle及調優

本章內容咱們學習一下 MapReduce 中的 Shuffle 過程，Shuffle 發生在 map 輸出到 reduce 輸入的過程，它的中文解釋是 「洗牌」，顧名思義該過程涉及數據的從新分配，主要分爲兩部分：1. map 任務輸出的數據分組、排序，寫入本地磁盤 2. reduce 任務拉取排序。因爲該過程涉及排序、磁盤IO、以及網絡IO 等消耗資源和 CPU 比較大的操做，所以該過程向來是「兵家必爭」之地，即你們會重點優化的一個地方，所以也是大數據面試中常常會被重點考察的地方。本文力求通俗、簡單地將 Shuffle 過程描述清楚。

包含 Shuffle 過程的 MapReduce 任務處理流程以下圖，圖片來自《Hadoop權威指南（第四版）》

接下來，分別介紹 Shuffle 所涉及的主要操做。

map 端

map 端輸出時，先將數據寫入內存中的環形緩衝區，默認大小爲 100M，能夠經過 mapreduce.task.io.sort.mb 來設置。map 端輸出過程以下：

• 當緩衝區的內容大小達到閾值（默認 0.8，即緩衝區大小的 80%，可經過 mapreduce.map.sort.spill.percent 設置），便有一個後臺線程會將寫入緩衝區的內容溢寫到磁盤。溢寫的過程當中 map 任務仍然能夠寫緩衝區，一旦緩衝區寫滿，map 任務阻塞，直到後臺線程寫磁盤結束
• 後臺線程寫磁盤以前會計算輸出的 key 的分區（一個分區對應一個 reduce 任務），同一個分區的 key 分在一組並按照 key 排序。最後寫到本地磁盤。若是設置 combiner 函數，會在寫磁盤以前調用 combaner 函數。咱們以前沒有介紹 combiner，不理解的同窗能夠先忽略，只需知道它是先將數據聚合爲了減小網絡IO，且不會影響 reduce 計算結果的一個操做便可
• 每一次溢寫都會產生一個溢出文件，map 輸出結束後會產生多個溢出文件。最終會被合併成一個分區的且有序的文件。這裏爲何要合併成 1 個，由於若是 map 輸出的數據比較多，產生本地的小文件會太多，影響系統性能。所以須要進行合併，經過 mapreduce.task.io.sort.factor 設置一次能夠合併的文件個數，默認爲 10 
• 輸出到磁盤的過程當中能夠設置壓縮， 默認不壓縮。經過設置 mapreduce.map.output.compress 爲 true 開啓壓縮
  

以上即是 map 任務輸出過程的主要操做，輸出到磁盤後，reducer 會經過 http 服務拉取輸出文件中屬於本身分區的數據。

reduce 端

reduce 端在 Shuffle 階段主要涉及複製和排序兩個過程。 reduce 端拉取 map 輸出數據的過程是複製階段，對應上圖中的 fetch。一個 reduce 任務須要從多個 map 輸出複製。所以只要有 map 任務完成，reduce 任務就能夠進行復制。複製的過程能夠是多線程併發進行，併發的線程個數由 mapreduce.reduce.shuffle.parallelcopies 設置，默認是 5 。

• map 任務完成後經過心跳通知 application master，reduce 端會有一個線程按期查詢 application master，以獲取完成的 map 任務的位置，從而去對應的機器複製數據
• reduce 複製的數據先寫到 reduce 任務的 JVM 內存，經過 mapreduce.reduce.shuffle.input.buffer.percent 控制能夠用的內存比例
• 若是複製的數據大小達到內存閾值（經過 mapreduce.reduce.shuffle.merge.percent 控制）或者複製的文件數達到閾值（經過 mapreduce.reduce.merge.inmem.threshold 控制，默認 1000）則將內存的數據合併溢寫到磁盤，若是設置了 combine 函數，寫磁盤前會調用 combine 函數以減小寫入磁盤的數據量
• 複製階段結束後，reduce 將進入排序階段。若是發生了上面第三步，即產生溢寫，那麼磁盤可能會有多個溢寫文件，此時須要將磁盤文件合併並排序。若是溢寫的文件較多，須要屢次合併，每次合併的文件數由 mapreduce.task.io.sort.factor 控制。最後一次合併排序的時候不會將數據寫到磁盤而直接做爲 reduce 任務的輸入

以上即是 reduce 任務前的複製、排序階段。至此，整個 Shuffle 過程就介紹完畢。

參數調優

咱們在上面介紹 Shuffle 過程時已經提到了一些參數，這裏統一整理並說明一下

map 端調優參數

參數名	默認值	說明
mapreduce.task.io.sort.mb	100	map 輸出是所使用的內存緩衝區大小，單位：MB
mapreduce.map.sort.spill.percent	0.80	map 輸出溢寫到磁盤的內存閾值
mapreduce.task.io.sort.factor	10	排序文件是一次能夠合併的流數
mapreduce.map.output.compress	false	map 輸出是否壓縮
mapreduce.map.output.compress.codec	org.apache.hadoop.io.compress.DefaultCodec	map 輸出壓縮的編解碼器

咱們但願在 map 輸出階段可以提供更多的內存空間，以提高性能。所以 map 函數應該儘可能少佔用內存，以便留出內存用於輸出。咱們也能夠評估 map 輸出，經過增大 mapreduce.task.io.sort.mb 值來減小溢寫的文件數。

reduce 端調優參數

參數名	默認值	說明
mapreduce.reduce.shuffle.parallelcopies	5	併發複製的線程數
mapreduce.task.io.sort.factor	10	同 map 端
mapreduce.reduce.shuffle.input.buffer.percent	0.70	Shuffle 的複製階段，用來存放 map 輸出緩衝區佔reduce 堆內存的百分比
mapreduce.reduce.shuffle.merge.percent	0.66	map 輸出緩衝區的閾值，超過該比例將進行合併和溢寫磁盤
mapreduce.reduce.merge.inmem.threshold	1000	閾值，當累積的 map 輸出文件數超過該值，進行合併和溢寫磁盤，0或者負值意味着改參數無效，合併和溢寫只由 mapreduce.reduce.shuffle.merge.percent 控制
mapreduce.reduce.input.buffer.percent	0.0	在 reduce 過程（開始運行 reduce 函數時），內存中保存 map 輸出的空間站整個堆空間的比例。 默認狀況下，reduce 任務開始前全部的 map 輸出合併到磁盤，以便爲 reducer 提供儘量多的內存。 若是 reducer 須要的內存較少，能夠增長此值以最小化磁盤訪問次數

在 reduce 端，進行 reduce 函數前，若是中間數據所有駐留內存能夠得到最佳性能，默認狀況是不能實現的。若是 reduce 函數內存需求不大，把 mapreduce.reduce.input.buffer.percent 參數設置大一些能夠提高性能。

總結

今天這章，咱們詳細介紹了 Shuffle 過程，關注 Shuffle 過程的性能對整個 MR 做業的性能調優相當重要。通過這章的介紹，咱們可以掌握 Shuffle 過程的關鍵技術點，雖然還不算深刻。同時，咱們介紹了常見的參數以及調優方法，但願可以在實際應用中不斷的嘗試、總結，寫出性能最佳的任務。