HADOOP-4845. Modify the reduce input byte counter to record only the

compressed size and add a human-readable label. Contributed by Yongqiang He


git-svn-id: https://svn.apache.org/repos/asf/hadoop/core/branches/branch-0.20@727006 13f79535-47bb-0310-9956-ffa450edef68

CHANGES.txt

@@ -428,6 +428,9 @@ Release 0.20.0 - Unreleased
     HADOOP-3921. Fixed clover (code coverage) target to work with JDK 6.
     (tomwhite via nigel)
 
+    HADOOP-4845. Modify the reduce input byte counter to record only the
+    compressed size and add a human-readable label. (Yongqiang He via cdouglas)
+
 Release 0.19.1 - Unreleased
 
   IMPROVEMENTS

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

@@ -111,8 +111,8 @@ class ReduceTask extends Task {
   private Progress copyPhase;
   private Progress sortPhase;
   private Progress reducePhase;
-  private Counters.Counter reduceInputBytes = 
-    getCounters().findCounter(Counter.REDUCE_INPUT_BYTES);
+  private Counters.Counter reduceShuffleBytes = 
+    getCounters().findCounter(Counter.REDUCE_SHUFFLE_BYTES);
   private Counters.Counter reduceInputKeyCounter = 
     getCounters().findCounter(Counter.REDUCE_INPUT_GROUPS);
   private Counters.Counter reduceInputValueCounter = 
@@ -380,7 +380,6 @@ class ReduceTask extends Task {
         throw new IOException("Task: " + getTaskID() + 
             " - The reduce copier failed", reduceCopier.mergeThrowable);
       }
-      reduceInputBytes.increment(reduceCopier.reducerInputBytes);
     }
     copyPhase.complete();                         // copy is already complete
     setPhase(TaskStatus.Phase.SORT);
@@ -919,7 +918,7 @@ class ReduceTask extends Task {
       
       byte[] data;
       final boolean inMemory;
-      long size;
+      long compressedSize;
       
       public MapOutput(TaskID mapId, TaskAttemptID mapAttemptId, 
                        Configuration conf, Path file, long size) {
@@ -928,14 +927,14 @@ class ReduceTask extends Task {
         
         this.conf = conf;
         this.file = file;
-        this.size = size;
+        this.compressedSize = size;
         
         this.data = null;
         
         this.inMemory = false;
       }
       
-      public MapOutput(TaskID mapId, TaskAttemptID mapAttemptId, byte[] data) {
+      public MapOutput(TaskID mapId, TaskAttemptID mapAttemptId, byte[] data, int compressedLength) {
         this.mapId = mapId;
         this.mapAttemptId = mapAttemptId;
         
@@ -943,7 +942,7 @@ class ReduceTask extends Task {
         this.conf = null;
         
         this.data = data;
-        this.size = data.length;
+        this.compressedSize = compressedLength;
         
         this.inMemory = true;
       }
@@ -1262,7 +1261,7 @@ class ReduceTask extends Task {
         }
         
         // The size of the map-output
-        long bytes = mapOutput.size;
+        long bytes = mapOutput.compressedSize;
         
         // lock the ReduceTask while we do the rename
         synchronized (ReduceTask.this) {
@@ -1467,7 +1466,7 @@ class ReduceTask extends Task {
         byte[] shuffleData = new byte[mapOutputLength];
         MapOutput mapOutput = 
           new MapOutput(mapOutputLoc.getTaskId(), 
-                        mapOutputLoc.getTaskAttemptId(), shuffleData);
+                        mapOutputLoc.getTaskAttemptId(), shuffleData, compressedLength);
         
         int bytesRead = 0;
         try {
@@ -1741,12 +1740,10 @@ class ReduceTask extends Task {
       return numInFlight > maxInFlight;
     }
     
-    long           reducerInputBytes = 0;
     
     public boolean fetchOutputs() throws IOException {
       int totalFailures = 0;
       int            numInFlight = 0, numCopied = 0;
-      long           bytesTransferred = 0;
       DecimalFormat  mbpsFormat = new DecimalFormat("0.00");
       final Progress copyPhase = 
         reduceTask.getProgress().phase();
@@ -1938,12 +1935,11 @@ class ReduceTask extends Task {
             if (cr.getSuccess()) {  // a successful copy
               numCopied++;
               lastProgressTime = System.currentTimeMillis();
-              bytesTransferred += cr.getSize();
-              reducerInputBytes += cr.getSize();
+              reduceShuffleBytes.increment(cr.getSize());
                 
               long secsSinceStart = 
                 (System.currentTimeMillis()-startTime)/1000+1;
-              float mbs = ((float)bytesTransferred)/(1024*1024);
+              float mbs = ((float)reduceShuffleBytes.getCounter())/(1024*1024);
               float transferRate = mbs/secsSinceStart;
                 
               copyPhase.startNextPhase();

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

@@ -63,7 +63,7 @@ abstract class Task implements Writable, Configurable {
     COMBINE_INPUT_RECORDS,
     COMBINE_OUTPUT_RECORDS,
     REDUCE_INPUT_GROUPS,
-    REDUCE_INPUT_BYTES, 
+    REDUCE_SHUFFLE_BYTES,
     REDUCE_INPUT_RECORDS,
     REDUCE_OUTPUT_RECORDS,
     REDUCE_SKIPPED_GROUPS,

src/mapred/org/apache/hadoop/mapred/Task_Counter.properties

@@ -10,6 +10,7 @@ MAP_SKIPPED_RECORDS.name=      Map skipped records
 COMBINE_INPUT_RECORDS.name=    Combine input records
 COMBINE_OUTPUT_RECORDS.name=   Combine output records
 REDUCE_INPUT_GROUPS.name=      Reduce input groups
+REDUCE_SHUFFLE_BYTES.name=	   Reduce shuffle bytes
 REDUCE_INPUT_RECORDS.name=     Reduce input records
 REDUCE_OUTPUT_RECORDS.name=    Reduce output records
 REDUCE_SKIPPED_RECORDS.name=   Reduce skipped records