Map

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);
            }
        }
}

基本上，上面建立了一個Mapclass繼承Mapper並且定義了一個方法叫做map。

Hadoop會把每一段文字個用value傳過來，因此用了tokenizer把裡面的word取出來。

每一個取出來的word，會被寫成一組key value pair(context.write(word,one))，word是key，value是數值1。

會一直做，直到整個word都處理完。

Reduce

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));
 }
}

Reducer和mapper類似，先定義一個class叫做Reduce繼承Reducer。

裡面有一個reduce的程式定義reduce階段要做什麼

在這邊，java已經有處理好把一樣的key放成一組，因此可以透過迴圈的方式把所有值加總。

最後把整個結果寫出去，一樣是key value pair，key還是原來的key，不過value是所有的加總。

設定

Map階段和Reduce階段的功能都定義好了之後，接下來需要做的是告訴程式執行的時候那個是Map和那個是Reduce。

package org.myorg;

import java.io.IOException;
import java.util.*;

import org.apache.hadoop.fs.Path;
import org.apache.hadoop.conf.*;
import org.apache.hadoop.io.*;
import org.apache.hadoop.mapred.*;
import org.apache.hadoop.util.*;

public class WordCount {

 // 剛剛定義的 Map
 ....

 // 剛剛定義的 Reduce
 ....

 public static void main(String[] args) throws Exception {
  JobConf conf = new JobConf(WordCount.class);
  conf.setJobName("wordcount");

  conf.setOutputKeyClass(Text.class);
  conf.setOutputValueClass(IntWritable.class);

  conf.setMapperClass(Map.class);
  conf.setCombinerClass(Reduce.class);
  conf.setReducerClass(Reduce.class);

  conf.setInputFormat(TextInputFormat.class);
  conf.setOutputFormat(TextOutputFormat.class);

  FileInputFormat.setInputPaths(conf, new Path(args[0]));
  FileOutputFormat.setOutputPath(conf, new Path(args[1]));

  JobClient.runJob(conf);
 }
}

這個程式應該蠻好理解，基本上就是把剛剛定義好的Map和Reduce做設定。

這邊比較特別是Combiner的部分，因為也是加總所以和reduce是一樣的概念。