diff --git a/CHANGES.txt b/CHANGES.txt
index 99cef35..b410256 100644
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -25,6 +25,9 @@
 
   IMPROVEMENTS
 
+    MAPREDUCE-2104. [Rumen] Add Cpu, Memory and Heap usages to 
+    TraceBuilder's output. (amarrk)
+
     MAPREDUCE-2554. [Gridmix]  Add distributed cache emulation system tests 
     to Gridmix. (Vinay Kumar Thota via amarrk)
  
diff --git a/src/test/mapred/org/apache/hadoop/tools/rumen/TestRumenJobTraces.java b/src/test/mapred/org/apache/hadoop/tools/rumen/TestRumenJobTraces.java
index 8a1be6b..c57875a 100644
--- a/src/test/mapred/org/apache/hadoop/tools/rumen/TestRumenJobTraces.java
+++ b/src/test/mapred/org/apache/hadoop/tools/rumen/TestRumenJobTraces.java
@@ -33,11 +33,13 @@
 import org.apache.hadoop.mapred.JobClient;
 import org.apache.hadoop.mapred.JobConf;
 import org.apache.hadoop.mapred.MiniMRCluster;
+import org.apache.hadoop.mapreduce.Counters;
 import org.apache.hadoop.mapreduce.Job;
 import org.apache.hadoop.mapreduce.JobID;
 import org.apache.hadoop.mapreduce.MRJobConfig;
 import org.apache.hadoop.mapreduce.MapReduceTestUtil;
 import org.apache.hadoop.mapreduce.TaskAttemptID;
+import org.apache.hadoop.mapreduce.TaskCounter;
 import org.apache.hadoop.mapreduce.TaskID;
 import org.apache.hadoop.mapreduce.TaskType;
 import org.apache.hadoop.mapreduce.TestNoJobSetupCleanup.MyOutputFormat;
@@ -230,7 +232,7 @@
       parser = new Hadoop20JHParser(ris);
       ArrayList<String> seenEvents = new ArrayList<String>(150);
 
-      getHistoryEvents(parser, seenEvents); // get events into seenEvents
+      getHistoryEvents(parser, seenEvents, null); // get events into seenEvents
 
       // Validate the events seen by history parser from
       // history file v20-single-input-log.gz
@@ -512,7 +514,11 @@
       // Test if the JobHistoryParserFactory can detect the parser correctly
       parser = JobHistoryParserFactory.getParser(ris);
 
-      getHistoryEvents(parser, seenEvents); // get events into seenEvents
+      // create a job builder
+      JobBuilder builder = new JobBuilder(id.toString());
+
+      // get events into seenEvents and also process them using builder
+      getHistoryEvents(parser, seenEvents, builder); 
 
       // Check against the gold standard
       System.out.println("testCurrentJHParser validating using gold std ");
@@ -523,6 +529,26 @@
       };
 
       validateSeenHistoryEvents(seenEvents, goldLinesExpected);
+      
+      // validate resource usage metrics
+      //  get the job counters
+      Counters counters = job.getTaskReports(TaskType.MAP)[0].getTaskCounters();
+      
+      //  get the logged job
+      LoggedJob loggedJob = builder.build();
+      //  get the logged attempts
+      LoggedTaskAttempt attempt = 
+        loggedJob.getMapTasks().get(0).getAttempts().get(0);
+      //  get the resource usage metrics
+      ResourceUsageMetrics metrics = attempt.getResourceUsageMetrics();
+      
+      //  check with the actual values
+      testResourceUsageMetricViaDeepCompare(metrics, 
+          counters.findCounter(TaskCounter.CPU_MILLISECONDS).getValue(), 
+          counters.findCounter(TaskCounter.VIRTUAL_MEMORY_BYTES).getValue(),
+          counters.findCounter(TaskCounter.PHYSICAL_MEMORY_BYTES).getValue(),
+          counters.findCounter(TaskCounter.COMMITTED_HEAP_BYTES).getValue(),
+          true);
     } finally {
       // stop the MR cluster
       mrCluster.shutdown();
@@ -687,6 +713,141 @@
     }
 
 
