Hadoop MapReduce Code Template

HadoopMapReduceJobTemplate.java

/**
 * Template class for Hadoop MapReduce Job
 */
public class HadoopMapReduceJobTemplate extends Configured implements Tool {
  
  /**
   * Mapper
   */
  public static class MapClass extends MapReduceBase implements Mapper<Text, Text, Text, Text> {
    public void map(Text key, Text value,
                    OutputCollector<Text, Text> output,
                    Reporter reporter) throws IOException {
      output.collect(value, key);
    }
  }
  
  /**
   * Reducer
   */
  public static class Reduce extends MapReduceBase implements Reducer<Text, Text, Text, Text> {
    public void reduce(Text key, Iterator<Text> values,
                        OutputCollector<Text, Text> output,
                        Reporter reporter) throws IOException {
      String csv = "";
      while (values.hasNext()) {
        if (csv.length() > 0) csv += ",";
          csv += values.next().toString();
        }
        output.collect(key, new Text(csv));
      }
    }
  
  /**
   * driver on Client Machine
   */
  public int run(String[] args) throws Exception {
    Configuration conf = getConf();
    
    JobConf job = new JobConf(conf, HadoopMapReduceJobTemplate.class);
    
    Path in = new Path(args[0]);
    Path out = new Path(args[1]);
    FileInputFormat.setInputPaths(job, in);
    FileOutputFormat.setOutputPath(job, out);
    
    job.setJobName("HadoopMapReduceJobTemplate");
    job.setMapperClass(MapClass.class);
    job.setReducerClass(Reduce.class);
    
    job.setInputFormat(KeyValueTextInputFormat.class);
    job.setOutputFormat(TextOutputFormat.class);
    job.setOutputKeyClass(Text.class);
    job.setOutputValueClass(Text.class);
    job.set("key.value.separator.in.input.line", ",");
    
    // Hadoop job trigger
    JobClient.runJob(job);
    
    return 0;
  }

  /**
   * Bootstrap
   */
  public static void main(String[] args) throws Exception {
    int res = ToolRunner.run(new Configuration(), new HadoopMapReduceJobTemplate(), args);
    System.exit(res);
  }
}