code: package org.godhuli.rhipe.mr;
  
  /**
   * User: perk387
   * Date: 1/10/14
   */
  import java.io.File;
  import java.io.IOException;
  import java.util.*;
  
  import org.apache.commons.io.FileUtils;
  import org.apache.hadoop.fs.Path;
  import org.apache.hadoop.conf.*;
  import org.apache.hadoop.io.*;
  import org.apache.hadoop.mapreduce.*;
  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;
  import org.apache.hadoop.util.Tool;
  import org.apache.hadoop.util.ToolRunner;
  
  public class WordCount extends Configured implements Tool {
  
      /**
       * Execute the command with the given arguments.
       *
       * @param args command specific arguments.
       * @return exit code.
       * @throws Exception
       */
  //    @Override
      public static void main(String[] args) throws Exception {
          System.exit(ToolRunner.run(new WordCount(), args));
  
          //        return 0;
      }
  
      public static class Map extends Mapper<LongWritable, Text, Text, IntWritable> {
          private final static IntWritable one = new IntWritable(1);
          private Text word = new Text();
  
          public void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
              String line = value.toString();
              StringTokenizer tokenizer = new StringTokenizer(line);
              while (tokenizer.hasMoreTokens()) {
                  word.set(tokenizer.nextToken());
                  context.write(word, one);
              }
          }
      }
  
      public static class Reduce extends Reducer<Text, IntWritable, Text, IntWritable> {
  
          public void reduce(Text key, Iterable<IntWritable> values, Context context)
                  throws IOException, InterruptedException {
              int sum = 0;
              for (IntWritable val : values) {
                  sum += val.get();
              }
              context.write(key, new IntWritable(sum));
          }
      }
  
      public int run(String[] args) throws Exception {
          Configuration conf = new Configuration();
  
          Job job = Job.getInstance(conf, "wordcount");
  //        Job job = new Job(conf, "wordcount");
          job.getConfiguration().set("mapred.job.tracker","localhost:9001");
          job.getConfiguration().set("fs.default.name","hdfs://localhost:9000");
  
          job.setOutputKeyClass(Text.class);
          job.setOutputValueClass(IntWritable.class);
  
          job.setMapperClass(Map.class);
          job.setReducerClass(Reduce.class);
  
          job.setInputFormatClass(TextInputFormat.class);
          job.setOutputFormatClass(TextOutputFormat.class);
  
          File outDir = new File(args[1]);
          if(outDir.exists()) {
              FileUtils.deleteDirectory(outDir);
          }
          
          FileInputFormat.addInputPath(job, new Path(args[0]));
          FileOutputFormat.setOutputPath(job, new Path(args[1]));
          
          
  
          job.setJarByClass(WordCount.class);
          return job.waitForCompletion(true) ? 0 : 1;
          
      }
  
  }