HADOOP-1698. Fixes performance problems in the map output sorting with

many reducers. Contributed by Devaraj Das.


git-svn-id: https://svn.apache.org/repos/asf/lucene/hadoop/trunk@565628 13f79535-47bb-0310-9956-ffa450edef68

CHANGES.txt

@@ -520,6 +520,9 @@ Branch 0.14 (unreleased changes)
 152. HADOOP-1629.  Added a upgrade test for HADOOP-1134.
      (Raghu Angadi via nigel)
 
+153. HADOOP-1698.  Fix performance problems on map output sorting for jobs
+     with large numbers of reduces. (Devaraj Das via omalley)
+
 Release 0.13.0 - 2007-06-08
 
  1. HADOOP-1047.  Fix TestReplication to succeed more reliably.

src/java/org/apache/hadoop/io/SequenceFile.java

@@ -648,9 +648,10 @@ public class SequenceFile {
     long lastSyncPos;                     // position of last sync
     byte[] sync;                          // 16 random bytes
     {
-      try {                                       // use hash of uid + host
+      try {                                       
         MessageDigest digester = MessageDigest.getInstance("MD5");
-        digester.update((new UID()+"@"+InetAddress.getLocalHost()).getBytes());
+        long time = System.currentTimeMillis();
+        digester.update((new UID()+"@"+time).getBytes());
         sync = digester.digest();
       } catch (Exception e) {
         throw new RuntimeException(e);

src/java/org/apache/hadoop/mapred/BasicTypeSorterBase.java

@@ -50,16 +50,21 @@ abstract class BasicTypeSorterBase implements BufferSorter {
   //4 for indices into startOffsets array in the
   //pointers array (ignored the partpointers list itself)
   static private final int BUFFERED_KEY_VAL_OVERHEAD = 16;
+  static private final int INITIAL_ARRAY_SIZE = 5;
+  //we maintain the max lengths of the key/val that we encounter.  During 
+  //iteration of the sorted results, we will create a DataOutputBuffer to
+  //return the keys. The max size of the DataOutputBuffer will be the max
+  //keylength that we encounter. Expose this value to model memory more
+  //accurately.
+  private int maxKeyLength = 0;
+  private int maxValLength = 0;
+
   //Reference to the Progressable object for sending KeepAlive
   private Progressable reporter;
 
   //Implementation of methods of the SorterBase interface
   //
   public void configure(JobConf conf) {
-    startOffsets = new int[1024];
-    keyLengths = new int[1024];
-    valueLengths = new int[1024];
-    pointers = new int[1024];
     comparator = conf.getOutputKeyComparator();
   }
   
@@ -70,10 +75,16 @@ abstract class BasicTypeSorterBase implements BufferSorter {
   public void addKeyValue(int recordOffset, int keyLength, int valLength) {
     //Add the start offset of the key in the startOffsets array and the
     //length in the keyLengths array.
-    if (count == startOffsets.length)
+    if (startOffsets == null || count == startOffsets.length)
       grow();
     startOffsets[count] = recordOffset;
     keyLengths[count] = keyLength;
+    if (keyLength > maxKeyLength) {
+      maxKeyLength = keyLength;
+    }
+    if (valLength > maxValLength) {
+      maxValLength = valLength;
+    }
     valueLengths[count] = valLength;
     pointers[count] = count;
     count++;
@@ -85,14 +96,30 @@ abstract class BasicTypeSorterBase implements BufferSorter {
   }
 
   public long getMemoryUtilized() {
-    return (startOffsets.length) * BUFFERED_KEY_VAL_OVERHEAD;
+    //the total length of the arrays + the max{Key,Val}Length (this will be the 
+    //max size of the DataOutputBuffers during the iteration of the sorted
+    //keys).
+    if (startOffsets != null) {
+      return (startOffsets.length) * BUFFERED_KEY_VAL_OVERHEAD + 
+              maxKeyLength + maxValLength;
+    }
+    else { //nothing from this yet
+      return 0;
+    }
   }
 
   public abstract RawKeyValueIterator sort();
   
   public void close() {
-    //just set count to 0; we reuse the arrays
+    //set count to 0; also, we don't reuse the arrays since we want to maintain
+    //consistency in the memory model
     count = 0;
+    startOffsets = null;
+    keyLengths = null;
+    valueLengths = null;
+    pointers = null;
+    maxKeyLength = 0;
+    maxValLength = 0;
   }
   //A compare method that references the keyValBuffer through the indirect
   //pointers
@@ -106,7 +133,11 @@ abstract class BasicTypeSorterBase implements BufferSorter {
   }
   
   private void grow() {
-    int newLength = startOffsets.length * 3/2;
+    int currLength = 0;
+    if (startOffsets != null) {
+      currLength = startOffsets.length;
+    }
+    int newLength = (int)(currLength * 1.1) + 1;
     startOffsets = grow(startOffsets, newLength);
     keyLengths = grow(keyLengths, newLength);
     valueLengths = grow(valueLengths, newLength);
@@ -115,7 +146,9 @@ abstract class BasicTypeSorterBase implements BufferSorter {
   
   private int[] grow(int[] old, int newLength) {
     int[] result = new int[newLength];
-    System.arraycopy(old, 0, result, 0, old.length);
+    if(old != null) { 
+      System.arraycopy(old, 0, result, 0, old.length);
+    }
     return result;
   }
 } //BasicTypeSorterBase

src/java/org/apache/hadoop/mapred/MapTask.java

@@ -330,6 +330,9 @@ class MapTask extends Task {
       }
       
       synchronized (this) {
+        if (keyValBuffer == null) {
+          keyValBuffer = new DataOutputBuffer();
+        }
         //dump the key/value to buffer
         int keyOffset = keyValBuffer.getLength(); 
         key.write(keyValBuffer);
@@ -350,7 +353,9 @@ class MapTask extends Task {
           totalMem += sortImpl[i].getMemoryUtilized();
         if ((keyValBuffer.getLength() + totalMem) >= maxBufferSize) {
           sortAndSpillToDisk();
-          keyValBuffer.reset();
+          //we don't reuse the keyValBuffer. We want to maintain consistency
+          //in the memory model (for negligible performance loss).
+          keyValBuffer = null;
           for (int i = 0; i < partitions; i++) {
             sortImpl[i].close();
           }

src/java/org/apache/hadoop/mapred/MergeSorter.java

@@ -58,4 +58,12 @@ implements Comparator<IntWritable> {
   public int compare (IntWritable i, IntWritable j) {
     return super.compare(i.get(), j.get());
   }
+  
+  /** Add the extra memory that will be utilized by the sort method */
+  public long getMemoryUtilized() {
+    //this is memory that will be actually utilized (considering the temp
+    //array that will be allocated by the sort() method (mergesort))
+    return super.getMemoryUtilized() + super.count * 4; 
+  }
+
 }