1 Map side tuning 參數

1.1 MapTask 運行內部原理

當map task 開始運算，并產生中間數據時，其產生的中間結果并非直接就簡單的寫入磁盤。這中間的過程比較復雜，并且利用到了內存buffer 來進行已經產生的部分結果的緩存，并在內存buffer 中進行一些預排序來優化整個map 的性能。如上圖所示，每一個map 都會對應存在一個內存buffer （MapOutputBuffer ，即上圖的buffer in memory ），map 會將已經產生的部分結果先寫入到該buffer 中，這個buffer 默認是100MB 大小，但是這個大小是可以根據job 提交時的參數設定來調整的，該參數即為： io.sort.mb 。當map 的產生數據非常大時，并且把io.sort.mb 調大，那么map 在整個計算過程中spill 的次數就勢必會降低，map task 對磁盤的操作就會變少，如果map tasks 的瓶頸在磁盤上，這樣調整就會大大提高map 的計算性能。map 做sort 和spill 的內存結構如下如所示：

map 在運行過程中，不停的向該buffer 中寫入已有的計算結果，但是該buffer 并不一定能將全部的map 輸出緩存下來，當map 輸出超出一定閾值（比如100M ），那么map 就必須將該buffer 中的數據寫入到磁盤中去，這個過程在mapreduce 中叫做spill 。map 并不是要等到將該buffer 全部寫滿時才進行spill ，因為如果全部寫滿了再去寫spill ，勢必會造成map 的計算部分等待buffer 釋放空間的情況。所以，map 其實是當buffer 被寫滿到一定程度（比如80% ）時，就開始進行spill 。這個閾值也是由一個job 的配置參數來控制，即 io.sort.spill.percent ，默認為0.80 或80% 。這個參數同樣也是影響spill 頻繁程度，進而影響map task 運行周期對磁盤的讀寫頻率的。但非特殊情況下，通常不需要人為的調整。調整io.sort.mb 對用戶來說更加方便。

當map task 的計算部分全部完成后，如果map 有輸出，就會生成一個或者多個spill 文件，這些文件就是map 的輸出結果。map 在正常退出之前，需要將這些spill 合并（merge ）成一個，所以map 在結束之前還有一個merge 的過程。merge 的過程中，有一個參數可以調整這個過程的行為，該參數為： io.sort.factor 。該參數默認為10 。它表示當merge spill 文件時，最多能有多少并行的stream 向merge 文件中寫入。比如如果map 產生的數據非常的大，產生的spill 文件大于10 ，而io.sort.factor 使用的是默認的10 ，那么當map 計算完成做merge 時，就沒有辦法一次將所有的spill 文件merge 成一個，而是會分多次，每次最多10 個stream 。這也就是說，當map 的中間結果非常大，調大io.sort.factor ，有利于減少merge 次數，進而減少map 對磁盤的讀寫頻率，有可能達到優化作業的目的。

當job 指定了combiner 的時候，我們都知道map 介紹后會在map 端根據combiner 定義的函數將map 結果進行合并。運行combiner 函數的時機有可能會是merge 完成之前，或者之后，這個時機可以由一個參數控制，即 min.num.spill.for.combine （default 3 ），當job 中設定了combiner ，并且spill 數最少有3 個的時候，那么combiner 函數就會在merge 產生結果文件之前運行。通過這樣的方式，就可以在spill 非常多需要merge ，并且很多數據需要做conbine 的時候，減少寫入到磁盤文件的數據數量，同樣是為了減少對磁盤的讀寫頻率，有可能達到優化作業的目的。

減少中間結果讀寫進出磁盤的方法不止這些，還有就是壓縮。也就是說map 的中間，無論是spill 的時候，還是最后merge 產生的結果文件，都是可以壓縮的。壓縮的好處在于，通過壓縮減少寫入讀出磁盤的數據量。對中間結果非常大，磁盤速度成為map 執行瓶頸的job ，尤其有用。控制map 中間結果是否使用壓縮的參數為： mapred.compress.map.output (true/false) 。將這個參數設置為true 時，那么map 在寫中間結果時，就會將數據壓縮后再寫入磁盤，讀結果時也會采用先解壓后讀取數據。這樣做的后果就是：寫入磁盤的中間結果數據量會變少，但是cpu 會消耗一些用來壓縮和解壓。所以這種方式通常適合job 中間結果非常大，瓶頸不在cpu ，而是在磁盤的讀寫的情況。說的直白一些就是用cpu 換IO 。根據觀察，通常大部分的作業cpu 都不是瓶頸，除非運算邏輯異常復雜。所以對中間結果采用壓縮通常來說是有收益的。以下是一個wordcount 中間結果采用壓縮和不采用壓縮產生的map 中間結果本地磁盤讀寫的數據量對比：

map 中間結果不壓縮：

map 中間結果壓縮：

可以看出，同樣的job ，同樣的數據，在采用壓縮的情況下，map 中間結果能縮小將近10 倍，如果map 的瓶頸在磁盤，那么job 的性能提升將會非常可觀。

當采用map 中間結果壓縮的情況下，用戶還可以選擇壓縮時???用哪種壓縮格式進行壓縮，現在Hadoop 支持的壓縮格式有： GzipCodec ， LzoCodec ， BZip2Codec ， LzmaCodec 等壓縮格式。通常來說，想要達到比較平衡的 cpu 和磁盤壓縮比， LzoCodec 比較適合。但也要取決于 job 的具體情況。用戶若想要自行選擇中間結果的壓縮算法，可以設置配置參數： mapred.map.output.compression.codec =org.apache.hadoop.io.compress.DefaultCodec 或者其他用戶自行選擇的壓縮方式。

