Yarn在Shuffle階段內存不足問題(error in shuffle in fetcher)

最近在使用MR跑一個任務的時候shuffle階段出現OOM，這個問題以前歷來沒有遇到過，上網找了一下，發現網友也遇到過想似的問題，如下是轉載的該問題的解決方法：

原文地址：http://blog.csdn.net/bigdatahappy/article/details/39295657

=====================================================================

在Hadoop集羣（CDH4.4, Mv2即Yarn框架）使用過程當中，發現處理大數據集時程序報出以下錯誤：

2016-12-15 08:10:57,726 WARN [main] org.apache.hadoop.mapred.YarnChild: Exception running child : org.apache.hadoop.mapreduce.task.reduce.Shuffle$ShuffleError: error in shuffle in fetcher#18
	at org.apache.hadoop.mapreduce.task.reduce.Shuffle.run(Shuffle.java:134)
	at org.apache.hadoop.mapred.ReduceTask.run(ReduceTask.java:377)
	at org.apache.hadoop.mapred.YarnChild$2.run(YarnChild.java:164)
	at java.security.AccessController.doPrivileged(Native Method)
	at javax.security.auth.Subject.doAs(Subject.java:422)
	at org.apache.hadoop.security.UserGroupInformation.doAs(UserGroupInformation.java:1657)
	at org.apache.hadoop.mapred.YarnChild.main(YarnChild.java:158)
Caused by: java.lang.OutOfMemoryError: Java heap space
	at org.apache.hadoop.io.BoundedByteArrayOutputStream.<init>(BoundedByteArrayOutputStream.java:56)
	at org.apache.hadoop.io.BoundedByteArrayOutputStream.<init>(BoundedByteArrayOutputStream.java:46)
	at org.apache.hadoop.mapreduce.task.reduce.InMemoryMapOutput.<init>(InMemoryMapOutput.java:63)
	at org.apache.hadoop.mapreduce.task.reduce.MergeManagerImpl.unconditionalReserve(MergeManagerImpl.java:305)
	at org.apache.hadoop.mapreduce.task.reduce.MergeManagerImpl.reserve(MergeManagerImpl.java:295)
	at org.apache.hadoop.mapreduce.task.reduce.Fetcher.copyMapOutput(Fetcher.java:514)
	at org.apache.hadoop.mapreduce.task.reduce.Fetcher.copyFromHost(Fetcher.java:336)
	at org.apache.hadoop.mapreduce.task.reduce.Fetcher.run(Fetcher.java:193)

Google一番後竟然無果！程序等着運行，老闆催着要結果，沒有大師協助，只能開始艱難地自救了！認真分析，求助於源代碼！ 首先發現的一點是：map任務百分比一直在遞增，出現reduce任務以後，每隔一段時間報一個相似上面的錯誤，reduce從0%從新開始，而Map任務繼續前進，reduce處理一段後再報，再從0開始。累計到第四個報錯後即整個Application宣佈Fail。 根據這一點，大體能夠得出這樣的結論： reduce任務每次嘗試都失敗了，失敗後從新開始； reduce任務失敗累計4次後整個Application退出，應該是設置了最大重試次數之類的配置項。 map任務與reduce任務是隔離的，之間不會干擾。這個從map、reduce任務原理也能夠了解到。 基於這一點，首先查詢到map-site.xml中的配置項mapreduce.reduce.maxattempts，表示Reduce Task最大失敗嘗試次數，這個配置默認是4，調整到400後接着嘗試。 mapreduce.reduce.maxattempts起了做用，可是報錯依然不斷，不過不會4次報錯就結束了，map進度一直向前，map到達100%後，reduce依然重複報錯的節奏。是時候查查這裏報錯的類究竟在作啥了。

org.apache.hadoop.mapreduce.task.reduce.Fetcher類位於hadoop-mapreduce-client-core-2.0.0-cdh4.4.0.jar包中，Maven的話在pom.xml添加以下配置，能夠獲取該包以及源碼：

