source: sample/WordCount.java @ 7

/*

 * map reduce sample code

 */

package tw.org.nchc.demo;



import java.io.IOException;

import java.util.Iterator;

import java.util.StringTokenizer;



import org.apache.hadoop.fs.FileSystem;

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.mapred.JobClient;

import org.apache.hadoop.mapred.JobConf;

import org.apache.hadoop.mapred.MapReduceBase;

import org.apache.hadoop.mapred.Mapper;

import org.apache.hadoop.mapred.OutputCollector;

import org.apache.hadoop.mapred.Reducer;

import org.apache.hadoop.mapred.Reporter;





public class WordCount {



  // mapper: emits (token, 1) for every word occurrence

  private static class MapClass extends MapReduceBase implements

      Mapper<LongWritable, Text, Text, IntWritable> {



    // reuse objects to save overhead of object creation

    private final static IntWritable one = new IntWritable(1);

    private Text word = new Text();



    public void map(LongWritable key, Text value,

        OutputCollector<Text, IntWritable> output, Reporter reporter)

        throws IOException {

      String line = ((Text) value).toString();

      StringTokenizer itr = new StringTokenizer(line);

      while (itr.hasMoreTokens()) {

        word.set(itr.nextToken());

        output.collect(word, one);

      }

    }

  }



  // reducer: sums up all the counts

  private static class ReduceClass extends MapReduceBase implements

      Reducer<Text, IntWritable, Text, IntWritable> {



    // reuse objects

    private final static IntWritable SumValue = new IntWritable();



    public void reduce(Text key, Iterator<IntWritable> values,

        OutputCollector<Text, IntWritable> output, Reporter reporter)

        throws IOException {

      // sum up values

      int sum = 0;

      while (values.hasNext()) {

        sum += values.next().get();

      }

      SumValue.set(sum);

      output.collect(key, SumValue);

    }

  }



  

  /**

   * Runs the demo.

   */

  public static void main(String[] args) throws IOException {

    String filename = "/user/waue/input/";

    String outputPath = "sample-counts";

    int mapTasks = 20;

    int reduceTasks = 1;



    JobConf conf = new JobConf(WordCount.class);

    conf.setJobName("wordcount");



    conf.setNumMapTasks(mapTasks);

    conf.setNumReduceTasks(reduceTasks);



    conf.setInputPath(new Path(filename));

    conf.setOutputKeyClass(Text.class);

    conf.setOutputValueClass(IntWritable.class);

    conf.setOutputPath(new Path(outputPath));



    conf.setMapperClass(MapClass.class);

    conf.setCombinerClass(ReduceClass.class);

    conf.setReducerClass(ReduceClass.class);

    

    // Delete the output directory if it exists already

    Path outputDir = new Path(outputPath);

    FileSystem.get(conf).delete(outputDir);



    JobClient.runJob(conf);

  }

}