MapReduce之單詞計數

最近在看google那篇經典的MapReduce論文，中文版可以參考孟岩推薦的 mapreduce 中文版 中文翻譯

論文中提到，MapReduce的編程模型就是：

計算利用一個輸入key/value對集,來產生一個輸出key/value對集.MapReduce庫的用戶用兩個函數表達這個計算:map和reduce.

用戶自定義的map函數,接受一個輸入對,然后產生一個中間key/value對集.MapReduce庫把所有具有相同中間key I的中間value聚合在一起,然后把它們傳遞給reduce函數.

用戶自定義的reduce函數,接受一個中間key I和相關的一個value集.它合並這些value,形成一個比較小的value集.一般的,每次reduce調用只產生0或1個輸出value.通過一個迭代器把中間value提供給用戶自定義的reduce函數.這樣可以使我們根據內存來控制value列表的大小.

那么研究MapReduce，一般是從hadoop開始，研究編程語言，一般從helloworld開始，那么我們研究hadoop，就先從官方實例wordcount開始。

按照上面提到的編程模型：

用戶自定義的map函數,接受一個輸入對,然后產生一個中間key/value對集.MapReduce庫把所有具有相同中間key I的中間value聚合在一起,然后把它們傳遞給reduce函數.

那么對於單詞計數這個程序來說：

map函數對輸入的文本進行分詞處理，然后輸出（單詞， 1）這樣的結果，例如“You are a young man”，輸出的就是（you， 1）， （are， 1） 之類的結果

 

代碼如下：

class TokenizerMapper extends Mapper<Object, Text, Text, IntWritable> {
    private final static IntWritable one = new IntWritable(1);
    private Text word = new Text();

    @Override
    protected void map(Object key, Text value, Context context) throws IOException, InterruptedException {
        StringTokenizer tokenizer = new StringTokenizer(value.toString());
        while (tokenizer.hasMoreTokens())
        {
            word.set(tokenizer.nextToken());
            context.write(word, one);
        }
    }
}

上面提到map函數的輸入也是k-v堆，從模板參數中可以看出。這個map函數的輸入K-V類型為 <Object, Text>

而map函數的輸出類型為<Text, IntWritable>，而這恰好就是reduce函數的輸入類型

 

reduce函數：

用戶自定義的reduce函數,接受一個中間key I和相關的一個value集.它合並這些value,形成一個比較小的value集.一般的,每次reduce調用只產生0或1個輸出value.通過一個迭代器把中間value提供給用戶自定義的reduce函數.這樣可以使我們根據內存來控制value列表的大小.

在單詞計數中，我們把具有相同key的結果聚合起來：

class IntSumReducer extends Reducer<Text, IntWritable, Text, IntWritable> {
    private IntWritable result = new IntWritable();

    @Override
    protected void reduce(Text key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException {
        int sum = 0;
        for (IntWritable val : values){
            sum += val.get();
        }
        result.set(sum);
        context.write(key, result);
    }
}

reduce函數的第二個參數類型為Iterable<IntWritable>， 這是一堆value的集合，他們具有相同的key，reduce函數的意義就是將這些結果聚合起來。

例如（”hello“， 1）和（”hello“， 1）聚合為（”hello“， 2），后者可能再次和（”hello“， 3） （”hello“， 1），聚合為（”hello“， 7）

可以通過控制values的大小，防止內存溢出，合理使用內存。

reduce函數的結果存儲到磁盤上，就是我們最終的結果。

 

完整的代碼為：

package com.zhihu;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

import java.io.IOException;
import java.util.StringTokenizer;

/**
 * Created by guochunyang on 15/9/22.
 */
public class WordCount {

    public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {
        Configuration conf = new Configuration();
        Job job = Job.getInstance(conf, "wordcount");
        job.setJarByClass(WordCount.class);

        job.setMapperClass(TokenizerMapper.class);
        job.setCombinerClass(IntSumReducer.class);
        job.setReducerClass(IntSumReducer.class);

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

        FileInputFormat.addInputPath(job, new Path("in"));
        FileOutputFormat.setOutputPath(job, new Path("out"));

        job.waitForCompletion(true);
    }
}

class TokenizerMapper extends Mapper<Object, Text, Text, IntWritable> {
    private final static IntWritable one = new IntWritable(1);
    private Text word = new Text();

    @Override
    protected void map(Object key, Text value, Context context) throws IOException, InterruptedException {
        StringTokenizer tokenizer = new StringTokenizer(value.toString());
        while (tokenizer.hasMoreTokens())
        {
            word.set(tokenizer.nextToken());
            context.write(word, one);
        }
    }
}

class IntSumReducer extends Reducer<Text, IntWritable, Text, IntWritable> {
    private IntWritable result = new IntWritable();

    @Override
    protected void reduce(Text key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException {
        int sum = 0;
        for (IntWritable val : values){
            sum += val.get();
        }
        result.set(sum);
        context.write(key, result);
    }
}