Hadoop Chain Mapper - Example

Objective:  

              Using  Hadoop  chainmapper how to chain multiple Mapper before it reaches reducer.

Here first Mapper class output will be the input for second Mapper class input.

Pattern : Mapper1 -> Mapper2 -> Reducer

Here I am taken simple wordcount example.

1st Mapper  will read the file from input txt file and split each text and store it in Context.

2nd Mapper will get 1stMapper output and convert all the key text  into Lower case key.

Lower case key will be stored into Context.

Reducer will get the value from 2ndMapper , Same key related values will go to one reducer task.

In Reducer we are simply doing word count operation with the given key.

SplitMapper.java [ 1st Mapper class ]

package com.tamil; import java.io.IOException; import java.util.StringTokenizer; import org.apache.hadoop.io.IntWritable; import org.apache.hadoop.io.LongWritable; import org.apache.hadoop.io.Text; import org.apache.hadoop.mapreduce.Mapper; public class SplitMapper extends Mapper<LongWritable, Text, Text, IntWritable> {               @Override               protected void map(LongWritable key, Text value, Context context)                                            throws IOException, InterruptedException {                              StringTokenizer tokenizer = new StringTokenizer(value.toString());                              IntWritable dummyValue = new IntWritable(1);                              while (tokenizer.hasMoreElements()) {                                            String content = (String) tokenizer.nextElement();                                            context.write(new Text(content), dummyValue);                              }               } }

LowerCaseMapper.java [ 2nd Mapper class ]

package com.tamil; import java.io.IOException; import org.apache.hadoop.io.IntWritable; import org.apache.hadoop.io.Text; import org.apache.hadoop.mapreduce.Mapper; public class LowerCaseMapper extends Mapper<Text, IntWritable, Text, IntWritable> {               @Override               protected void map(Text key, IntWritable value, Context context)                                            throws IOException, InterruptedException {                              String val = key.toString().toLowerCase();                              Text newKey = new Text(val);                              context.write(newKey, value);               } }

ChainMapReducer.java [ Reducer class ]

package com.tamil; import java.io.IOException; import org.apache.hadoop.io.IntWritable; import org.apache.hadoop.io.Text; import org.apache.hadoop.mapreduce.Reducer; public class ChainMapReducer extends Reducer<Text, IntWritable, Text, IntWritable> {               @Override               protected void reduce(Text key, Iterable<IntWritable> values,               Context context) throws IOException, InterruptedException {                              int sum = 0;                              for (IntWritable value : values) {sum += value.get();}                              context.write(key, new IntWritable(sum));               } }

ChainMapperDriver.java [ Driver class]

package com.tamil; import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.fs.Path; import org.apache.hadoop.io.IntWritable; import org.apache.hadoop.io.LongWritable; import org.apache.hadoop.io.Text; import org.apache.hadoop.mapreduce.Job; import org.apache.hadoop.mapreduce.lib.chain.ChainMapper; import org.apache.hadoop.mapreduce.lib.input.FileInputFormat; import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat; import org.apache.hadoop.util.GenericOptionsParser; public class ChainMapperDriver {               public static void main(String[] args) throws Exception {                              Configuration config = new Configuration();                              String[] otherArgs = new GenericOptionsParser(config, args).getRemainingArgs();                              if (otherArgs.length != 2) {                                            System.err.print("Useage: wordcount <in> <out>");                                            System.exit(2);                              }                              Job job = Job.getInstance();                              Configuration splitMapConfig = new Configuration(false);                              ChainMapper.addMapper(job, SplitMapper.class, LongWritable.class,                                                           Text.class, Text.class, IntWritable.class, splitMapConfig);                              Configuration lowerCaseMapConfig = new Configuration(false);                              ChainMapper.addMapper(job, LowerCaseMapper.class, Text.class,                                                           IntWritable.class, Text.class, IntWritable.class,                                                           lowerCaseMapConfig);                              job.setJarByClass(ChainMapperDriver.class);                              job.setCombinerClass(ChainMapReducer.class);                              job.setReducerClass(ChainMapReducer.class);                              job.setOutputKeyClass(Text.class);                              job.setOutputValueClass(IntWritable.class);                              FileInputFormat.addInputPath(job, new Path(otherArgs[0]));                              FileOutputFormat.setOutputPath(job, new Path(otherArgs[1]));                              System.exit(job.waitForCompletion(true) ? 0 : 1);               } }

That's it.

Create a jar file & run it :-)