svn commit: r758180 - in /hadoop/core/trunk: CHANGES.txt src/mapred/org/apache/hadoop/mapred/Merger.java

[dd...@apache.org]

Author: ddas
Date: Wed Mar 25 09:00:52 2009
New Revision: 758180

URL: http://svn.apache.org/viewvc?rev=758180&view=rev
Log:
HADOOP-5210. Solves a problem in the progress report of the reduce task. Contributed by Ravi Gummadi.

Modified:
    hadoop/core/trunk/CHANGES.txt
    hadoop/core/trunk/src/mapred/org/apache/hadoop/mapred/Merger.java

Modified: hadoop/core/trunk/CHANGES.txt
URL: http://svn.apache.org/viewvc/hadoop/core/trunk/CHANGES.txt?rev=758180&r1=758179&r2=758180&view=diff
==============================================================================
--- hadoop/core/trunk/CHANGES.txt (original)
+++ hadoop/core/trunk/CHANGES.txt Wed Mar 25 09:00:52 2009
@@ -310,6 +310,9 @@
 
     HADOOP-5507. Fix JMXGet javadoc warnings.  (Boris Shkolnik via szetszwo)
 
+    HADOOP-5210. Solves a problem in the progress report of the reduce task.
+    (Ravi Gummadi via ddas)
+
 Release 0.20.0 - Unreleased
 
   INCOMPATIBLE CHANGES

Modified: hadoop/core/trunk/src/mapred/org/apache/hadoop/mapred/Merger.java
URL: http://svn.apache.org/viewvc/hadoop/core/trunk/src/mapred/org/apache/hadoop/mapred/Merger.java?rev=758180&r1=758179&r2=758180&view=diff
==============================================================================
--- hadoop/core/trunk/src/mapred/org/apache/hadoop/mapred/Merger.java (original)
+++ hadoop/core/trunk/src/mapred/org/apache/hadoop/mapred/Merger.java Wed Mar 25 09:00:52 2009
@@ -363,6 +363,7 @@
           new ArrayList<Segment<K, V>>();
         int segmentsConsidered = 0;
         int numSegmentsToConsider = factor;
+        long startBytes = 0; // starting bytes of segments of this merge
         while (true) {
           //extract the smallest 'factor' number of segments  
           //Call cleanup on the empty segments (no key/value data)
@@ -375,8 +376,8 @@
             long startPos = segment.getPosition();
             boolean hasNext = segment.next();
             long endPos = segment.getPosition();
-            totalBytesProcessed += endPos - startPos;
-            mergeProgress.set(totalBytesProcessed * progPerByte);
+            startBytes += endPos - startPos;
+            
             if (hasNext) {
               segmentsToMerge.add(segment);
               segmentsConsidered++;
@@ -406,6 +407,12 @@
         //if we have lesser number of segments remaining, then just return the
         //iterator, else do another single level merge
         if (numSegments <= factor) {
+          // Reset totalBytesProcessed to track the progress of the final merge.
+          // This is considered the progress of the reducePhase, the 3rd phase
+          // of reduce task. Currently totalBytesProcessed is not used in sort
+          // phase of reduce task(i.e. when intermediate merges happen).
+          totalBytesProcessed = startBytes;
+          
           //calculate the length of the remaining segments. Required for 
           //calculating the merge progress
           long totalBytes = 0;