+  /**
+   * Test {@link ResourceUsageMetrics}.
+   */
+  @Test
+  public void testResourceUsageMetrics() throws Exception {
+    final long cpuUsage = 100;
+    final long pMemUsage = 200;
+    final long vMemUsage = 300;
+    final long heapUsage = 400;
+    
+    // test ResourceUsageMetrics's setters
+    ResourceUsageMetrics metrics = new ResourceUsageMetrics();
+    metrics.setCumulativeCpuUsage(cpuUsage);
+    metrics.setPhysicalMemoryUsage(pMemUsage);
+    metrics.setVirtualMemoryUsage(vMemUsage);
+    metrics.setHeapUsage(heapUsage);
+    // test cpu usage value
+    assertEquals("Cpu usage values mismatch via set", cpuUsage, 
+                 metrics.getCumulativeCpuUsage());
+    // test pMem usage value
+    assertEquals("Physical memory usage values mismatch via set", pMemUsage, 
+                 metrics.getPhysicalMemoryUsage());
+    // test vMem usage value
+    assertEquals("Virtual memory usage values mismatch via set", vMemUsage, 
+                 metrics.getVirtualMemoryUsage());
+    // test heap usage value
+    assertEquals("Heap usage values mismatch via set", heapUsage, 
+                 metrics.getHeapUsage());
+    
+    // test deepCompare() (pass case)
+    testResourceUsageMetricViaDeepCompare(metrics, cpuUsage, vMemUsage, 
+                                          pMemUsage, heapUsage, true);
+    
+    // test deepCompare (fail case)
+    // test cpu usage mismatch
+    testResourceUsageMetricViaDeepCompare(metrics, 0, vMemUsage, pMemUsage, 
+                                          heapUsage, false);
+    // test pMem usage mismatch
+    testResourceUsageMetricViaDeepCompare(metrics, cpuUsage, vMemUsage, 0, 
+                                          heapUsage, false);
+    // test vMem usage mismatch
+    testResourceUsageMetricViaDeepCompare(metrics, cpuUsage, 0, pMemUsage, 
+                                          heapUsage, false);
+    // test heap usage mismatch
+    testResourceUsageMetricViaDeepCompare(metrics, cpuUsage, vMemUsage, 
+                                          pMemUsage, 0, false);
+    
+    // define a metric with a fixed value of size()
+    ResourceUsageMetrics metrics2 = new ResourceUsageMetrics() {
+      @Override
+      public int size() {
+        return -1;
+      }
+    };
+    metrics2.setCumulativeCpuUsage(cpuUsage);
+    metrics2.setPhysicalMemoryUsage(pMemUsage);
+    metrics2.setVirtualMemoryUsage(vMemUsage);
+    metrics2.setHeapUsage(heapUsage);
+    
+    // test with size mismatch
+    testResourceUsageMetricViaDeepCompare(metrics2, cpuUsage, vMemUsage, 
+                                          pMemUsage, heapUsage, false);
+  }
+  
+  // test ResourceUsageMetric's deepCompare() method
+  private static void testResourceUsageMetricViaDeepCompare(
+                        ResourceUsageMetrics metrics, long cpuUsage, 
+                        long vMemUsage, long pMemUsage, long heapUsage,
+                        boolean shouldPass) {
+    ResourceUsageMetrics testMetrics = new ResourceUsageMetrics();
+    testMetrics.setCumulativeCpuUsage(cpuUsage);
+    testMetrics.setPhysicalMemoryUsage(pMemUsage);
+    testMetrics.setVirtualMemoryUsage(vMemUsage);
+    testMetrics.setHeapUsage(heapUsage);
+    
+    Boolean passed = null;
+    try {
+      metrics.deepCompare(testMetrics, new TreePath(null, "<root>"));
+      passed = true;
+    } catch (DeepInequalityException die) {
+      passed = false;
+    }
+    
+    assertEquals("ResourceUsageMetrics deepCompare() failed!", 
+                 shouldPass, passed);
+  }
+  
+  /**
+   * Testing {@link ResourceUsageMetrics} using {@link HadoopLogsAnalyzer}.
+   */
+  @Test
+  @SuppressWarnings("deprecation")
+  public void testResourceUsageMetricsWithHadoopLogsAnalyzer() 
+  throws IOException {
+    Configuration conf = new Configuration();
+    // get the input trace file
+    Path rootInputDir =
+      new Path(System.getProperty("test.tools.input.dir", ""));
+    Path rootInputSubFolder = new Path(rootInputDir, "rumen/small-trace-test");
+    Path traceFile = new Path(rootInputSubFolder, "v20-resource-usage-log.gz");
+    
+    FileSystem lfs = FileSystem.getLocal(conf);
+    
+    // define the root test directory
+    Path rootTempDir =
+        new Path(System.getProperty("test.build.data", "/tmp"));
+
+    // define output directory
+    Path outputDir = 
+      new Path(rootTempDir, "testResourceUsageMetricsWithHadoopLogsAnalyzer");
+    lfs.delete(outputDir, true);
+    lfs.deleteOnExit(outputDir);
+    
+    // run HadoopLogsAnalyzer
+    HadoopLogsAnalyzer analyzer = new HadoopLogsAnalyzer();
+    analyzer.setConf(conf);
+    Path traceOutput = new Path(outputDir, "trace.json");
+    analyzer.run(new String[] {"-write-job-trace", traceOutput.toString(), 
+                               "-v1", traceFile.toString()});
+    
+    // test HadoopLogsAnalyzer's output w.r.t ResourceUsageMetrics
+    //  get the logged job
+    JsonObjectMapperParser<LoggedJob> traceParser =
+      new JsonObjectMapperParser<LoggedJob>(traceOutput, LoggedJob.class, 
+                                            conf);
+    
+    //  get the logged job from the output trace file
+    LoggedJob job = traceParser.getNext();
+    LoggedTaskAttempt attempt = job.getMapTasks().get(0).getAttempts().get(0);
+    ResourceUsageMetrics metrics = attempt.getResourceUsageMetrics();
+    
+    //  test via deepCompare()
+    testResourceUsageMetricViaDeepCompare(metrics, 200, 100, 75, 50, true);
+  }
+  
   @Test
   public void testTopologyBuilder() throws Exception {
     final TopologyBuilder subject = new TopologyBuilder();
@@ -795,12 +956,15 @@
    * @throws IOException
    */
   private void getHistoryEvents(JobHistoryParser parser,
-      ArrayList<String> events) throws IOException {
+      ArrayList<String> events, JobBuilder builder) throws IOException {
     HistoryEvent e;
     while ((e = parser.nextEvent()) != null) {
       String eventString = e.getClass().getSimpleName();
       System.out.println(eventString);
       events.add(eventString);
+      if (builder != null) {
+        builder.process(e);
+      }
     }
   }
 
diff --git a/src/test/tools/data/rumen/small-trace-test/v20-resource-usage-log.gz b/src/test/tools/data/rumen/small-trace-test/v20-resource-usage-log.gz
new file mode 100644
index 0000000..6d0dbeb
--- /dev/null
+++ b/src/test/tools/data/rumen/small-trace-test/v20-resource-usage-log.gz
Binary files differ
diff --git a/src/tools/org/apache/hadoop/tools/rumen/HadoopLogsAnalyzer.java b/src/tools/org/apache/hadoop/tools/rumen/HadoopLogsAnalyzer.java
index c393147..b8044ec 100644
--- a/src/tools/org/apache/hadoop/tools/rumen/HadoopLogsAnalyzer.java
+++ b/src/tools/org/apache/hadoop/tools/rumen/HadoopLogsAnalyzer.java
@@ -1208,6 +1208,38 @@
         attempt.spilledRecords = val;
       }
     }, counterString, "SPILLED_RECORDS");
