HADOOP-3001. Add job counters that measure the number of bytes

read and written to HDFS, S3, KFS, and local file systems.


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

CHANGES.txt

@@ -108,6 +108,9 @@ Trunk (unreleased changes)
     HADOOP-2551. More environment variables like HADOOP_NAMENODE_OPTS
     for better control of HADOOP_OPTS for each component. (rangadi)
 
+    HADOOP-3001. Add job counters that measure the number of bytes
+    read and written to HDFS, S3, KFS, and local file systems. (omalley)
+
   IMPROVEMENTS
 
     HADOOP-2655. Copy on write for data and metadata files in the 

src/java/org/apache/hadoop/dfs/DFSClient.java

@@ -70,6 +70,7 @@ class DFSClient implements FSConstants {
   private short defaultReplication;
   private SocketFactory socketFactory;
   private int socketTimeout;
+  private FileSystem.Statistics stats;
     
   /**
    * A map from name -> DFSOutputStream of files that are currently being
@@ -158,9 +159,11 @@ class DFSClient implements FSConstants {
   /** 
    * Create a new DFSClient connected to the given namenode server.
    */
-  public DFSClient(InetSocketAddress nameNodeAddr, Configuration conf)
+  public DFSClient(InetSocketAddress nameNodeAddr, Configuration conf,
+                   FileSystem.Statistics stats)
     throws IOException {
     this.conf = conf;
+    this.stats = stats;
     this.socketTimeout = conf.getInt("dfs.socket.timeout", 
                                      FSConstants.READ_TIMEOUT);
     this.socketFactory = NetUtils.getSocketFactory(conf, ClientProtocol.class);
@@ -186,6 +189,12 @@ class DFSClient implements FSConstants {
     this.leaseChecker.start();
   }
 
+  public DFSClient(InetSocketAddress nameNodeAddr, 
+                   Configuration conf) throws IOException {
+    this(nameNodeAddr, conf, 
+         FileSystem.getStatistics(DistributedFileSystem.class));
+  }
+
   private void checkOpen() throws IOException {
     if (!clientRunning) {
       IOException result = new IOException("Filesystem closed");
@@ -326,15 +335,17 @@ class DFSClient implements FSConstants {
   }
 
   public DFSInputStream open(String src) throws IOException {
-    return open(src, conf.getInt("io.file.buffer.size", 4096), true);
+    return open(src, conf.getInt("io.file.buffer.size", 4096), true, null);
   }
+
   /**
    * Create an input stream that obtains a nodelist from the
    * namenode, and then reads from all the right places.  Creates
    * inner subclass of InputStream that does the right out-of-band
    * work.
    */
-  DFSInputStream open(String src, int buffersize, boolean verifyChecksum
+  DFSInputStream open(String src, int buffersize, boolean verifyChecksum,
+                      FileSystem.Statistics stats
       ) throws IOException {
     checkOpen();
     //    Get block info from namenode
@@ -1026,6 +1037,7 @@ class DFSClient implements FSConstants {
     private Block currentBlock = null;
     private long pos = 0;
     private long blockEnd = -1;
+
     /* XXX Use of CocurrentHashMap is temp fix. Need to fix 
      * parallel accesses to DFSInputStream (through ptreads) properly */
     private ConcurrentHashMap<DatanodeInfo, DatanodeInfo> deadNodes = 
@@ -1039,7 +1051,7 @@ class DFSClient implements FSConstants {
     }
     
     DFSInputStream(String src, int buffersize, boolean verifyChecksum
-        ) throws IOException {
+                   ) throws IOException {
       this.verifyChecksum = verifyChecksum;
       this.buffersize = buffersize;
       this.src = src;
@@ -1309,6 +1321,9 @@ class DFSClient implements FSConstants {
               // got a EOS from reader though we expect more data on it.
               throw new IOException("Unexpected EOS from the reader");
             }
+            if (stats != null && result != -1) {
+              stats.incrementBytesRead(result);
+            }
             return result;
           } catch (ChecksumException ce) {
             throw ce;            
@@ -1456,6 +1471,9 @@ class DFSClient implements FSConstants {
         offset += bytesToRead;
       }
       assert remaining == 0 : "Wrong number of bytes read.";
+      if (stats != null) {
+        stats.incrementBytesRead(realLen);
+      }
       return realLen;
     }
      

src/java/org/apache/hadoop/dfs/DistributedFileSystem.java

@@ -66,7 +66,8 @@ public class DistributedFileSystem extends FileSystem {
     if (host == null || port == -1) {
       throw new IOException("Incomplete HDFS URI, no host/port: "+ uri);
     }
-    this.dfs = new DFSClient(new InetSocketAddress(host, port), conf);
+    this.dfs = new DFSClient(new InetSocketAddress(host, port), conf,
+                             statistics);
     this.uri = URI.create("hdfs://"+host+":"+port);
     this.workingDir = getHomeDirectory();
   }
@@ -128,7 +129,7 @@ public class DistributedFileSystem extends FileSystem {
   public FSDataInputStream open(Path f, int bufferSize) throws IOException {
     try {
       return new DFSClient.DFSDataInputStream(
-          dfs.open(getPathName(f), bufferSize, verifyChecksum));
+          dfs.open(getPathName(f), bufferSize, verifyChecksum, statistics));
     } catch(RemoteException e) {
       if (IOException.class.getName().equals(e.getClassName()) &&
           e.getMessage().startsWith(
@@ -147,8 +148,10 @@ public class DistributedFileSystem extends FileSystem {
     int bufferSize, short replication, long blockSize,
     Progressable progress) throws IOException {
 
-    return new FSDataOutputStream(dfs.create(getPathName(f), permission,
-        overwrite, replication, blockSize, progress, bufferSize));
+    return new FSDataOutputStream
+       (dfs.create(getPathName(f), permission,
+                   overwrite, replication, blockSize, progress, bufferSize),
+        statistics);
   }
 
   public boolean setReplication(Path src, 

src/java/org/apache/hadoop/fs/ChecksumFileSystem.java

@@ -272,7 +272,7 @@ public abstract class ChecksumFileSystem extends FilterFileSystem {
   @Override
   public FSDataInputStream open(Path f, int bufferSize) throws IOException {
     return new FSDataInputStream(
-        new ChecksumFSInputChecker(this, f, bufferSize) );
+        new ChecksumFSInputChecker(this, f, bufferSize));
   }
 
   /**
@@ -358,8 +358,11 @@ public abstract class ChecksumFileSystem extends FilterFileSystem {
     if (parent != null && !mkdirs(parent)) {
       throw new IOException("Mkdirs failed to create " + parent);
     }
-    return new FSDataOutputStream(new ChecksumFSOutputSummer(
-        this, f, overwrite, bufferSize, replication, blockSize, progress));
+    return new FSDataOutputStream
+            (new ChecksumFSOutputSummer
+                (this, f, overwrite, bufferSize, replication, 
+                 blockSize, progress),
+             null);
   }
 
   /**

src/java/org/apache/hadoop/fs/FSDataOutputStream.java

@@ -26,20 +26,29 @@ public class FSDataOutputStream extends DataOutputStream {
   private OutputStream wrappedStream;
 
   private static class PositionCache extends FilterOutputStream {
+    private FileSystem.Statistics statistics;
     long position;
 
-    public PositionCache(OutputStream out) throws IOException {
+    public PositionCache(OutputStream out, 
+                         FileSystem.Statistics stats) throws IOException {
       super(out);
+      statistics = stats;
     }
 
     public void write(int b) throws IOException {
       out.write(b);
       position++;
+      if (statistics != null) {
+        statistics.incrementBytesWritten(1);
+      }
     }
     
     public void write(byte b[], int off, int len) throws IOException {
       out.write(b, off, len);
       position += len;                            // update position
+      if (statistics != null) {
+        statistics.incrementBytesWritten(len);
+      }
     }
       
     public long getPos() throws IOException {
@@ -51,9 +60,14 @@ public class FSDataOutputStream extends DataOutputStream {
     }
   }
 
-  public FSDataOutputStream(OutputStream out)
+  @Deprecated
+  public FSDataOutputStream(OutputStream out) throws IOException {
+    this(out, null);
+  }
+
+  public FSDataOutputStream(OutputStream out, FileSystem.Statistics stats)
     throws IOException {
-    super(new PositionCache(out));
+    super(new PositionCache(out, stats));
     wrappedStream = out;
   }
   

src/java/org/apache/hadoop/fs/FileSystem.java

@@ -20,6 +20,7 @@ package org.apache.hadoop.fs;
 import java.io.*;
 import java.net.*;
 import java.util.*;
+import java.util.concurrent.atomic.AtomicLong;
 import java.util.regex.Pattern;
 
 import org.apache.commons.logging.*;
@@ -57,7 +58,16 @@ public abstract class FileSystem extends Configured implements Closeable {
 
   /** FileSystem cache */
   private static final Cache CACHE = new Cache();
+  /** Recording statistics per a FileSystem class */
+  private static final Map<Class<? extends FileSystem>, Statistics> 
+    statisticsTable =
+      new IdentityHashMap<Class<? extends FileSystem>, Statistics>();
 
+  /**
+   * The statistics for this file system.
+   */
+  protected final Statistics statistics;
+  
   /**
    * Parse the cmd-line args, starting at i.  Remove consumed args
    * from array.  We expect param in the form:
@@ -268,6 +278,7 @@ public abstract class FileSystem extends Configured implements Closeable {
 
   protected FileSystem() {
     super(null);
+    statistics = getStatistics(this.getClass());
   }
 
   /** Check that a Path belongs to this FileSystem. */
@@ -1372,4 +1383,70 @@ public abstract class FileSystem extends Configured implements Closeable {
       }
     }
   }
+  
+  public static final class Statistics {
+    private AtomicLong bytesRead = new AtomicLong();
+    private AtomicLong bytesWritten = new AtomicLong();
+    
+    /**
+     * Increment the bytes read in the statistics
+     * @param newBytes the additional bytes read
+     */
+    public void incrementBytesRead(long newBytes) {
+      bytesRead.getAndAdd(newBytes);
+    }
+    
+    /**
+     * Increment the bytes written in the statistics
+     * @param newBytes the additional bytes written
+     */
+    public void incrementBytesWritten(long newBytes) {
+      bytesWritten.getAndAdd(newBytes);
+    }
+    
+    /**
+     * Get the total number of bytes read
+     * @return the number of bytes
+     */
+    public long getBytesRead() {
+      return bytesRead.get();
+    }
+    
+    /**
+     * Get the total number of bytes written
+     * @return the number of bytes
+     */
+    public long getBytesWritten() {
+      return bytesWritten.get();
+    }
+    
+    public String toString() {
+      return bytesRead + " bytes read and " + bytesWritten + 
+             " bytes written";
+    }
+  }
+  
+  /**
+   * Get the statistics for a particular file system
+   * @param cls the class to lookup
+   * @return a statistics object
+   */
+  public static synchronized 
+  Statistics getStatistics(Class<? extends FileSystem> cls) {
+    Statistics result = statisticsTable.get(cls);
+    if (result == null) {
+      result = new Statistics();
+      statisticsTable.put(cls, result);
+    }
+    return result;
+  }
+  
+  public static synchronized
+  void printStatistics() throws IOException {
+    for (Map.Entry<Class<? extends FileSystem>, Statistics> pair: 
+            statisticsTable.entrySet()) {
+      System.out.println("  FileSystem " + pair.getKey().getName() + 
+                         ": " + pair.getValue());
+    }
+  }
 }

src/java/org/apache/hadoop/fs/InMemoryFileSystem.java

@@ -223,7 +223,8 @@ public class InMemoryFileSystem extends ChecksumFileSystem {
       // map) until close is called on the outputstream that this method is
       // going to return
       // Create an output stream out of data byte array
-      return new FSDataOutputStream(new InMemoryOutputStream(f, fAttr));
+      return new FSDataOutputStream(new InMemoryOutputStream(f, fAttr), 
+                                    statistics);
     }
 
     public void close() throws IOException {

src/java/org/apache/hadoop/fs/RawLocalFileSystem.java

@@ -65,6 +65,36 @@ public class RawLocalFileSystem extends FileSystem {
     setConf(conf);
   }
   
+  class TrackingFileInputStream extends FileInputStream {
+    public TrackingFileInputStream(File f) throws IOException {
+      super(f);
+    }
+    
+    public int read() throws IOException {
+      int result = super.read();
+      if (result != -1) {
+        statistics.incrementBytesRead(1);
+      }
+      return result;
+    }
+    
+    public int read(byte[] data) throws IOException {
+      int result = super.read(data);
+      if (result != -1) {
+        statistics.incrementBytesRead(result);
+      }
+      return result;
+    }
+    
+    public int read(byte[] data, int offset, int length) throws IOException {
+      int result = super.read(data, offset, length);
+      if (result != -1) {
+        statistics.incrementBytesRead(result);
+      }
+      return result;
+    }
+  }
+
   /*******************************************************
    * For open()'s FSInputStream
    *******************************************************/
@@ -73,7 +103,7 @@ public class RawLocalFileSystem extends FileSystem {
     private long position;
 
     public LocalFSFileInputStream(Path f) throws IOException {
-      this.fis = new FileInputStream(pathToFile(f));
+      this.fis = new TrackingFileInputStream(pathToFile(f));
     }
     
     public void seek(long pos) throws IOException {
@@ -190,7 +220,8 @@ public class RawLocalFileSystem extends FileSystem {
       throw new IOException("Mkdirs failed to create " + parent.toString());
     }
     return new FSDataOutputStream(
-        new BufferedOutputStream(new LocalFSFileOutputStream(f), bufferSize));
+        new BufferedOutputStream(new LocalFSFileOutputStream(f), bufferSize),
+        statistics);
   }
 
   /** {@inheritDoc} */

src/java/org/apache/hadoop/fs/kfs/KFSImpl.java

@@ -20,23 +20,27 @@
 package org.apache.hadoop.fs.kfs;
 
 import java.io.*;
-import java.net.*;
 
-import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.fs.FSDataInputStream;
 import org.apache.hadoop.fs.FSDataOutputStream;
 import org.apache.hadoop.fs.FileSystem;
-import org.apache.hadoop.fs.FileUtil;
-import org.apache.hadoop.fs.Path;
-import org.apache.hadoop.util.Progressable;
 
 import org.kosmix.kosmosfs.access.KfsAccess;
 
 class KFSImpl implements IFSImpl {
     private KfsAccess kfsAccess = null;
+    private FileSystem.Statistics statistics;
 
-    public KFSImpl(String metaServerHost, int metaServerPort) throws IOException {
+    @Deprecated
+    public KFSImpl(String metaServerHost, int metaServerPort
+                   ) throws IOException {
+      this(metaServerHost, metaServerPort, null);
+    }
+
+    public KFSImpl(String metaServerHost, int metaServerPort, 
+                   FileSystem.Statistics stats) throws IOException {
         kfsAccess = new KfsAccess(metaServerHost, metaServerPort);
+        statistics = stats;
     }
 
     public boolean exists(String path) throws IOException {
@@ -95,10 +99,12 @@ class KFSImpl implements IFSImpl {
     }
 
     public FSDataOutputStream create(String path, short replication, int bufferSize) throws IOException {
-        return new FSDataOutputStream(new KFSOutputStream(kfsAccess, path, replication));
+        return new FSDataOutputStream(new KFSOutputStream(kfsAccess, path, replication), 
+                                      statistics);
     }
 
     public FSDataInputStream open(String path, int bufferSize) throws IOException {
-        return new FSDataInputStream(new KFSInputStream(kfsAccess, path));
+        return new FSDataInputStream(new KFSInputStream(kfsAccess, path, 
+                                                        statistics));
     }
-};
+}

src/java/org/apache/hadoop/fs/kfs/KFSInputStream.java

@@ -21,28 +21,28 @@
 package org.apache.hadoop.fs.kfs;
 
 import java.io.*;
-import java.net.*;
-import java.util.*;
 import java.nio.ByteBuffer;
 
-import org.apache.hadoop.conf.Configuration;
-import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.fs.FileSystem;
 import org.apache.hadoop.fs.FSInputStream;
-import org.apache.hadoop.util.Progressable;
 
 import org.kosmix.kosmosfs.access.KfsAccess;
 import org.kosmix.kosmosfs.access.KfsInputChannel;
 
 class KFSInputStream extends FSInputStream {
 
-    private String path;
     private KfsInputChannel kfsChannel;
-
+    private FileSystem.Statistics statistics;
     private long fsize;
 
+    @Deprecated
     public KFSInputStream(KfsAccess kfsAccess, String path) {
-        this.path = path;
+      this(kfsAccess, path, null);
+    }
 
+    public KFSInputStream(KfsAccess kfsAccess, String path,
+                            FileSystem.Statistics stats) {
+        this.statistics = stats;
         this.kfsChannel = kfsAccess.kfs_open(path);
         if (this.kfsChannel != null)
             this.fsize = kfsAccess.kfs_filesize(path);
@@ -81,8 +81,12 @@ class KFSInputStream extends FSInputStream {
         }
         byte b[] = new byte[4];
         int res = read(b, 0, 4);
-        if (res == 4)
-            return (b[0] + (b[1] << 8) + (b[2] << 16) + (b[3] << 24));
+        if (res == 4) {
+          if (statistics != null) {
+            statistics.incrementBytesRead(1);
+          }
+          return (b[0] + (b[1] << 8) + (b[2] << 16) + (b[3] << 24));          
+        }
         return -1;
     }
 
@@ -96,6 +100,9 @@ class KFSInputStream extends FSInputStream {
 	// Use -1 to signify EOF
 	if (res == 0)
 	    return -1;
+	if (statistics != null) {
+	  statistics.incrementBytesRead(res);
+	}
 	return res;
     }
 

src/java/org/apache/hadoop/fs/kfs/KosmosFileSystem.java

@@ -63,7 +63,8 @@ public class KosmosFileSystem extends FileSystem {
         try {
             if (kfsImpl == null) {
                 kfsImpl = new KFSImpl(conf.get("fs.kfs.metaServerHost", ""),
-                                      conf.getInt("fs.kfs.metaServerPort", -1));
+                                      conf.getInt("fs.kfs.metaServerPort", -1),
+                                      statistics);
             }
             this.localFs = FileSystem.getLocal(conf);
             this.uri = URI.create(uri.getScheme() + "://" + uri.getAuthority());

src/java/org/apache/hadoop/fs/s3/S3FileSystem.java

@@ -203,15 +203,17 @@ public class S3FileSystem extends FileSystem {
         }
       }      
     }
-    return new FSDataOutputStream(
-        new S3OutputStream(getConf(), store, makeAbsolute(file),
-                           blockSize, progress, bufferSize));
+    return new FSDataOutputStream
+        (new S3OutputStream(getConf(), store, makeAbsolute(file),
+                            blockSize, progress, bufferSize),
+         statistics);
   }
 
   @Override
   public FSDataInputStream open(Path path, int bufferSize) throws IOException {
     INode inode = checkFile(path);
-    return new FSDataInputStream(new S3InputStream(getConf(), store, inode));
+    return new FSDataInputStream(new S3InputStream(getConf(), store, inode,
+                                                   statistics));
   }
 
   @Override

src/java/org/apache/hadoop/fs/s3/S3InputStream.java

@@ -25,6 +25,7 @@ import java.io.IOException;
 
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.fs.FSInputStream;
+import org.apache.hadoop.fs.FileSystem;
 
 class S3InputStream extends FSInputStream {
 
@@ -43,11 +44,20 @@ class S3InputStream extends FSInputStream {
   private DataInputStream blockStream;
 
   private long blockEnd = -1;
+  
+  private FileSystem.Statistics stats;
 
+  @Deprecated
   public S3InputStream(Configuration conf, FileSystemStore store,
                        INode inode) {
+    this(conf, store, inode, null);
+  }
+
+  public S3InputStream(Configuration conf, FileSystemStore store,
+                       INode inode, FileSystem.Statistics stats) {
     
     this.store = store;
+    this.stats = stats;
     this.blocks = inode.getBlocks();
     for (Block block : blocks) {
       this.fileLength += block.getLength();
@@ -93,6 +103,9 @@ class S3InputStream extends FSInputStream {
         pos++;
       }
     }
+    if (stats != null & result >= 0) {
+      stats.incrementBytesRead(1);
+    }
     return result;
   }
 
@@ -110,6 +123,9 @@ class S3InputStream extends FSInputStream {
       if (result >= 0) {
         pos += result;
       }
+      if (stats != null && result > 0) {
+        stats.incrementBytesRead(result);
+      }
       return result;
     }
     return -1;

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

@@ -22,6 +22,8 @@ import java.io.DataInput;
 import java.io.DataOutput;
 import java.io.IOException;
 import java.net.URI;
+import java.util.ArrayList;
+import java.util.List;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.text.NumberFormat;
 
@@ -29,11 +31,15 @@ import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 import org.apache.hadoop.conf.Configurable;
 import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.dfs.DistributedFileSystem;
 import org.apache.hadoop.fs.FileStatus;
 import org.apache.hadoop.fs.FileSystem;
 import org.apache.hadoop.fs.FileUtil;
 import org.apache.hadoop.fs.Path;
 import org.apache.hadoop.fs.LocalDirAllocator;
+import org.apache.hadoop.fs.RawLocalFileSystem;
+import org.apache.hadoop.fs.kfs.KosmosFileSystem;
+import org.apache.hadoop.fs.s3.S3FileSystem;
 import org.apache.hadoop.io.Text;
 import org.apache.hadoop.io.UTF8;
 import org.apache.hadoop.io.Writable;
@@ -59,6 +65,16 @@ abstract class Task implements Writable, Configurable {
     REDUCE_INPUT_RECORDS,
     REDUCE_OUTPUT_RECORDS
   }
+  
+  /**
+   * Counters to measure the usage of the different file systems.
+   */
+  protected static enum FileSystemCounter {
+    LOCAL_READ, LOCAL_WRITE, 
+    HDFS_READ, HDFS_WRITE, 
+    S3_READ, S3_WRITE,
+    KFS_READ, KFSWRITE
+  }
 
   ///////////////////////////////////////////////////////////
   // Helper methods to construct task-output paths
@@ -293,6 +309,7 @@ abstract class Task implements Writable, Configurable {
               
               if (sendProgress) {
                 // we need to send progress update
+                updateCounters();
                 taskStatus.statusUpdate(taskProgress.get(), taskProgress.toString(), 
                         counters);
                 taskFound = umbilical.statusUpdate(taskId, taskStatus);
@@ -362,9 +379,81 @@ abstract class Task implements Writable, Configurable {
     setProgressFlag();
   }
 
+  /**
+   * An updater that tracks the last number reported for a given file
+   * system and only creates the counters when they are needed.
+   */
+  class FileSystemStatisticUpdater {
+    private long prevReadBytes = 0;
+    private long prevWriteBytes = 0;
+    private FileSystem.Statistics stats;
+    private Counters.Counter readCounter = null;
+    private Counters.Counter writeCounter = null;
+    private FileSystemCounter read;
+    private FileSystemCounter write;
+
+    FileSystemStatisticUpdater(FileSystemCounter read,
+                               FileSystemCounter write,
+                               Class<? extends FileSystem> cls) {
+      stats = FileSystem.getStatistics(cls);
+      this.read = read;
+      this.write = write;
+    }
+
+    void updateCounters() {
+      long newReadBytes = stats.getBytesRead();
+      long newWriteBytes = stats.getBytesWritten();
+      if (prevReadBytes != newReadBytes) {
+        if (readCounter == null) {
+          readCounter = counters.findCounter(read);
+        }
+        readCounter.increment(newReadBytes - prevReadBytes);
+        prevReadBytes = newReadBytes;
+      }
+      if (prevWriteBytes != newWriteBytes) {
+        if (writeCounter == null) {
+          writeCounter = counters.findCounter(write);
+        }
+        writeCounter.increment(newWriteBytes - prevWriteBytes);
+        prevWriteBytes = newWriteBytes;
+      }
+    }
+  }
+  
+  /**
+   * A list of all of the file systems that we want to report on.
+   */
+  private List<FileSystemStatisticUpdater> statisticUpdaters =
+     new ArrayList<FileSystemStatisticUpdater>();
+  {
+    statisticUpdaters.add
+      (new FileSystemStatisticUpdater(FileSystemCounter.LOCAL_READ,
+                                      FileSystemCounter.LOCAL_WRITE,
+                                      RawLocalFileSystem.class));
+    statisticUpdaters.add
+      (new FileSystemStatisticUpdater(FileSystemCounter.HDFS_READ,
+                                      FileSystemCounter.HDFS_WRITE,
+                                      DistributedFileSystem.class));
+    statisticUpdaters.add
+    (new FileSystemStatisticUpdater(FileSystemCounter.KFS_READ,
+                                    FileSystemCounter.KFSWRITE,
+                                    KosmosFileSystem.class));
+    statisticUpdaters.add
+    (new FileSystemStatisticUpdater(FileSystemCounter.S3_READ,
+                                    FileSystemCounter.S3_WRITE,
+                                    S3FileSystem.class));
+  }
+
+  private synchronized void updateCounters() {
+    for(FileSystemStatisticUpdater updater: statisticUpdaters) {
+      updater.updateCounters();
+    }
+  }
+
   public void done(TaskUmbilicalProtocol umbilical) throws IOException {
     int retries = 10;
     boolean needProgress = true;
+    updateCounters();
     taskDone.set(true);
     while (true) {
       try {

src/java/org/apache/hadoop/mapred/Task_FileSystemCounter.properties

@@ -0,0 +1,12 @@
+# ResourceBundle properties file for Map-Reduce counters
+
+CounterGroupName=              File Systems
+
+LOCAL_READ.name=               Local bytes read
+LOCAL_WRITE.name=              Local bytes written
+HDFS_READ.name=                HDFS bytes read
+HDFS_WRITE.name=               HDFS bytes written
+S3_READ.name=                  S3 bytes read
+S3_WRITE.name=                 S3 bytes written
+KFS_READ.name=                 KFS bytes read
+KFS_WRITE.name=                KFS bytes written

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

@@ -178,10 +178,8 @@ public class TestMiniMRWithDFS extends TestCase {
     TestResult result;
     final Path inDir = new Path("./wc/input");
     final Path outDir = new Path("./wc/output");
-    result = launchWordCount(jobConf, inDir, outDir,
-                             "The quick brown fox\nhas many silly\n" + 
-                             "red fox sox\n",
-                             3, 1);
+    String input = "The quick brown fox\nhas many silly\nred fox sox\n";
+    result = launchWordCount(jobConf, inDir, outDir, input, 3, 1);
     assertEquals("The\t1\nbrown\t1\nfox\t2\nhas\t1\nmany\t1\n" +
                  "quick\t1\nred\t1\nsilly\t1\nsox\t1\n", result.output);
     String jobid = result.job.getJobID();
@@ -189,8 +187,17 @@ public class TestMiniMRWithDFS extends TestCase {
     checkTaskDirectories(mr, new String[]{jobid}, new String[]{taskid});
     // test with maps=0
     jobConf = mr.createJobConf();
-    result = launchWordCount(jobConf, inDir, outDir, "owen is oom", 0, 1);
+    input = "owen is oom";
+    result = launchWordCount(jobConf, inDir, outDir, input, 0, 1);
     assertEquals("is\t1\noom\t1\nowen\t1\n", result.output);
+    Counters counters = result.job.getCounters();
+    long hdfsRead = 
+      counters.findCounter(Task.FileSystemCounter.HDFS_READ).getCounter();
+    long hdfsWrite = 
+      counters.findCounter(Task.FileSystemCounter.HDFS_WRITE).getCounter();
+    assertEquals(result.output.length(), hdfsWrite);
+    assertEquals(input.length(), hdfsRead);
+
     // Run a job with input and output going to localfs even though the 
     // default fs is hdfs.
     {
@@ -207,6 +214,7 @@ public class TestMiniMRWithDFS extends TestCase {
       assertEquals("all\t1\nbase\t1\nbelong\t1\nto\t1\nus\t1\nyour\t1\n", 
                    result.output);
       assertTrue("outputs on localfs", localfs.exists(localOut));
+
     }
   }