学习Hadoop MapReduce与WordCount例子分析

/*
MapReduce框架一直围绕着key-value这样的数据结构，下面以官方自带的WordCount为例子，自己分析MapReduce的工作机制。MapReduce可以分为Map和Reduce过程，
代码实现了两个类，分别是继承Mapper和Reduceer，Mapper类里面有map接口，Reduceer类有reduce接口，对于统计单词这个例子来说，MapReduce会把文件以行为
拆分对象，每分析一行就会调用Mapper类里面的map接口，然后map接口里面的代码由程序员实现其逻辑，然后把map接口处理完的结果输送给Reduceer的reduce的接
口，中间还可以插入一个combiner的接口用于对map接口的数据进行中间结果处理再丢给reduce做最终的汇总。具体流程看代码注释。
*/

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

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;

public class WordCount {

public static class TokenizerMapper
extends Mapper<Object, Text, Text, IntWritable>{
/*
Mapper他是一个模板类，Class Mapper<KEYIN,VALUEIN,KEYOUT,VALUEOUT>，
KEYIN 输入key的类型，VALUEIN输入value的类型
KEYOUT 输出key的类型，VALUEOUT输出value的类型
四个类型决定了map接口的输入与输出类型

比较形象地描述key，value，在map，combiner，reduce流转的
(input) <k1, v1> -> map -> <k2, v2> -> combine -> <k2, v2> -> reduce -> <k3, v3> (output)

其中还有规定，就是KEY和VALUE类型必须是实现了Writeable接口的，KEY类型还需要额外实现WritableComparable接口

通常在Mapper模板里面，KEYIN是不需要特指定的，就用基类Object就可以了，VAULEIN指定为Text，这个Text是<pre name="code" class="java">    org.apache.hadoop.io.Text，这个Text已经满足了实现Writeable接口的条件了，在这个例子里面VALUE就是文件的行内
容，所以定义类型为Text。
对于KEYOUT和VALUEOUT，作为输出key类型和value类型，这里定义为Text和IntWritable，keyout就是需要统计单词个数
的单词，IntWriteable就是对应某个单词的次数，其实这个就是一个Int类型，为了符合接口需要所以就基础了Writeable
Context它是一个贯通map接口<-->combiner接口<-->reduce接口的上下文数据，在map接口里面，单词对应次数会保存在context
里面，到了reduce接口，MapReduce会把之前map的context用key对应结果集合的形式给reduce接口。
*/

private final static IntWritable one = new IntWritable(1); private Text word = new Text(); public void map(Object key, Text value, Context context ) throws IOException, InterruptedException { StringTokenizer itr = new StringTokenizer(value.toString()); while (itr.hasMoreTokens()) { word.set(itr.nextToken()); context.write(word, one); } } }

/*
下面是对两个文件统计单词调用map接口之后的context结果
For the given sample input the first map emits:
< Hello, 1>
< World, 1>
< Bye, 1>
< World, 1>

The second map emits:
< Hello, 1>
< Hadoop, 1>
< Goodbye, 1>
< Hadoop, 1>
*/

public static class IntSumReducer extends Reducer<Text,IntWritable,Text,IntWritable> { /*
Reduceer也是一个类模板，跟Mapper一样需要指定KEYIN，VALUEIN，KEYOUT，VALUEOUT，
其中KEYIN和VALUEIN必须跟Mapper的KEYOUT，VALUEOUT一一对应，因为map接口输出的结果key->value
就是reduce接口的输入，只是MapReduce框架把map接口里面相同的key变成一个key->values
的values集合，所以在reduce接口里面KEYIN是Text也就是单词，VALUEOUT是IntWriteable集合的
迭代器Interable<IntWriteable>，context就是reduce的输出结果了

*/

private IntWritable result = new IntWritable(); public 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); } }

/*
在例子里面，还指定了combiner，其实cominer和reduce都是同一个接口reduce，第一次调用reduce接口是combiner过程，把每个文件
的单词做了key->value 到 key->values的汇总，结果如下
The output of the first map:
< Bye, 1>
< Hello, 1>
< World, 2>

The output of the second map:
< Goodbye, 1>
< Hadoop, 2>
< Hello, 1>
*/

/*
第二次调用reduce接口，就是reduce的过程，把combiner处理过的中间结果做一次最终的汇总
< Bye, 1>
< Goodbye, 1>
< Hadoop, 2>
< Hello, 2>
< World, 2>
*/

public static void main(String[] args) throws Exception {
Configuration conf = new Configuration();
Job job = Job.getInstance(conf, "word count");
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(args[0]));
FileOutputFormat.setOutputPath(job, new Path(args[1]));
System.exit(job.waitForCompletion(true) ? 0 : 1);
}
}