本文通过官方文档中的word count例子来学习如何利用Hadoop进行分布式计算。
1. Inputs and Outputs
任何一个程序，都需要有输入和输出，Hadoop MapReduce框架只能操作<Key, Value>对，就是说Hadoop的输入和输出都是<Key, Value>对，当然，key和value可以有多种类型。
key和value会被Hadoop序列化。Hadoop默认的序列化机制要求key和value实现Writable接口。除此之外，key还必须能够被排序，所以，需要让key实现WritableComparable接口。
下面是一个MapReduce job的简易流程：
(input) <k1, v1> -> map -> <k2, v2> -> combine* -> <k2, v2> -> reduce -> <k3, v3> (output)
注意，其中的combine phase执行的次数可能是0次，也可能是多次。
2. Example: Word Count 1.0
WordCount，顾名思义，也就是把输入中出现的所有词的个数统计出来。
首先我们来看代码，在WordCount类中实现下面两个函数：
01 public int run(String [] args) throws Exception {
02 Job job = new Job(getConf());
03 job.setJarByClass(WordCount.class);
04 job.setJobName("wordcount");
05
06 job.setOutputKeyClass(Text.class);
07 job.setOutputValueClass(IntWritable.class);
08
09 job.setMapperClass(Map.class);
10 job.setCombinerClass(Reduce.class);
11 job.setReducerClass(Reduce.class);
12
13 job.setInputFormatClass(TextInputFormat.class);
14 job.setOutputFormatClass(TextOutputFormat.class);
15
16 FileInputFormat.setInputPaths(job, new Path(args[0]));
17 FileOutputFormat.setOutputPath(job, new Path(args[1]));
18
19 boolean success = job.waitForCompletion(true);
20 return success ? 0 : 1;
21 }
22
23 public static void main(String[] args) throws Exception {
24 int ret = ToolRunner.run(new WordCount(), args);
25 System.exit(ret);
26 }
其中main函数是入口，其中ToolRunner.run(new WordCount(), args)启动MapReduce job。WordCount是程序员实现的MapReduce逻辑。ToolRunner.run(new WordCount(), args)相当于执行了WordCount.java的run(args)。
下面我们来看一下run函数:
02 Job job = new Job(getConf());
03 job.setJarByClass(WordCount.class);
04 job.setJobName("wordcount");
首先根据Configuration来初始化job，可能你会奇怪，getConf()的实现在哪？这个方法继承自Configured类。然后setJarByClass，通过class文件来确定jar文件。最后给MapReduce Job起个名字。
06 job.setOutputKeyClass(Text.class);
07 job.setOutputValueClass(IntWritable.class);
这两行是确定输出的KeyValue对的类型分别是什么。Text对应String，IntWritable即integer。
09 job.setMapperClass(Map.class);
10 job.setCombinerClass(Reduce.class);
11 job.setReducerClass(Reduce.class);
指定mapper和reducer的class。其中Combiner的作用是对mapper输出的中间结果做local aggregation，这样可以降低mapper和reducer之间传输的数据量。
13 job.setInputFormatClass(TextInputFormat.class);
14 job.setOutputFormatClass(TextOutputFormat.class);
设置输入输出文件的类型。对于TextInputFormat.class，即一行作为一个map的输入，然后默认按空白切分。
16 FileInputFormat.setInputPaths(job, new Path(args[0]));
17 FileOutputFormat.setOutputPath(job, new Path(args[1]));
设置输入输出文件的路径
19 boolean success = job.waitForCompletion(true);
job.waitForCompletion(true)把Job提交给JobTracker，并且监控该job，直到job完成后返回。