Word Count Example of Hadoop V1.0 – Hadoop Job的启动

gyts62

　　本文通过官方文档中的word count例子来学习如何利用Hadoop进行分布式计算。
　　1. Inputs and Outputs
　　任何一个程序，都需要有输入和输出，Hadoop MapReduce框架只能操作对，就是说Hadoop的输入和输出都是对，当然，key和value可以有多种类型。
　　key和value会被Hadoop序列化。Hadoop默认的序列化机制要求key和value实现Writable接口。除此之外，key还必须能够被排序，所以，需要让key实现WritableComparable接口。
　　下面是一个MapReduce job的简易流程：
　　(input)  -> map ->  -> combine* ->  -> reduce ->  (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完成后返回。