Dongwook Kwon created MAPREDUCE-6108:
----------------------------------------

             Summary: ShuffleError OOM while reserving by MergeManagerImpl
                 Key: MAPREDUCE-6108
                 URL: https://issues.apache.org/jira/browse/MAPREDUCE-6108
             Project: Hadoop Map/Reduce
          Issue Type: Bug
    Affects Versions: 2.5.1, 2.4.1, 2.5.0, 2.4.0
            Reporter: Dongwook Kwon
            Priority: Minor


Shuffle has OOM issue from time to time.  

Such as this email reported.
http://mail-archives.apache.org/mod_mbox/hadoop-mapreduce-dev/201408.mbox/%3ccabwxxjnk-on0xtrmurijd8sdgjjtamsvqw2czpm3oekj3ym...@mail.gmail.com%3E

{code}

Error: org.apache.hadoop.mapreduce.task.reduce.Shuffle$ShuffleError: error in 
shuffle in fetcher#14
        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:167)
        at java.security.AccessController.doPrivileged(Native Method)
        at javax.security.auth.Subject.doAs(Subject.java:415)
        at 
org.apache.hadoop.security.UserGroupInformation.doAs(UserGroupInformation.java:1548)
        at org.apache.hadoop.mapred.YarnChild.main(YarnChild.java:162)
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:297)
        at 
org.apache.hadoop.mapreduce.task.reduce.MergeManagerImpl.reserve(MergeManagerImpl.java:287)
        at 
org.apache.hadoop.mapreduce.task.reduce.Fetcher.copyMapOutput(Fetcher.java:411)
        at 
org.apache.hadoop.mapreduce.task.reduce.Fetcher.copyFromHost(Fetcher.java:341)
        at org.apache.hadoop.mapreduce.task.reduce.Fetcher.run(Fetcher.java:165)

{code}

Lowering mapreduce.reduce.shuffle.input.buffer.percent value mitigate the 
issue. However depending on the data and the memory system had, the issue comes 
back.

>From my test, when it's happening , the issue is very constant, memory foot 
>print, and the point OOM happens was the same, regardless of the value of 
>mapreduce.reduce.shuffle.input.buffer.percent( my test had default 0.7).  


Here is what I found.

According to MergeManagerImpl which implemented by 
https://issues.apache.org/jira/browse/MAPREDUCE-4808, it appears the reserve 
method deliberately allows just one thread(fetcher) to go over "memoryLimit" by 
checking the condition (usedMemory > memoryLimit) instead of (usedMemory + 
requestedSize > memoryLimit) to prevent stalling all fetchers issue as comment 
indicated. This seems working well most of times. However when the one fetcher 
tries to reserver usedMemory + requestedSize more than 
memoryLimit(Runtime.getRuntime().maxMemory()), I think there is OOM issue.

{code}
 @Override
public synchronized MapOutput<K,V> reserve(TaskAttemptID mapId,
long requestedSize,
int fetcher
) throws IOException {
if (!canShuffleToMemory(requestedSize)) {
LOG.info(mapId + ": Shuffling to disk since " + requestedSize +
" is greater than maxSingleShuffleLimit (" +
maxSingleShuffleLimit + ")");
return new OnDiskMapOutput<K,V>(mapId, reduceId, this, requestedSize,
jobConf, mapOutputFile, fetcher, true);
}
// Stall shuffle if we are above the memory limit
// It is possible that all threads could just be stalling and not make
// progress at all. This could happen when:
//
// requested size is causing the used memory to go above limit &&
// requested size < singleShuffleLimit &&
// current used size < mergeThreshold (merge will not get triggered)
//
// To avoid this from happening, we allow exactly one thread to go past
// the memory limit. We check (usedMemory > memoryLimit) and not
// (usedMemory + requestedSize > memoryLimit). When this thread is done
// fetching, this will automatically trigger a merge thereby unlocking
// all the stalled threads
if (usedMemory > memoryLimit) {
LOG.debug(mapId + ": Stalling shuffle since usedMemory (" + usedMemory
+ ") is greater than memoryLimit (" + memoryLimit + ")." +
" CommitMemory is (" + commitMemory + ")");
return null;
}
// Allow the in-memory shuffle to progress
LOG.debug(mapId + ": Proceeding with shuffle since usedMemory ("
+ usedMemory + ") is lesser than memoryLimit (" + memoryLimit + ")."
+ "CommitMemory is (" + commitMemory + ")");
return unconditionalReserve(mapId, requestedSize, true);
}

{code}

https://github.com/apache/hadoop-common/blob/trunk/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapreduce/task/reduce/MergeManagerImpl.java#L256

When the one fetcher tries to reserve (usedMemory + requestedSize > 
memoryLimit), depending on the memory the reducer has,  
BoundedByteArrayOutputStream has the OOM issue at 

{code}
 public BoundedByteArrayOutputStream(int capacity, int limit) {
this(new byte[capacity], 0, limit);
}
{code}

https://github.com/apache/hadoop-common/blob/trunk/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/io/BoundedByteArrayOutputStream.java#L56

memoryLimit is Runtime.getRuntime().maxMemory(), 
MRJobConfig.REDUCE_MEMORY_TOTAL_BYTES seems for unit test.

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


It explains why lowering mapreduce.reduce.shuffle.input.buffer.percent value 
resolves this issue and why the same setting sometimes works and doesn't.

But I wasn't sure this is correct and what is the expected behavior for 
stalling fetchers issue to fix OOM issue as commented pointed out.



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

Reply via email to