1.2 Map side 相關參數調優

2 Reduce side tuning 參數

2.1 ReduceTask 運行內部原理

reduce 的運行是分成三個階段的。分別為 copy->sort->reduce 。由于 job 的每一個 map 都會根據 reduce(n) 數將數據分成 map 輸出結果分成 n 個 partition ，所以 map 的中間結果中是有可能包含每一個 reduce 需要處理的部分數據的。所以，為了優化 reduce 的執行時間， hadoop 中是等 job 的第一個 map 結束后，所有的 reduce 就開始嘗試從完成的 map 中下載該 reduce 對應的 partition 部分數據。這個過程就是通常所說的 shuffle ，也就是 copy 過程。

       Reduce task 在做 shuffle 時，實際上就是從不同的已經完成的 map 上去下載屬于自己這個 reduce 的部分數據，由于 map 通常有許多個，所以對一個 reduce 來說，下載也可以是并行的從多個 map 下載，這個并行度是可以調整的，調整參數為： mapred.reduce.parallel.copies （ default 5 ）。默認情況下，每個只會有 5 個并行的下載線程在從 map 下數據，如果一個時間段內 job 完成的 map 有 100 個或者更多，那么 reduce 也最多只能同時下載 5 個 map 的數據，所以這個參數比較適合 map 很多并且完成的比較快的 job 的情況下調大，有利于 reduce 更快的獲取屬于自己部分的數據。

       reduce 的每一個下載線程在下載某個 map 數據的時候，有可能因為那個 map 中間結果所在機器發生錯誤，或者中間結果的文件丟失，或者網絡瞬斷等等情況，這樣 reduce 的下載就有可能失敗，所以 reduce 的下載線程并不會無休止的等待下去，當一定時間后下載仍然失敗，那么下載線程就會放棄這次下載，并在隨后嘗試從另外的地方下載（因為這段時間 map 可能重跑）。所以 reduce 下載線程的這個最大的下載時間段是可以調整的，調整參數為： mapred.reduce.copy.backoff （ default 300 秒）。如果集群環境的網絡本身是瓶頸，那么用戶可以通過調大這個參數來避免 reduce 下載線程被誤判為失敗的情況。不過在網絡環境比較好的情況下，沒有必要調整。通常來說專業的集群網絡不應該有太大問題，所以這個參數需要調整的情況不多。

       Reduce 將 map 結果下載到本地時，同樣也是需要進行 merge 的，所以 io.sort.factor 的配置選項同樣會影響 reduce 進行 merge 時的行為，該參數的詳細介紹上文已經提到，當發現 reduce 在 shuffle 階段 iowait 非常的高的時候，就有可能通過調大這個參數來加大一次 merge 時的并發吞吐，優化 reduce 效率。

       Reduce 在 shuffle 階段對下載來的 map 數據，并不是立刻就寫入磁盤的，而是會先緩存在內存中，然后當使用內存達到一定量的時候才刷入磁盤。這個內存大小的控制就不像 map 一樣可以通過 io.sort.mb 來設定了，而是通過另外一個參數來設置： mapred.job.shuffle.input.buffer.percent （ default 0.7 ），這個參數其實是一個百分比，意思是說， shuffile 在 reduce 內存中的數據最多使用內存量為： 0.7 × maxHeap of reduce task 。也就是說，如果該 reduce task 的最大 heap 使用量（通常通過 mapred.child.java.opts 來設置，比如設置為 -Xmx1024m ）的一定比例用來緩存數據。默認情況下， reduce 會使用其 heapsize 的 70% 來在內存中緩存數據。如果 reduce 的 heap 由于業務原因調整的比較大，相應的緩存大小也會變大，這也是為什么 reduce 用來做緩存的參數是一個百分比，而不是一個固定的值了。

假設 mapred.job.shuffle.input.buffer.percent 為 0.7 ， reduce task 的 max heapsize 為 1G ，那么用來做下載數據緩存的內存就為大概 700MB 左右，這 700M 的內存，跟 map 端一樣，也不是要等到全部寫滿才會往磁盤刷的，而是當這 700M 中被使用到了一定的限度（通常是一個百分比），就會開始往磁盤刷。這個限度閾值也是可以通過 job 參數來設定的，設定參數為： mapred.job.shuffle.merge.percent （ default 0.66 ）。如果下載速度很快，很容易就把內存緩存撐大，那么調整一下這個參數有可能會對 reduce 的性能有所幫助。

當 reduce 將所有的 map 上對應自己 partition 的數據下載完成后，就會開始真正的 reduce 計算階段（中間有個 sort 階段通常時間非常短，幾秒鐘就完成了，因為整個下載階段就已經是邊下載邊 sort ，然后邊 merge 的）。當 reduce task 真正進入 reduce 函數的計算階段的時候，有一個參數也是可以調整 reduce 的計算行為。也就是： mapred.job.reduce.input.buffer.percent （ default 0.0 ）。由于 reduce 計算時肯定也是需要消耗內存的，而在讀取 reduce 需要的數據時，同樣是需要內存作為 buffer ，這個參數是控制，需要多少的內存百分比來作為 reduce 讀已經 sort 好的數據的 buffer 百分比。默認情況下為 0 ，也就是說，默認情況下， reduce 是全部從磁盤開始讀處理數據。如果這個參數大于 0 ，那么就會有一定量的數據被緩存在內存并輸送給 reduce ，當 reduce 計算邏輯消耗內存很小時，可以分一部分內存用來緩存數據，反正 reduce 的內存閑著也是閑著。

2.2 Reduce side 相關參數調優