+    
+    // incorporate CPU usage
+    incorporateCounter(new SetField(attempt2) {
+      @Override
+      void set(long val) {
+        attempt.getResourceUsageMetrics().setCumulativeCpuUsage(val);
+      }
+    }, counterString, "CPU_MILLISECONDS");
+    
+    // incorporate virtual memory usage
+    incorporateCounter(new SetField(attempt2) {
+      @Override
+      void set(long val) {
+        attempt.getResourceUsageMetrics().setVirtualMemoryUsage(val);
+      }
+    }, counterString, "VIRTUAL_MEMORY_BYTES");
+    
+    // incorporate physical memory usage
+    incorporateCounter(new SetField(attempt2) {
+      @Override
+      void set(long val) {
+        attempt.getResourceUsageMetrics().setPhysicalMemoryUsage(val);
+      }
+    }, counterString, "PHYSICAL_MEMORY_BYTES");
+    
+    // incorporate heap usage
+    incorporateCounter(new SetField(attempt2) {
+      @Override
+      void set(long val) {
+        attempt.getResourceUsageMetrics().setHeapUsage(val);
+      }
+    }, counterString, "COMMITTED_HEAP_BYTES");
   }
 
   private ParsedHost getAndRecordParsedHost(String hostName) {
diff --git a/src/tools/org/apache/hadoop/tools/rumen/LoggedTaskAttempt.java b/src/tools/org/apache/hadoop/tools/rumen/LoggedTaskAttempt.java
index 18b26ca..0df30e6 100644
--- a/src/tools/org/apache/hadoop/tools/rumen/LoggedTaskAttempt.java
+++ b/src/tools/org/apache/hadoop/tools/rumen/LoggedTaskAttempt.java
@@ -28,7 +28,6 @@
 //                the Jackson implementation of JSON doesn't handle a 
 //                superclass-valued field.
 
-import org.apache.hadoop.mapreduce.jobhistory.Events;
 import org.apache.hadoop.mapreduce.jobhistory.JhCounter;
 import org.apache.hadoop.mapreduce.jobhistory.JhCounterGroup;
 import org.apache.hadoop.mapreduce.jobhistory.JhCounters;
@@ -69,6 +68,9 @@
 
   LoggedLocation location;
 
+  // Initialize to default object for backward compatibility
+  ResourceUsageMetrics metrics = new ResourceUsageMetrics();
+  
   LoggedTaskAttempt() {
     super();
   }
@@ -354,8 +356,50 @@
         attempt.spilledRecords = val;
       }
     }, counters, "SPILLED_RECORDS");
