Page Comparison

public void beforeSubmit(MapReduceContext context) throws Exception {
  context.addOutput("cleanCounts");
  context.addOutput("invalidCounts");
  // ...
}

public static class Counter extends AbstractReducer<Text, IntWritable, byte[], Long> {
  private MultipleOutputs mos;

  @Override
  public void reduce(Text key, Iterable<IntWritable> values, ContextMapReduceTaskContext context) {
    // do computation and output to the desired dataset
    if ( ... ) {
      context.write("cleanCounts", key.getBytes(), val);
    } else {
      context.write("invalidCounts", key.getBytes(), val);
    }
  }