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

川长思鸟来 2022-10-02 06:44 1阅读 0赞

本文通过官方文档中的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完成后返回。

转载于:https://www.cnblogs.com/licheng/archive/2011/11/08/2241721.html

[Writable]: http://hadoop.apache.org/mapreduce/docs/current/api/org/apache/hadoop/io/Writable.html
[WritableComparable]: http://hadoop.apache.org/mapreduce/docs/current/api/org/apache/hadoop/io/WritableComparable.html
[Configured]: http://hadoop.apache.org/common/docs/current/api/org/apache/hadoop/conf/Configured.html#getConf%28%29