+    
+    // incorporate CPU usage
+    incorporateCounter(new SetField(this) {
+      @Override
+      void set(long val) {
+        metrics.setCumulativeCpuUsage(val);
+      }
+    }, counters, "CPU_MILLISECONDS");
+    
+    // incorporate virtual memory usage
+    incorporateCounter(new SetField(this) {
+      @Override
+      void set(long val) {
+        metrics.setVirtualMemoryUsage(val);
+      }
+    }, counters, "VIRTUAL_MEMORY_BYTES");
+    
+    // incorporate physical memory usage
+    incorporateCounter(new SetField(this) {
+      @Override
+      void set(long val) {
+        metrics.setPhysicalMemoryUsage(val);
+      }
+    }, counters, "PHYSICAL_MEMORY_BYTES");
+    
+    // incorporate heap usage
+    incorporateCounter(new SetField(this) {
+      @Override
+      void set(long val) {
+        metrics.setHeapUsage(val);
+      }
+    }, counters, "COMMITTED_HEAP_BYTES");
   }
 
+  // Get the resource usage metrics
+  public ResourceUsageMetrics getResourceUsageMetrics() {
+    return metrics;
+  }
+  
+  // Set the resource usage metrics
+  void setResourceUsageMetrics(ResourceUsageMetrics metrics) {
+    this.metrics = metrics;
+  }
+  
   private static String canonicalizeCounterName(String nonCanonicalName) {
     String result = nonCanonicalName.toLowerCase();
 
diff --git a/src/tools/org/apache/hadoop/tools/rumen/ResourceUsageMetrics.java b/src/tools/org/apache/hadoop/tools/rumen/ResourceUsageMetrics.java
new file mode 100644
index 0000000..a0944c8
--- /dev/null
+++ b/src/tools/org/apache/hadoop/tools/rumen/ResourceUsageMetrics.java
@@ -0,0 +1,160 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.hadoop.tools.rumen;
+
+import java.io.DataInput;
+import java.io.DataOutput;
+import java.io.IOException;
+
+import org.apache.hadoop.io.Writable;
+import org.apache.hadoop.io.WritableUtils;
+
+/**
+ * Captures the resource usage metrics.
+ */
+public class ResourceUsageMetrics implements Writable, DeepCompare  {
+  private long cumulativeCpuUsage;
+  private long virtualMemoryUsage;
+  private long physicalMemoryUsage;
+  private long heapUsage;
+  
+  public ResourceUsageMetrics() {
+  }
+  
+  /**
+   * Get the cumulative CPU usage.
+   */
+  public long getCumulativeCpuUsage() {
+    return cumulativeCpuUsage;
+  }
+  
+  /**
+   * Set the cumulative CPU usage.
+   */
+  public void setCumulativeCpuUsage(long usage) {
+    cumulativeCpuUsage = usage;
+  }
+  
+  /**
+   * Get the virtual memory usage.
+   */
+  public long getVirtualMemoryUsage() {
+    return virtualMemoryUsage;
+  }
+  
+  /**
+   * Set the virtual memory usage.
+   */
+  public void setVirtualMemoryUsage(long usage) {
+    virtualMemoryUsage = usage;
+  }
+  
+  /**
+   * Get the physical memory usage.
+   */
+  public long getPhysicalMemoryUsage() {
+    return physicalMemoryUsage;
+  }
+  
+  /**
+   * Set the physical memory usage.
+   */
+  public void setPhysicalMemoryUsage(long usage) {
+    physicalMemoryUsage = usage;
+  }
+  
+  /**
+   * Get the total heap usage.
+   */
+  public long getHeapUsage() {
+    return heapUsage;
+  }
+  
+  /**
+   * Set the total heap usage.
+   */
+  public void setHeapUsage(long usage) {
+    heapUsage = usage;
+  }
+  
+  /**
+   * Returns the size of the serialized data
+   */
+  public int size() {
+    int size = 0;
+    size += WritableUtils.getVIntSize(cumulativeCpuUsage);   // long #1
+    size += WritableUtils.getVIntSize(virtualMemoryUsage);   // long #2
+    size += WritableUtils.getVIntSize(physicalMemoryUsage);  // long #3
+    size += WritableUtils.getVIntSize(heapUsage);            // long #4
+    return size;
+  }
+  
+  @Override
+  public void readFields(DataInput in) throws IOException {
+    cumulativeCpuUsage = WritableUtils.readVLong(in);  // long #1
+    virtualMemoryUsage = WritableUtils.readVLong(in);  // long #2
+    physicalMemoryUsage = WritableUtils.readVLong(in); // long #3
+    heapUsage = WritableUtils.readVLong(in);           // long #4
+  }
+  
+  @Override
+  public void write(DataOutput out) throws IOException {
+    //TODO Write resources version no too
+    WritableUtils.writeVLong(out, cumulativeCpuUsage);  // long #1
+    WritableUtils.writeVLong(out, virtualMemoryUsage);  // long #2
+    WritableUtils.writeVLong(out, physicalMemoryUsage); // long #3
+    WritableUtils.writeVLong(out, heapUsage);           // long #4
+  }
+
+  private static void compareMetric(long m1, long m2, TreePath loc) 
+  throws DeepInequalityException {
+    if (m1 != m2) {
+      throw new DeepInequalityException("Value miscompared:" + loc.toString(), 
+                                        loc);
+    }
+  }
+  
+  private static void compareSize(ResourceUsageMetrics m1, 
+                                  ResourceUsageMetrics m2, TreePath loc) 
+  throws DeepInequalityException {
+    if (m1.size() != m2.size()) {
+      throw new DeepInequalityException("Size miscompared: " + loc.toString(), 
+                                        loc);
+    }
+  }
+  
+  @Override
+  public void deepCompare(DeepCompare other, TreePath loc)
+      throws DeepInequalityException {
+    if (!(other instanceof ResourceUsageMetrics)) {
+      throw new DeepInequalityException("Comparand has wrong type", loc);
+    }
+
+    ResourceUsageMetrics metrics2 = (ResourceUsageMetrics) other;
+    compareMetric(getCumulativeCpuUsage(), metrics2.getCumulativeCpuUsage(), 
+                  new TreePath(loc, "cumulativeCpu"));
+    compareMetric(getVirtualMemoryUsage(), metrics2.getVirtualMemoryUsage(), 
+                  new TreePath(loc, "virtualMemory"));
+    compareMetric(getPhysicalMemoryUsage(), metrics2.getPhysicalMemoryUsage(), 
+                  new TreePath(loc, "physicalMemory"));
+    compareMetric(getHeapUsage(), metrics2.getHeapUsage(), 
+                  new TreePath(loc, "heapUsage"));
+    compareSize(this, metrics2, new TreePath(loc, "size"));
+  }
+}
+
diff --git a/src/tools/org/apache/hadoop/tools/rumen/TaskInfo.java b/src/tools/org/apache/hadoop/tools/rumen/TaskInfo.java
index 4cc6816..9aa6373 100644
--- a/src/tools/org/apache/hadoop/tools/rumen/TaskInfo.java
+++ b/src/tools/org/apache/hadoop/tools/rumen/TaskInfo.java
@@ -23,14 +23,22 @@
   private final long bytesOut;
   private final int recsOut;
   private final long maxMemory;
+  private final ResourceUsageMetrics metrics;
 
   public TaskInfo(long bytesIn, int recsIn, long bytesOut, int recsOut,
       long maxMemory) {
+    this(bytesIn, recsIn, bytesOut, recsOut, maxMemory, 
+         new ResourceUsageMetrics());
+  }
+  
+  public TaskInfo(long bytesIn, int recsIn, long bytesOut, int recsOut,
+                  long maxMemory, ResourceUsageMetrics metrics) {
     this.bytesIn = bytesIn;
     this.recsIn = recsIn;
     this.bytesOut = bytesOut;
     this.recsOut = recsOut;
     this.maxMemory = maxMemory;
+    this.metrics = metrics;
   }
 
   /**
@@ -70,4 +78,10 @@
     return maxMemory;
   }
 
+  /**
+   * @return Resource usage metrics
+   */
+  public ResourceUsageMetrics getResourceUsageMetrics() {
+    return metrics;
+  }
 }
diff --git a/src/tools/org/apache/hadoop/tools/rumen/ZombieJob.java b/src/tools/org/apache/hadoop/tools/rumen/ZombieJob.java
index 7c2160e..838995d 100644
--- a/src/tools/org/apache/hadoop/tools/rumen/ZombieJob.java
+++ b/src/tools/org/apache/hadoop/tools/rumen/ZombieJob.java
@@ -636,6 +636,7 @@
     long outputBytes = -1;
     long outputRecords = -1;
     long heapMegabytes = -1;
+    ResourceUsageMetrics metrics = new ResourceUsageMetrics();
 
     Values type = loggedTask.getTaskType();
     if ((type != Values.MAP) && (type != Values.REDUCE)) {
@@ -670,12 +671,15 @@
             (job.getJobReduceMB() > 0) ? job.getJobReduceMB() : job
                 .getHeapMegabytes();
       }
+      // set the resource usage metrics
+      metrics = attempt.getResourceUsageMetrics();
       break;
     }
 
     TaskInfo taskInfo =
         new TaskInfo(inputBytes, (int) inputRecords, outputBytes,
-            (int) outputRecords, (int) heapMegabytes);
+            (int) outputRecords, (int) heapMegabytes,
+            metrics);
     return taskInfo;
   }