<dependency>
    <groupId >org.apache.hadoop</ groupId>
    <artifactId >hadoop-mapreduce -client-core</ artifactId>
    <version >2.0.0-cdh4.4.0</ version>
</dependency>

問題的入口是run中的：

// Shuffle

copyFromHost(host);

跟蹤到copyMapOutput，是要準備從Map節點本地拷貝map的output進行shuffle。其中出錯點：

// Get the location for the map output – either in-memory or on-disk

mapOutput = merger.reserve(mapId, decompressedLength, id );

merger指向了MergeManagerImpl對象，調用其reserve函數，而這個函數中定義了shuffle的處理方式，是將output塞入內存(InMemoryMapOutput)仍是放在磁盤上慢慢作(OnDiskMapOutput)？ 從咱們這邊的出錯信息，顯然能夠看到任務選擇了InMemoryMapOutput，在檢查爲何做出這樣的選擇前，咱們看看map的輸出結果到底有多大：
 

shell>cd /data/1/mrlocal/yarn/local/usercache/hdfs/appcache/application_1385983958793_0001/output 

shell>du -sh * | grep _r_ 7.3G attempt_1385983958793_0001_r_000000_1

6.5G attempt_1385983958793_0001_r_000000_12

5.2G attempt_1385983958793_0001_r_000000_5

5.8G attempt_1385983958793_0001_r_000000_7

這樣大的輸出放到內存裏，顯然要OOM了，能夠有兩種選擇，它爲何不選擇OnDiskMapOutput呢？

以下這段很顯然是關鍵所在：
 

if (!canShuffleToMemory(requestedSize)) { 

    LOG.info(mapId + 「: Shuffling to disk since 」 + requestedSize + 」 is greater than maxSingleShuffleLimit (」 + maxSingleShuffleLimit + 「)」 ); 

    return new OnDiskMapOutput(mapId, reduceId, this , requestedSize, jobConf, mapOutputFile , fetcher, true); 

}

再看canShuffleToMemory：

private boolean canShuffleToMemory( long requestedSize) { return (requestedSize < maxSingleShuffleLimit); }

requestedSize從源碼上並不能清楚瞭解其真實含義，問題最終落在maxSingleShuffleLimit這個參數的含義和來源上，進一步細查能夠發現其來源：

this.maxSingleShuffleLimit = (long)( memoryLimit * singleShuffleMemoryLimitPercent);

兩個變量的取值：

// Allow unit tests to fix Runtime memory this.

memoryLimit = (long)(jobConf.getLong(MRJobConfig. REDUCE_MEMORY_TOTAL_BYTES, Math. min(Runtime.getRuntime ().maxMemory(), Integer.MAX_VALUE)) * maxInMemCopyUse); 

final float singleShuffleMemoryLimitPercent = jobConf.getFloat(MRJobConfig. SHUFFLE_MEMORY_LIMIT_PERCENT, DEFAULT_SHUFFLE_MEMORY_LIMIT_PERCENT );

singleShuffleMemoryLimitPercent 取的是mapreduce.reduce.shuffle.memory.limit.percent這個配置的取值，官網給出的解釋是：

Expert: Maximum percentage of the in-memory limit that a single shuffle can consume

單個shuffle可以消耗的內存佔reduce全部內存的比例，默認值爲0.25。Expert」專家模式」，說的很唬人。。

那麼下降mapreduce.reduce.shuffle.memory.limit.percent這個參數應該可使得程序選擇OnDiskMapout而不是選擇InMemory，調低至0.06在測試，順利執行，再也不報錯。

收穫：選擇了最新的框架，意味着會遇到最新的問題。無助時，瞭解原理，查詢源碼，總能找到想要的答案。

遺留：

1.查看源碼，不少不清晰的地方都略過了，其中memoryLimit的取值，即reduce全部可以使用的內存，實際取值如何肯定，須要進一步找尋答案。

2.如何控制mapreduce.reduce.shuffle.memory.limit.percent使得咱們可以使用合理的配置來最大化的使用內存，待續。