HADOOP-5710. Counter MAP_INPUT_BYTES missing from new mapreduce api. Contributed by Amareshwari Sriramadasu.

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

CHANGES.txt

@@ -673,6 +673,9 @@ Trunk (unreleased changes)
     HADOOP-5252. Streaming overrides -inputformat option (Klaas Bosteels 
     via sharad)
 
+    HADOOP-5710. Counter MAP_INPUT_BYTES missing from new mapreduce api. 
+    (Amareshwari Sriramadasu via sharad)
+
 Release 0.20.1 - Unreleased
 
   INCOMPATIBLE CHANGES

src/contrib/vaidya/src/java/org/apache/hadoop/vaidya/statistics/job/JobStatistics.java

@@ -417,8 +417,6 @@ public class JobStatistics implements JobStatisticsInterface {
           mapTask.setValue(MapTaskKeys.INPUT_RECORDS, parts[1]);
         } else if (parts[0].equals("Map-Reduce Framework.Map output records")) {
           mapTask.setValue(MapTaskKeys.OUTPUT_RECORDS, parts[1]);
-        } else if (parts[0].equals("Map-Reduce Framework.Map input bytes")) {
-          mapTask.setValue(MapTaskKeys.INPUT_BYTES, parts[1]);
         } else if (parts[0].equals("Map-Reduce Framework.Map output bytes")) {
           mapTask.setValue(MapTaskKeys.OUTPUT_BYTES, parts[1]);
         } else if (parts[0].equals("Map-Reduce Framework.Combine input records")) {
@@ -427,6 +425,8 @@ public class JobStatistics implements JobStatisticsInterface {
           mapTask.setValue(MapTaskKeys.COMBINE_OUTPUT_RECORDS, parts[1]);
         } else if (parts[0].equals("Map-Reduce Framework.Spilled Records")) {
           mapTask.setValue(MapTaskKeys.SPILLED_RECORDS, parts[1]);
+        } else if (parts[0].equals("FileInputFormatCounters.BYTES_READ")) {
+          mapTask.setValue(MapTaskKeys.INPUT_BYTES, parts[1]);
         } else {
           System.out.println("MapCounterKey:<"+parts[0]+"> ==> NOT INCLUDED IN PERFORMANCE ADVISOR MAP TASK");
         }

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

@@ -36,6 +36,7 @@ import org.apache.hadoop.io.IntWritable;
 import org.apache.hadoop.io.Text;
 import org.apache.hadoop.io.Writable;
 import org.apache.hadoop.io.WritableUtils;
+import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
 import org.apache.hadoop.util.StringUtils;
 
 /**
@@ -398,6 +399,13 @@ public class Counters implements Writable, Iterable<Counters.Group> {
    * @return the counter for that name
    */
   public synchronized Counter findCounter(String group, String name) {
+    if (name.equals("MAP_INPUT_BYTES")) {
+      group = FileInputFormat.COUNTER_GROUP; 
+      name = FileInputFormat.BYTES_READ; 
+      LOG.warn("Counter name MAP_INPUT_BYTES is deprecated. " +
+               "Use FileInputFormatCounters as group name and " +
+               " BYTES_READ as counter name instead");
+    }
     return getGroup(group).getCounterForName(name);
   }
 
@@ -411,7 +419,7 @@ public class Counters implements Writable, Iterable<Counters.Group> {
    */
   @Deprecated
   public synchronized Counter findCounter(String group, int id, String name) {
-    return getGroup(group).getCounterForName(name);
+    return findCounter(group, name);
   }
 
   /**
@@ -432,7 +440,7 @@ public class Counters implements Writable, Iterable<Counters.Group> {
    * @param amount amount by which counter is to be incremented
    */
   public synchronized void incrCounter(String group, String counter, long amount) {
-    getGroup(group).getCounterForName(counter).increment(amount);
+    findCounter(group, counter).increment(amount);
   }
   
   /**

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

@@ -51,6 +51,7 @@ import org.apache.hadoop.io.serializer.Serializer;
 import org.apache.hadoop.mapred.IFile.Writer;
 import org.apache.hadoop.mapred.Merger.Segment;
 import org.apache.hadoop.mapred.SortedRanges.SkipRangeIterator;
+import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
 import org.apache.hadoop.mapreduce.TaskAttemptContext;
 import org.apache.hadoop.mapreduce.TaskCounter;
 import org.apache.hadoop.util.IndexedSortable;
@@ -157,7 +158,9 @@ class MapTask extends Task {
       throws IOException{
       rawIn = raw;
       inputRecordCounter = reporter.getCounter(TaskCounter.MAP_INPUT_RECORDS);
-      inputByteCounter = reporter.getCounter(TaskCounter.MAP_INPUT_BYTES);
+      inputByteCounter = reporter.getCounter(
+                           FileInputFormat.COUNTER_GROUP,
+                           FileInputFormat.BYTES_READ);
       this.reporter = reporter;
     }
 

src/mapred/org/apache/hadoop/mapreduce/TaskCounter.java

@@ -23,7 +23,6 @@ public enum TaskCounter {
   MAP_INPUT_RECORDS, 
   MAP_OUTPUT_RECORDS,
   MAP_SKIPPED_RECORDS,
-  MAP_INPUT_BYTES, 
   MAP_OUTPUT_BYTES,
   COMBINE_INPUT_RECORDS,
   COMBINE_OUTPUT_RECORDS,

src/mapred/org/apache/hadoop/mapreduce/TaskCounter.properties

@@ -3,7 +3,6 @@
 CounterGroupName=              Map-Reduce Framework
 
 MAP_INPUT_RECORDS.name=        Map input records
-MAP_INPUT_BYTES.name=          Map input bytes
 MAP_OUTPUT_RECORDS.name=       Map output records
 MAP_OUTPUT_BYTES.name=         Map output bytes
 MAP_SKIPPED_RECORDS.name=      Map skipped records

src/mapred/org/apache/hadoop/mapreduce/lib/input/FileInputFormat.java

@@ -49,6 +49,9 @@ import org.apache.hadoop.util.StringUtils;
  * not split-up and are processed as a whole by {@link Mapper}s.
  */
 public abstract class FileInputFormat<K, V> extends InputFormat<K, V> {
+  public static final String COUNTER_GROUP = 
+                                "FileInputFormatCounters";
+  public static final String BYTES_READ = "BYTES_READ";
 
   private static final Log LOG = LogFactory.getLog(FileInputFormat.class);
 

src/mapred/org/apache/hadoop/mapreduce/lib/input/LineRecordReader.java

@@ -28,9 +28,11 @@ import org.apache.hadoop.io.LongWritable;
 import org.apache.hadoop.io.Text;
 import org.apache.hadoop.io.compress.CompressionCodec;
 import org.apache.hadoop.io.compress.CompressionCodecFactory;
+import org.apache.hadoop.mapreduce.Counter;
 import org.apache.hadoop.mapreduce.InputSplit;
 import org.apache.hadoop.mapreduce.RecordReader;
 import org.apache.hadoop.mapreduce.TaskAttemptContext;
+import org.apache.hadoop.mapreduce.MapContext;
 import org.apache.hadoop.util.LineReader;
 import org.apache.commons.logging.LogFactory;
 import org.apache.commons.logging.Log;
@@ -49,10 +51,13 @@ public class LineRecordReader extends RecordReader<LongWritable, Text> {
   private int maxLineLength;
   private LongWritable key = null;
   private Text value = null;
+  private Counter inputByteCounter;
 
   public void initialize(InputSplit genericSplit,
                          TaskAttemptContext context) throws IOException {
     FileSplit split = (FileSplit) genericSplit;
+    inputByteCounter = ((MapContext)context).getCounter(
+      FileInputFormat.COUNTER_GROUP, FileInputFormat.BYTES_READ);
     Configuration job = context.getConfiguration();
     this.maxLineLength = job.getInt("mapred.linerecordreader.maxlength",
                                     Integer.MAX_VALUE);
@@ -101,6 +106,7 @@ public class LineRecordReader extends RecordReader<LongWritable, Text> {
         break;
       }
       pos += newSize;
+      inputByteCounter.increment(newSize);
       if (newSize < maxLineLength) {
         break;
       }

src/mapred/org/apache/hadoop/mapreduce/lib/input/SequenceFileRecordReader.java

@@ -25,7 +25,9 @@ import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.fs.FileSystem;
 import org.apache.hadoop.fs.Path;
 import org.apache.hadoop.io.*;
+import org.apache.hadoop.mapreduce.Counter;
 import org.apache.hadoop.mapreduce.InputSplit;
+import org.apache.hadoop.mapreduce.MapContext;
 import org.apache.hadoop.mapreduce.RecordReader;
 import org.apache.hadoop.mapreduce.TaskAttemptContext;
 
@@ -38,12 +40,16 @@ public class SequenceFileRecordReader<K, V> extends RecordReader<K, V> {
   private K key = null;
   private V value = null;
   protected Configuration conf;
+  private Counter inputByteCounter;
+  private long pos;
   
   @Override
   public void initialize(InputSplit split, 
                          TaskAttemptContext context
                          ) throws IOException, InterruptedException {
     FileSplit fileSplit = (FileSplit) split;
+    inputByteCounter = ((MapContext)context).getCounter(
+      FileInputFormat.COUNTER_GROUP, FileInputFormat.BYTES_READ);
     conf = context.getConfiguration();    
     Path path = fileSplit.getPath();
     FileSystem fs = path.getFileSystem(conf);
@@ -64,7 +70,8 @@ public class SequenceFileRecordReader<K, V> extends RecordReader<K, V> {
     if (!more) {
       return false;
     }
-    long pos = in.getPosition();
+    inputByteCounter.increment(in.getPosition()-pos);
+    pos = in.getPosition();
     key = (K) in.next(key);
     if (key == null || (pos >= end && in.syncSeen())) {
       more = false;

src/test/findbugsExcludeFile.xml

@@ -83,6 +83,19 @@
        <Class name="org.apache.hadoop.mapred.OutputCommitter" />
        <Bug pattern="BC_UNCONFIRMED_CAST" />
      </Match>
+     <!--
+       We intentionally cast from a super class to sub class.
+     -->
+     <Match>
+       <Class name="org.apache.hadoop.mapreduce.lib.input.LineRecordReader" />
+       <Method name="initialize" />
+       <Bug pattern="BC_UNCONFIRMED_CAST" />
+     </Match>
+     <Match>
+       <Class name="org.apache.hadoop.mapreduce.lib.input.SequenceFileRecordReader" />
+       <Method name="initialize" />
+       <Bug pattern="BC_UNCONFIRMED_CAST" />
+     </Match>
      <!--
        We intentionally do the get name from the inner class
      -->

src/test/mapred/org/apache/hadoop/mapred/TestCounters.java

@@ -70,7 +70,7 @@ public class TestCounters extends TestCase {
   }
   
   public void testCounters() throws IOException {
-    Enum[] keysWithResource = {TaskCounter.MAP_INPUT_BYTES, 
+    Enum[] keysWithResource = {TaskCounter.MAP_INPUT_RECORDS, 
                                TaskCounter.MAP_OUTPUT_BYTES};
     
     Enum[] keysWithoutResource = {myCounters.TEST1, myCounters.TEST2};

src/test/mapred/org/apache/hadoop/mapred/TestMiniMRDFSSort.java

@@ -30,7 +30,6 @@ import org.apache.hadoop.hdfs.MiniDFSCluster;
 import org.apache.hadoop.io.BytesWritable;
 import org.apache.hadoop.io.Text;
 import org.apache.hadoop.mapred.lib.NullOutputFormat;
-import org.apache.hadoop.mapreduce.TaskCounter;
 import org.apache.hadoop.fs.FileSystem;
 import org.apache.hadoop.fs.Path;
 import org.apache.hadoop.util.ToolRunner;
@@ -98,8 +97,10 @@ public class TestMiniMRDFSSort extends TestCase {
     Sort sort = new Sort();
     assertEquals(ToolRunner.run(job, sort, sortArgs), 0);
     Counters counters = sort.getResult().getCounters();
-    long mapInput = counters.findCounter(TaskCounter.MAP_INPUT_BYTES
-    ).getValue();
+    long mapInput = counters.findCounter(
+      org.apache.hadoop.mapreduce.lib.input.FileInputFormat.COUNTER_GROUP,
+      org.apache.hadoop.mapreduce.lib.input.FileInputFormat.BYTES_READ).
+      getValue();
     long hdfsRead = counters.findCounter(Task.FILESYSTEM_COUNTER_GROUP,
                                          "HDFS_BYTES_READ").getValue();
     // the hdfs read should be between 100% and 110% of the map input bytes

src/test/mapred/org/apache/hadoop/mapreduce/TestMapReduceLocal.java

@@ -118,6 +118,9 @@ public class TestMapReduceLocal extends TestCase {
                  out);
     Counters ctrs = job.getCounters();
     System.out.println("Counters: " + ctrs);
+    long mapIn = ctrs.findCounter(FileInputFormat.COUNTER_GROUP, 
+                                  FileInputFormat.BYTES_READ).getValue();
+    assertTrue(mapIn != 0);    
     long combineIn = ctrs.findCounter(COUNTER_GROUP,
                                       "COMBINE_INPUT_RECORDS").getValue();
     long combineOut = ctrs.findCounter(COUNTER_GROUP,