You are viewing a plain text version of this content. The canonical link for it is here.
Posted to common-commits@hadoop.apache.org by zs...@apache.org on 2008/12/11 06:57:03 UTC

svn commit: r725588 - in /hadoop/core/trunk/src/mapred/org/apache/hadoop/mapred: ReduceTask.java Task.java

Author: zshao
Date: Wed Dec 10 21:57:03 2008
New Revision: 725588

URL: http://svn.apache.org/viewvc?rev=725588&view=rev
Log:
HADOOP-4749. Added a new counter REDUCE_INPUT_BYTES. (Yongqiang He via zshao)

Modified:
    hadoop/core/trunk/src/mapred/org/apache/hadoop/mapred/ReduceTask.java
    hadoop/core/trunk/src/mapred/org/apache/hadoop/mapred/Task.java

Modified: hadoop/core/trunk/src/mapred/org/apache/hadoop/mapred/ReduceTask.java
URL: http://svn.apache.org/viewvc/hadoop/core/trunk/src/mapred/org/apache/hadoop/mapred/ReduceTask.java?rev=725588&r1=725587&r2=725588&view=diff
==============================================================================
--- hadoop/core/trunk/src/mapred/org/apache/hadoop/mapred/ReduceTask.java (original)
+++ hadoop/core/trunk/src/mapred/org/apache/hadoop/mapred/ReduceTask.java Wed Dec 10 21:57:03 2008
@@ -107,6 +107,8 @@
   private Progress copyPhase;
   private Progress sortPhase;
   private Progress reducePhase;
+  private Counters.Counter reduceInputBytes = 
+    getCounters().findCounter(Counter.REDUCE_INPUT_BYTES);
   private Counters.Counter reduceInputKeyCounter = 
     getCounters().findCounter(Counter.REDUCE_INPUT_GROUPS);
   private Counters.Counter reduceInputValueCounter = 
@@ -372,6 +374,7 @@
         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);
@@ -1624,6 +1627,8 @@
       return numInFlight > maxInFlight;
     }
     
+    long           reducerInputBytes = 0;
+    
     public boolean fetchOutputs() throws IOException {
       //The map for (Hosts, List of MapIds from this Host)
       HashMap<String, List<MapOutputLocation>> mapLocations = 
@@ -1852,6 +1857,7 @@
               numCopied++;
               lastProgressTime = System.currentTimeMillis();
               bytesTransferred += cr.getSize();
+              reducerInputBytes += cr.getSize();
                 
               long secsSinceStart = 
                 (System.currentTimeMillis()-startTime)/1000+1;

Modified: hadoop/core/trunk/src/mapred/org/apache/hadoop/mapred/Task.java
URL: http://svn.apache.org/viewvc/hadoop/core/trunk/src/mapred/org/apache/hadoop/mapred/Task.java?rev=725588&r1=725587&r2=725588&view=diff
==============================================================================
--- hadoop/core/trunk/src/mapred/org/apache/hadoop/mapred/Task.java (original)
+++ hadoop/core/trunk/src/mapred/org/apache/hadoop/mapred/Task.java Wed Dec 10 21:57:03 2008
@@ -65,6 +65,7 @@
     COMBINE_INPUT_RECORDS,
     COMBINE_OUTPUT_RECORDS,
     REDUCE_INPUT_GROUPS,
+    REDUCE_INPUT_BYTES, 
     REDUCE_INPUT_RECORDS,
     REDUCE_OUTPUT_RECORDS,
     REDUCE_SKIPPED_GROUPS,