HDFS-14824. [Dynamometer] Dynamometer in org.apache.hadoop.tools does not output the benchmark results. (#1685)

hadoop-tools/hadoop-dynamometer/hadoop-dynamometer-infra/src/main/java/org/apache/hadoop/tools/dynamometer/Client.java

@@ -166,6 +166,7 @@ public class Client extends Configured implements Tool {
   public static final String WORKLOAD_REPLAY_ENABLE_ARG =
       "workload_replay_enable";
   public static final String WORKLOAD_INPUT_PATH_ARG = "workload_input_path";
+  public static final String WORKLOAD_OUTPUT_PATH_ARG = "workload_output_path";
   public static final String WORKLOAD_THREADS_PER_MAPPER_ARG =
       "workload_threads_per_mapper";
   public static final String WORKLOAD_START_DELAY_ARG = "workload_start_delay";
@@ -231,6 +232,8 @@ public class Client extends Configured implements Tool {
   private volatile Job workloadJob;
   // The input path for the workload job.
   private String workloadInputPath = "";
+  // The output path for the workload job metric results.
+  private String workloadOutputPath = "";
   // The number of threads to use per mapper for the workload job.
   private int workloadThreadsPerMapper;
   // The startup delay for the workload job.
@@ -347,6 +350,8 @@ public Client(String... dependencyJars) {
         + "audit logs against the HDFS cluster which is started.");
     opts.addOption(WORKLOAD_INPUT_PATH_ARG, true,
         "Location of the audit traces to replay (Required for workload)");
+    opts.addOption(WORKLOAD_OUTPUT_PATH_ARG, true,
+        "Location of the metrics output (Required for workload)");
     opts.addOption(WORKLOAD_THREADS_PER_MAPPER_ARG, true, "Number of threads "
         + "per mapper to use to replay the workload. (default "
         + AuditReplayMapper.NUM_THREADS_DEFAULT + ")");
@@ -476,6 +481,7 @@ public boolean init(String[] args) throws ParseException, IOException {
       }
       launchWorkloadJob = true;
       workloadInputPath = commandLine.getOptionValue(WORKLOAD_INPUT_PATH_ARG);
+      workloadOutputPath = commandLine.getOptionValue(WORKLOAD_OUTPUT_PATH_ARG);
       workloadThreadsPerMapper = Integer
           .parseInt(commandLine.getOptionValue(WORKLOAD_THREADS_PER_MAPPER_ARG,
               String.valueOf(AuditReplayMapper.NUM_THREADS_DEFAULT)));
@@ -1032,6 +1038,7 @@ private void launchAndMonitorWorkloadDriver(Properties nameNodeProperties) {
           + workloadStartDelayMs;
       Configuration workloadConf = new Configuration(getConf());
       workloadConf.set(AuditReplayMapper.INPUT_PATH_KEY, workloadInputPath);
+      workloadConf.set(AuditReplayMapper.OUTPUT_PATH_KEY, workloadOutputPath);
       workloadConf.setInt(AuditReplayMapper.NUM_THREADS_KEY,
           workloadThreadsPerMapper);
       workloadConf.setDouble(AuditReplayMapper.RATE_FACTOR_KEY,

hadoop-tools/hadoop-dynamometer/hadoop-dynamometer-infra/src/test/java/org/apache/hadoop/tools/dynamometer/TestDynamometerInfra.java

@@ -122,7 +122,7 @@ public class TestDynamometerInfra {
   private static final String HADOOP_BIN_PATH_KEY = "dyno.hadoop.bin.path";
   private static final String HADOOP_BIN_VERSION_KEY =
       "dyno.hadoop.bin.version";
-  private static final String HADOOP_BIN_VERSION_DEFAULT = "3.1.2";
+  private static final String HADOOP_BIN_VERSION_DEFAULT = "3.1.3";
   private static final String FSIMAGE_FILENAME = "fsimage_0000000000000061740";
   private static final String VERSION_FILENAME = "VERSION";
 
@@ -132,6 +132,8 @@ public class TestDynamometerInfra {
   private static final String NAMENODE_NODELABEL = "dyno_namenode";
   private static final String DATANODE_NODELABEL = "dyno_datanode";
 
+  private static final String OUTPUT_PATH = "/tmp/trace_output_direct";
+
   private static MiniDFSCluster miniDFSCluster;
   private static MiniYARNCluster miniYARNCluster;
   private static YarnClient yarnClient;
@@ -408,6 +410,7 @@ public void testNameNodeInYARN() throws Exception {
         return false;
       }
     }, 3000, 60000);
+    assertTrue(fs.exists(new Path(OUTPUT_PATH)));
   }
 
   private void assertClusterIsFunctional(Configuration localConf,
@@ -477,6 +480,8 @@ private Client createAndStartClient(Configuration localConf) {
             "-" + Client.WORKLOAD_REPLAY_ENABLE_ARG,
             "-" + Client.WORKLOAD_INPUT_PATH_ARG,
             fs.makeQualified(new Path("/tmp/audit_trace_direct")).toString(),
+            "-" + Client.WORKLOAD_OUTPUT_PATH_ARG,
+            fs.makeQualified(new Path(OUTPUT_PATH)).toString(),
             "-" + Client.WORKLOAD_THREADS_PER_MAPPER_ARG, "1",
             "-" + Client.WORKLOAD_START_DELAY_ARG, "10s",
             "-" + AMOptions.NAMENODE_ARGS_ARG,

hadoop-tools/hadoop-dynamometer/hadoop-dynamometer-workload/src/main/java/org/apache/hadoop/tools/dynamometer/workloadgenerator/CreateFileMapper.java

@@ -48,7 +48,8 @@
  * </ul>
  */
 public class CreateFileMapper
-    extends WorkloadMapper<NullWritable, NullWritable> {
+    extends WorkloadMapper<NullWritable, NullWritable, NullWritable,
+    NullWritable> {
 
   public static final String NUM_MAPPERS_KEY = "createfile.num-mappers";
   public static final String DURATION_MIN_KEY = "createfile.duration-min";

hadoop-tools/hadoop-dynamometer/hadoop-dynamometer-workload/src/main/java/org/apache/hadoop/tools/dynamometer/workloadgenerator/WorkloadDriver.java

@@ -32,9 +32,7 @@
 import org.apache.commons.cli.PosixParser;
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.conf.Configured;
-import org.apache.hadoop.io.NullWritable;
 import org.apache.hadoop.mapreduce.MRJobConfig;
-import org.apache.hadoop.mapreduce.lib.output.NullOutputFormat;
 import org.apache.hadoop.mapreduce.Job;
 import org.apache.hadoop.util.Tool;
 import org.apache.hadoop.util.ToolRunner;
@@ -124,7 +122,7 @@ public int run(String[] args) throws Exception {
       startTimestampMs = tmpConf.getTimeDuration(tmpConfKey, 0,
           TimeUnit.MILLISECONDS) + System.currentTimeMillis();
     }
-    Class<? extends WorkloadMapper<?, ?>> mapperClass = getMapperClass(
+    Class<? extends WorkloadMapper<?, ?, ?, ?>> mapperClass = getMapperClass(
         cli.getOptionValue(MAPPER_CLASS_NAME));
     if (!mapperClass.newInstance().verifyConfigurations(getConf())) {
       System.err
@@ -140,8 +138,9 @@ public int run(String[] args) throws Exception {
   }
 
   public static Job getJobForSubmission(Configuration baseConf, String nnURI,
-      long startTimestampMs, Class<? extends WorkloadMapper<?, ?>> mapperClass)
+      long startTimestampMs, Class<? extends WorkloadMapper<?, ?, ?, ?>>
-      throws IOException, InstantiationException, IllegalAccessException {
+      mapperClass) throws IOException, InstantiationException,
+      IllegalAccessException {
     Configuration conf = new Configuration(baseConf);
     conf.set(NN_URI, nnURI);
     conf.setBoolean(MRJobConfig.MAP_SPECULATIVE, false);
@@ -153,16 +152,9 @@ public static Job getJobForSubmission(Configuration baseConf, String nnURI,
     conf.setLong(START_TIMESTAMP_MS, startTimestampMs);
 
     Job job = Job.getInstance(conf, "Dynamometer Workload Driver");
-    job.setOutputFormatClass(NullOutputFormat.class);
     job.setJarByClass(mapperClass);
     job.setMapperClass(mapperClass);
-    job.setInputFormatClass(mapperClass.newInstance().getInputFormat(conf));
+    mapperClass.newInstance().configureJob(job);
-    job.setOutputFormatClass(NullOutputFormat.class);
-    job.setNumReduceTasks(0);
-    job.setMapOutputKeyClass(NullWritable.class);
-    job.setMapOutputValueClass(NullWritable.class);
-    job.setOutputKeyClass(NullWritable.class);
-    job.setOutputValueClass(NullWritable.class);
 
     return job;
   }
@@ -175,8 +167,8 @@ public static void main(String[] args) throws Exception {
   // The cast is actually checked via isAssignableFrom but the compiler doesn't
   // recognize this
   @SuppressWarnings("unchecked")
-  private Class<? extends WorkloadMapper<?, ?>> getMapperClass(String className)
+  private Class<? extends WorkloadMapper<?, ?, ?, ?>> getMapperClass(
-      throws ClassNotFoundException {
+      String className) throws ClassNotFoundException {
     if (!className.contains(".")) {
       className = WorkloadDriver.class.getPackage().getName() + "." + className;
     }
@@ -185,13 +177,14 @@ public static void main(String[] args) throws Exception {
       throw new IllegalArgumentException(className + " is not a subclass of "
           + WorkloadMapper.class.getCanonicalName());
     }
-    return (Class<? extends WorkloadMapper<?, ?>>) mapperClass;
+    return (Class<? extends WorkloadMapper<?, ?, ?, ?>>) mapperClass;
   }
 
   private String getMapperUsageInfo(String mapperClassName)
       throws ClassNotFoundException, InstantiationException,
       IllegalAccessException {
-    WorkloadMapper<?, ?> mapper = getMapperClass(mapperClassName).newInstance();
+    WorkloadMapper<?, ?, ?, ?> mapper = getMapperClass(mapperClassName)
+        .newInstance();
     StringBuilder builder = new StringBuilder("Usage for ");
     builder.append(mapper.getClass().getSimpleName());
     builder.append(":\n");

hadoop-tools/hadoop-dynamometer/hadoop-dynamometer-workload/src/main/java/org/apache/hadoop/tools/dynamometer/workloadgenerator/WorkloadMapper.java

@@ -21,25 +21,18 @@
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.io.NullWritable;
 import org.apache.hadoop.mapreduce.InputFormat;
+import org.apache.hadoop.mapreduce.Job;
 import org.apache.hadoop.mapreduce.Mapper;
+import org.apache.hadoop.mapreduce.lib.output.NullOutputFormat;
 
 /**
  * Represents the base class for a generic workload-generating mapper. By
  * default, it will expect to use {@link VirtualInputFormat} as its
- * {@link InputFormat}. Subclasses expecting a different {@link InputFormat}
+ * {@link InputFormat}. Subclasses requiring a reducer or expecting a different
- * should override the {@link #getInputFormat(Configuration)} method.
+ * {@link InputFormat} should override the {@link #configureJob(Job)} method.
  */
-public abstract class WorkloadMapper<KEYIN, VALUEIN>
+public abstract class WorkloadMapper<KEYIN, VALUEIN, KEYOUT, VALUEOUT> extends
-    extends Mapper<KEYIN, VALUEIN, NullWritable, NullWritable> {
+    Mapper<KEYIN, VALUEIN, KEYOUT, VALUEOUT> {
-
-  /**
-   * Return the input class to be used by this mapper.
-   * @param conf configuration.
-   * @return the {@link InputFormat} implementation for the mapper.
-   */
-  public Class<? extends InputFormat> getInputFormat(Configuration conf) {
-    return VirtualInputFormat.class;
-  }
 
   /**
    * Get the description of the behavior of this mapper.
@@ -62,4 +55,16 @@ public Class<? extends InputFormat> getInputFormat(Configuration conf) {
    */
   public abstract boolean verifyConfigurations(Configuration conf);
 
+  /**
+   * Setup input and output formats and optional reducer.
+   */
+  public void configureJob(Job job) {
+    job.setInputFormatClass(VirtualInputFormat.class);
+
+    job.setNumReduceTasks(0);
+    job.setOutputKeyClass(NullWritable.class);
+    job.setOutputValueClass(NullWritable.class);
+    job.setOutputFormatClass(NullOutputFormat.class);
+  }
+
 }

hadoop-tools/hadoop-dynamometer/hadoop-dynamometer-workload/src/main/java/org/apache/hadoop/tools/dynamometer/workloadgenerator/audit/AuditReplayMapper.java

@@ -20,6 +20,10 @@
 import com.google.common.collect.Lists;
 import java.util.Optional;
 import java.util.function.Function;
+
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.mapreduce.Job;
+import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
 import org.apache.hadoop.tools.dynamometer.workloadgenerator.WorkloadDriver;
 import org.apache.hadoop.tools.dynamometer.workloadgenerator.WorkloadMapper;
 import java.io.IOException;
@@ -35,7 +39,6 @@
 import org.apache.hadoop.fs.FileSystem;
 import org.apache.hadoop.io.LongWritable;
 import org.apache.hadoop.io.Text;
-import org.apache.hadoop.mapreduce.InputFormat;
 import org.apache.hadoop.mapreduce.MRJobConfig;
 import org.apache.hadoop.mapreduce.Mapper;
 import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
@@ -73,9 +76,11 @@
  * are replayed. For example, a rate factor of 2 would make the replay occur
  * twice as fast, and a rate factor of 0.5 would make it occur half as fast.
  */
-public class AuditReplayMapper extends WorkloadMapper<LongWritable, Text> {
+public class AuditReplayMapper extends WorkloadMapper<LongWritable, Text,
+    UserCommandKey, CountTimeWritable> {
 
   public static final String INPUT_PATH_KEY = "auditreplay.input-path";
+  public static final String OUTPUT_PATH_KEY = "auditreplay.output-path";
   public static final String NUM_THREADS_KEY = "auditreplay.num-threads";
   public static final int NUM_THREADS_DEFAULT = 1;
   public static final String CREATE_BLOCKS_KEY = "auditreplay.create-blocks";
@@ -170,11 +175,6 @@ public enum CommandType {
   private AuditCommandParser commandParser;
   private ScheduledThreadPoolExecutor progressExecutor;
 
-  @Override
-  public Class<? extends InputFormat> getInputFormat(Configuration conf) {
-    return NoSplitTextInputFormat.class;
-  }
-
   @Override
   public String getDescription() {
     return "This mapper replays audit log files.";
@@ -185,6 +185,7 @@ public List<String> getConfigDescriptions() {
     return Lists.newArrayList(
         INPUT_PATH_KEY
             + " (required): Path to directory containing input files.",
+        OUTPUT_PATH_KEY + " (required): Path to destination for output files.",
         NUM_THREADS_KEY + " (default " + NUM_THREADS_DEFAULT
             + "): Number of threads to use per mapper for replay.",
         CREATE_BLOCKS_KEY + " (default " + CREATE_BLOCKS_DEFAULT
@@ -199,7 +200,8 @@ public List<String> getConfigDescriptions() {
 
   @Override
   public boolean verifyConfigurations(Configuration conf) {
-    return conf.get(INPUT_PATH_KEY) != null;
+    return conf.get(INPUT_PATH_KEY) != null
+        && conf.get(OUTPUT_PATH_KEY) != null;
   }
 
   @Override
@@ -256,7 +258,8 @@ public void map(LongWritable lineNum, Text inputLine, Mapper.Context context)
   }
 
   @Override
-  public void cleanup(Mapper.Context context) throws InterruptedException {
+  public void cleanup(Mapper.Context context)
+      throws InterruptedException, IOException {
     for (AuditReplayThread t : threads) {
       // Add in an indicator for each thread to shut down after the last real
       // command
@@ -266,6 +269,7 @@ public void cleanup(Mapper.Context context) throws InterruptedException {
     for (AuditReplayThread t : threads) {
       t.join();
       t.drainCounters(context);
+      t.drainCommandLatencies(context);
       if (t.getException() != null) {
         threadException = Optional.of(t.getException());
       }
@@ -287,4 +291,21 @@ public void cleanup(Mapper.Context context) throws InterruptedException {
       LOG.info("Percentage of invalid ops: " + percentageOfInvalidOps);
     }
   }
+
+  @Override
+  public void configureJob(Job job) {
+    job.setMapOutputKeyClass(UserCommandKey.class);
+    job.setMapOutputValueClass(CountTimeWritable.class);
+    job.setInputFormatClass(NoSplitTextInputFormat.class);
+
+    job.setNumReduceTasks(1);
+    job.setReducerClass(AuditReplayReducer.class);
+    job.setOutputKeyClass(UserCommandKey.class);
+    job.setOutputValueClass(CountTimeWritable.class);
+    job.setOutputFormatClass(TextOutputFormat.class);
+
+    TextOutputFormat.setOutputPath(job, new Path(
+        job.getConfiguration().get(OUTPUT_PATH_KEY)));
+    job.getConfiguration().set(TextOutputFormat.SEPARATOR, ",");
+  }
 }

hadoop-tools/hadoop-dynamometer/hadoop-dynamometer-workload/src/main/java/org/apache/hadoop/tools/dynamometer/workloadgenerator/audit/AuditReplayReducer.java

@@ -0,0 +1,44 @@
+/**
+ * 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.dynamometer.workloadgenerator.audit;
+
+import org.apache.hadoop.mapreduce.Reducer;
+
+import java.io.IOException;
+
+/**
+ * AuditReplayReducer aggregates the returned latency values from
+ * {@link AuditReplayMapper} and sums them up by {@link UserCommandKey}, which
+ * combines the user's id that ran the command and the type of the command
+ * (READ/WRITE).
+ */
+public class AuditReplayReducer extends Reducer<UserCommandKey,
+    CountTimeWritable, UserCommandKey, CountTimeWritable> {
+
+  @Override
+  protected void reduce(UserCommandKey key, Iterable<CountTimeWritable> values,
+      Context context) throws IOException, InterruptedException {
+    long countSum = 0;
+    long timeSum = 0;
+    for (CountTimeWritable v : values) {
+      countSum += v.getCount();
+      timeSum += v.getTime();
+    }
+    context.write(key, new CountTimeWritable(countSum, timeSum));
+  }
+}

hadoop-tools/hadoop-dynamometer/hadoop-dynamometer-workload/src/main/java/org/apache/hadoop/tools/dynamometer/workloadgenerator/audit/AuditReplayThread.java

@@ -76,6 +76,8 @@ public class AuditReplayThread extends Thread {
   // and merge them all together at the end.
   private Map<REPLAYCOUNTERS, Counter> replayCountersMap = new HashMap<>();
   private Map<String, Counter> individualCommandsMap = new HashMap<>();
+  private Map<UserCommandKey, CountTimeWritable> commandLatencyMap
+      = new HashMap<>();
 
   AuditReplayThread(Mapper.Context mapperContext,
       DelayQueue<AuditReplayCommand> queue,
@@ -123,6 +125,14 @@ void drainCounters(Mapper.Context context) {
     }
   }
 
+  void drainCommandLatencies(Mapper.Context context)
+      throws InterruptedException, IOException {
+    for (Map.Entry<UserCommandKey, CountTimeWritable> ent
+        : commandLatencyMap.entrySet()) {
+      context.write(ent.getKey(), ent.getValue());
+    }
+  }
+
   /**
    * Add a command to this thread's processing queue.
    *
@@ -279,6 +289,14 @@ private boolean replayLog(final AuditReplayCommand command) {
         throw new RuntimeException("Unexpected command: " + replayCommand);
       }
       long latency = System.currentTimeMillis() - startTime;
+
+      UserCommandKey userCommandKey = new UserCommandKey(command.getSimpleUgi(),
+          replayCommand.toString(), replayCommand.getType().toString());
+      commandLatencyMap.putIfAbsent(userCommandKey, new CountTimeWritable());
+      CountTimeWritable latencyWritable = commandLatencyMap.get(userCommandKey);
+      latencyWritable.setCount(latencyWritable.getCount() + 1);
+      latencyWritable.setTime(latencyWritable.getTime() + latency);
+
       switch (replayCommand.getType()) {
       case WRITE:
         replayCountersMap.get(REPLAYCOUNTERS.TOTALWRITECOMMANDLATENCY)

hadoop-tools/hadoop-dynamometer/hadoop-dynamometer-workload/src/main/java/org/apache/hadoop/tools/dynamometer/workloadgenerator/audit/CountTimeWritable.java

@@ -0,0 +1,82 @@
+/**
+ * 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.dynamometer.workloadgenerator.audit;
+
+import java.io.DataInput;
+import java.io.DataOutput;
+import java.io.IOException;
+import org.apache.hadoop.io.LongWritable;
+import org.apache.hadoop.io.Writable;
+
+/**
+ * UserCommandKey is a {@link Writable} used as a composite value that
+ * accumulates the count and cumulative latency of replayed commands. It is
+ * used as the output value for AuditReplayMapper and AuditReplayReducer.
+ */
+public class CountTimeWritable implements Writable {
+  private LongWritable count;
+  private LongWritable time;
+
+  public CountTimeWritable() {
+    count = new LongWritable();
+    time = new LongWritable();
+  }
+
+  public CountTimeWritable(LongWritable count, LongWritable time) {
+    this.count = count;
+    this.time = time;
+  }
+
+  public CountTimeWritable(long count, long time) {
+    this.count = new LongWritable(count);
+    this.time = new LongWritable(time);
+  }
+
+  public long getCount() {
+    return count.get();
+  }
+
+  public long getTime() {
+    return time.get();
+  }
+
+  public void setCount(long count) {
+    this.count.set(count);
+  }
+
+  public void setTime(long time) {
+    this.time.set(time);
+  }
+
+  @Override
+  public void write(DataOutput out) throws IOException {
+    count.write(out);
+    time.write(out);
+  }
+
+  @Override
+  public void readFields(DataInput in) throws IOException {
+    count.readFields(in);
+    time.readFields(in);
+  }
+
+  @Override
+  public String toString() {
+    return getCount() + "," + getTime();
+  }
+}

hadoop-tools/hadoop-dynamometer/hadoop-dynamometer-workload/src/main/java/org/apache/hadoop/tools/dynamometer/workloadgenerator/audit/UserCommandKey.java

@@ -0,0 +1,111 @@
+/**
+ * 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.dynamometer.workloadgenerator.audit;
+
+import org.apache.hadoop.io.Text;
+import org.apache.hadoop.io.WritableComparable;
+
+import javax.annotation.Nonnull;
+import java.io.DataInput;
+import java.io.DataOutput;
+import java.io.IOException;
+import java.util.Objects;
+
+/**
+ * UserCommandKey is a {@link org.apache.hadoop.io.Writable} used as a composite
+ * key combining the user id, name, and type of a replayed command. It is used
+ * as the output key for AuditReplayMapper and the keys for AuditReplayReducer.
+ */
+public class UserCommandKey implements WritableComparable {
+  private Text user;
+  private Text command;
+  private Text type;
+
+  public UserCommandKey() {
+    user = new Text();
+    command = new Text();
+    type = new Text();
+  }
+
+  public UserCommandKey(Text user, Text command, Text type) {
+    this.user = user;
+    this.command = command;
+    this.type = type;
+  }
+
+  public UserCommandKey(String user, String command, String type) {
+    this.user = new Text(user);
+    this.command = new Text(command);
+    this.type = new Text(type);
+  }
+
+  public String getUser() {
+    return user.toString();
+  }
+
+  public String getCommand() {
+    return command.toString();
+  }
+
+  public String getType() {
+    return type.toString();
+  }
+
+  @Override
+  public void write(DataOutput out) throws IOException {
+    user.write(out);
+    command.write(out);
+    type.write(out);
+  }
+
+  @Override
+  public void readFields(DataInput in) throws IOException {
+    user.readFields(in);
+    command.readFields(in);
+    type.readFields(in);
+  }
+
+  @Override
+  public int compareTo(@Nonnull Object o) {
+    return toString().compareTo(o.toString());
+  }
+
+  @Override
+  public String toString() {
+    return getUser() + "," + getType() + "," + getCommand();
+  }
+
+  @Override
+  public boolean equals(Object o) {
+    if (this == o) {
+      return true;
+    }
+    if (o == null || getClass() != o.getClass()) {
+      return false;
+    }
+    UserCommandKey that = (UserCommandKey) o;
+    return getUser().equals(that.getUser()) &&
+        getCommand().equals(that.getCommand()) &&
+        getType().equals(that.getType());
+  }
+
+  @Override
+  public int hashCode() {
+    return Objects.hash(getUser(), getCommand(), getType());
+  }
+}

hadoop-tools/hadoop-dynamometer/hadoop-dynamometer-workload/src/test/java/org/apache/hadoop/tools/dynamometer/workloadgenerator/TestWorkloadGenerator.java

@@ -17,11 +17,15 @@
  */
 package org.apache.hadoop.tools.dynamometer.workloadgenerator;
 
+import org.apache.commons.io.IOUtils;
+import org.apache.hadoop.fs.FSDataInputStream;
 import org.apache.hadoop.tools.dynamometer.workloadgenerator.audit.AuditCommandParser;
 import org.apache.hadoop.tools.dynamometer.workloadgenerator.audit.AuditLogDirectParser;
 import org.apache.hadoop.tools.dynamometer.workloadgenerator.audit.AuditLogHiveTableParser;
 import org.apache.hadoop.tools.dynamometer.workloadgenerator.audit.AuditReplayMapper;
 import java.io.IOException;
+import java.nio.charset.StandardCharsets;
+
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.conf.Configured;
 import org.apache.hadoop.fs.FileSystem;
@@ -34,9 +38,12 @@
 import org.apache.hadoop.security.UserGroupInformation;
 import org.apache.hadoop.security.authorize.AuthorizationException;
 import org.apache.hadoop.security.authorize.ImpersonationProvider;
+import org.jline.utils.Log;
 import org.junit.After;
 import org.junit.Before;
 import org.junit.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 import static org.apache.hadoop.fs.CommonConfigurationKeysPublic.HADOOP_SECURITY_IMPERSONATION_PROVIDER_CLASS;
 import static org.junit.Assert.assertEquals;
@@ -46,6 +53,8 @@
 
 /** Tests for {@link WorkloadDriver} and related classes. */
 public class TestWorkloadGenerator {
+  private static final Logger LOG =
+      LoggerFactory.getLogger(TestWorkloadGenerator.class);
 
   private Configuration conf;
   private MiniDFSCluster miniCluster;
@@ -73,22 +82,27 @@ public void tearDown() throws Exception {
   }
 
   @Test
-  public void testAuditWorkloadDirectParser() throws Exception {
+  public void testAuditWorkloadDirectParserWithOutput() throws Exception {
     String workloadInputPath = TestWorkloadGenerator.class.getClassLoader()
         .getResource("audit_trace_direct").toString();
+    String auditOutputPath = "/tmp/trace_output_direct";
     conf.set(AuditReplayMapper.INPUT_PATH_KEY, workloadInputPath);
+    conf.set(AuditReplayMapper.OUTPUT_PATH_KEY, auditOutputPath);
     conf.setLong(AuditLogDirectParser.AUDIT_START_TIMESTAMP_KEY, 60 * 1000);
-    testAuditWorkload();
+    testAuditWorkloadWithOutput(auditOutputPath);
   }
 
   @Test
-  public void testAuditWorkloadHiveParser() throws Exception {
+  public void testAuditWorkloadHiveParserWithOutput() throws Exception {
-    String workloadInputPath = TestWorkloadGenerator.class.getClassLoader()
+    String workloadInputPath =
-        .getResource("audit_trace_hive").toString();
+        TestWorkloadGenerator.class.getClassLoader()
+            .getResource("audit_trace_hive").toString();
+    String auditOutputPath = "/tmp/trace_output_hive";
     conf.set(AuditReplayMapper.INPUT_PATH_KEY, workloadInputPath);
+    conf.set(AuditReplayMapper.OUTPUT_PATH_KEY, auditOutputPath);
     conf.setClass(AuditReplayMapper.COMMAND_PARSER_KEY,
         AuditLogHiveTableParser.class, AuditCommandParser.class);
-    testAuditWorkload();
+    testAuditWorkloadWithOutput(auditOutputPath);
   }
 
   /**
@@ -114,7 +128,8 @@ public void authorize(UserGroupInformation user, String remoteAddress)
     }
   }
 
-  private void testAuditWorkload() throws Exception {
+  private void testAuditWorkloadWithOutput(String auditOutputPath)
+      throws Exception {
     long workloadStartTime = System.currentTimeMillis() + 10000;
     Job workloadJob = WorkloadDriver.getJobForSubmission(conf,
         dfs.getUri().toString(), workloadStartTime, AuditReplayMapper.class);
@@ -132,5 +147,18 @@ private void testAuditWorkload() throws Exception {
     assertTrue(
         dfs.getFileStatus(new Path("/tmp/testDirRenamed")).isDirectory());
     assertFalse(dfs.exists(new Path("/denied")));
+
+    assertTrue(dfs.exists(new Path(auditOutputPath)));
+    try (FSDataInputStream auditOutputFile = dfs.open(new Path(auditOutputPath,
+        "part-r-00000"))) {
+      String auditOutput = IOUtils.toString(auditOutputFile,
+          StandardCharsets.UTF_8);
+      Log.info(auditOutput);
+      assertTrue(auditOutput.matches(
+          ".*(hdfs,WRITE,[A-Z]+,[13]+,[0-9]+\\n){3}.*"));
+      // Matches three lines of the format "hdfs,WRITE,name,count,time"
+      // Using [13] for the count group because each operation is run either
+      // 1 or 3 times but the output order isn't guaranteed
+    }
   }
 }

hadoop-tools/hadoop-dynamometer/src/site/markdown/Dynamometer.md

@@ -232,6 +232,9 @@ within that file. A best effort is made to faithfully replay the audit log event
 originally occurred (optionally, this can be adjusted by specifying `auditreplay.rate-factor` which is a multiplicative
 factor towards the rate of replay, e.g. use 2.0 to replay the events at twice the original speed).
 
+The AuditReplayMapper will output the benchmark results to a file `part-r-00000` in the output directory in CSV format.
+Each line is in the format `user,type,operation,numops,cumulativelatency`, e.g. `hdfs,WRITE,MKDIRS,2,150`.
+
 ### Integrated Workload Launch
 
 To have the infrastructure application client launch the workload automatically, parameters for the workload job