Using Counters in MapReduce to Track Bad Records

The MapReduce framework provides Counters as an efficient mechanism for tracking the occurrences of global events within the map and reduces the phases of jobs. For example, a typical MapReduce job will kick off several mapper instances, one for each block of the input data, all running the same code. These instances are part of the same job, but run independent of one another. Counters allow a developer to track aggregated events from all of those separate instances. A more concrete use of Counters can be found in the MapReduce framework itself. Each MapReduce job defines several standard Counters. The output of these Counters can be found in the job details of the Job Tracker web UI.

Counter Map Reduce Total

Map input records 71,228,085,554 0 71,228,085,554

Reduce shuffle bytes 0 5,419,119,866,175 5,419,119,866,175

Spilled Records 142,174,003,132 55,141,286,320 197,315,289,452

Map output bytes 5,513,542,463,958 0 5,513,542,463,958

CPU_MILLISECONDS 2,658,192,290 2,352,883,960 5,011,076,250

Combine input records 139,937,676,997 69,478,386,276 209,416,063,273

SPLIT_RAW_BYTES 1,999,232 0 1,999,232

Reduce input records 0 58,006,566,989 58,006,566,989

Reduce input groups 0 3,331,430 3,331,430

Combine output records 139,937,676,997 6,478,386,262 209,416,063,259

PHYSICAL_MEMORY_BYTES 12,647,158,763,520 196,662,726,656 12,843,821,490,176

Reduce output records 0 406,433,728 406,433,728

VIRTUAL_MEMORY_BYTES 33,014,737,301,504 273,806,381,056 33,288,543,682,560

Map output records 71,228,085,554/td] 0 71,228,085,554

GC time elapsed (ms) 98,623,849 342,650,729 441,274,578

The UI shows the Counter group, name, mapper totals, reducer totals, and job totals. Counters should be limited to tracking metadata about the job. The standard Counters are good examples of this. The Map input records counter provides useful information about a particular execution of a job. If Counters did not exist, these kinds of statistics would have to be part of the job’s main output, where they don’t belong; or more likely as part of a secondary output, complicating the logic of the job.

The following recipe is a simple map-only job that filters out bad records and uses a counter to log the number of records that were filtered.

Suppose you have written a MapReduce job that contains the following code:

public class Example {
 private enum COUNTERS {
  GOOD,
  BAD
 }

 public static void main(String[] args) throws Exception {
  Configuration conf = new Configuration();
  Job job = new Job(conf);

  job.setMapperClass(Example.MapTask.class);
  job.setOutputKeyClass(Text.class);
  job.setOutputValueClass(NullWritable.class);

  FileInputFormat.addInputPath(job, new Path(args[0]));
  FileOutputFormat.setOutputPath(job, new Path(args[1]));

  job.waitForCompletion(false);

  Counters counters = job.getCounters();

  System.out.printf("Good: %d, Bad: %dn",
      counters.findCounter(COUNTERS.GOOD).getValue(),
      counters.findCounter(COUNTERS.BAD).getValue());
 }

 public static class MapTask
   extends Mapper<LongWritable, Text, Text, NullWritable> {
  private Pattern p =
    Pattern.compile("^(["']?)dd:dd1,(["']?)[A-Z]w+2,.*$");

  public void map(LongWritable key, Text value, Context context)
    throws IOException, InterruptedException {
   if (p.matcher(value.toString()).matches()) {
    context.getCounter(COUNTERS.GOOD).increment(1L);
    context.write(value, NullWritable.get());
   } else {
    context.getCounter(COUNTERS.BAD).increment(1L);
   }
  }
 }
}

Assuming the output directory in HDFS does not exist, what would be the command-line output (ignoring log messages) of this application if the contents of the input.csv were as follows:

9:58,Added,Queue,18762
10:23,Added,Queue,7432e01
10:53,"Removed","Queue","7432e01"
'11:01',"Logged Out",

Undoubtedly, the answer would be Good:2, Bad:2. Lets see how:

1. The first record is marked bad because the hour is single-digit, and the regular expression requires double-digit hour and minute.
2. The second record is marked good because the time is in the correct format, the word in the second column starts with a capital and contains only letters, and there are more than two columns.
3. The third record is marked good because the time is in the correct format, the word in the second column starts with a capital, contains only letters, and is wrapped in matching quotes, and there are more than two columns.
4. The fourth row is marked bad because the second column contains a space. The regex doesn’t allow spaces, quotes or not.

The false in waitForCompletion() turns off the progress output; it has nothing to do with whether the driver will wait for the job to complete.

Assuming the output directory in HDFS does not exist, what would be the command-line output (ignoring log messages) of this application if the contents of the input.csv were as follows:

10:23,Added,,queue 2,7432e01
'10:28','Added','Queue","56845'
11:01,"Logout"
11:01,X,Y,Z

Again, the answer would be Good:2, Bad:2. Let us check how:

1. The first record is marked good because the time is in the correct format, the word in the second column starts with a capital and contains only letters, and there are more than two columns.
2. The second record is marked good because the time is in the correct format and is wrapped in matching quotes, the word in the second column starts with a capital, contains only letters, and is wrapped in matching quotes, and there are more than two columns.
3. The third record is marked bad because it does not have more than two columns.
4. The fourth row is marked bad because the second column contains only a single character. The regular expression requires a capital letter followed by at least one additional character.

The false in waitForCompletion() turns off the progress output; it has nothing to do with whether the driver will wait for the job to complete.