Hadoop Counters Example or How to track the number of records processed by mapper and reducer

We should use Hadoop counters as much as possible in Map reduce programs. So that we can keep track of number of records processed in Mappers and Reducers.

Input
----------------------
Johny, Johny!
Yes, Papa
Eating sugar?
No, Papa
Telling lies?
No, Papa
Open your mouth!
Ha! Ha! Ha!

output Console
------------------------------
Total Number of Records Processed in MAP: 8
Total Number of Records Processed in Reducer: 8

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85

package com.my.cert.example; import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.fs.Path; import org.apache.hadoop.io.LongWritable; import org.apache.hadoop.io.NullWritable; import org.apache.hadoop.io.Text; import org.apache.hadoop.mapreduce.Counter; import org.apache.hadoop.mapreduce.Job; import org.apache.hadoop.mapreduce.Mapper; import org.apache.hadoop.mapreduce.Reducer; import org.apache.hadoop.mapreduce.lib.input.FileInputFormat; import org.apache.hadoop.mapreduce.lib.input.TextInputFormat; import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat; import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat; public class CounterUseCaseExp { public static void main(String[] args) throws Exception { Path inputPath = new Path("C:\\hadoop\\test\\test.txt"); Path outputDir = new Path("C:\\hadoop\\test\\test1"); // Create configuration Configuration conf = new Configuration(true); // Create job Job job = new Job(conf, "Hadoop Counter Example"); job.setJarByClass(CounterUseCaseExp.class); // Setup MapReduce job.setMapperClass(CounterUseCaseExp.MapTask.class); job.setReducerClass(CounterUseCaseExp.ReduceTask.class); job.setNumReduceTasks(1); job.setOutputKeyClass(NullWritable.class); job.setOutputValueClass(Text.class); // Input FileInputFormat.addInputPath(job, inputPath); job.setInputFormatClass(TextInputFormat.class); // Output FileOutputFormat.setOutputPath(job, outputDir); job.setOutputFormatClass(TextOutputFormat.class); int code = job.waitForCompletion(true) ? 0 : 1; Counter mapperCounter = job.getCounters().findCounter(MapTask.MapCounters.MAP_RECORD_COUNTER); Counter reducerCounter = job.getCounters().findCounter(ReduceTask.ReducerCounters.REDUCER_RECORD_COUNTER); System.out.println("Total Number of Records Processed in MAP: "+mapperCounter.getValue()); System.out.println("Total Number of Records Processed in Reducer: "+reducerCounter.getValue()); System.exit(code); } public static class MapTask extends Mapper<LongWritable, Text, NullWritable, Text> { static enum MapCounters { MAP_RECORD_COUNTER } public void map(LongWritable key, Text value, Context context) throws java.io.IOException, InterruptedException { context.getCounter(MapCounters.MAP_RECORD_COUNTER).increment(1); String line = value.toString(); context.write(NullWritable.get(), new Text(line)); } } public static class ReduceTask extends Reducer<NullWritable, Text, NullWritable, Text> { static enum ReducerCounters { REDUCER_RECORD_COUNTER } public void reduce(NullWritable key, Iterable<Text> list, Context context) throws java.io.IOException, InterruptedException { for (Text item : list) { context.write(key, item); context.getCounter(ReducerCounters.REDUCER_RECORD_COUNTER).increment(1); } } } }