WORD COUNT

Posted 2020-10-09

单词计数是最简单也是最能体现MapReduce思想的程序之一，可以称为MapReduce版"Hello World"，该程序的完整代码可以在Hadoop安装包的"src/examples"目录下找到。单词计数主要完成功能是：统计一系列文本文件中每个单词出现的次数

程序的功能：假设现在有n个文本，WordCount程序就是利用MR计算模型来统计这n个文本中每个单词出现的总次数。

程序的源代码

1. import java.io.IOException;  
2. import java.util.ArrayList;  
3. import java.util.Iterator;  
4. import java.util.List;  
5. import java.util.StringTokenizer;  
6.   
7. import org.apache.hadoop.conf.Configuration;  
8. import org.apache.hadoop.conf.Configured;  
9. import org.apache.hadoop.fs.Path;  
10. import org.apache.hadoop.io.IntWritable;  
11. import org.apache.hadoop.io.LongWritable;  
12. import org.apache.hadoop.io.Text;  
13. import org.apache.hadoop.mapred.FileInputFormat;  
14. import org.apache.hadoop.mapred.FileOutputFormat;  
15. import org.apache.hadoop.mapred.JobClient;  
16. import org.apache.hadoop.mapred.JobConf;  
17. import org.apache.hadoop.mapred.MapReduceBase;  
18. import org.apache.hadoop.mapred.Mapper;  
19. import org.apache.hadoop.mapred.OutputCollector;  
20. import org.apache.hadoop.mapred.Reducer;  
21. import org.apache.hadoop.mapred.Reporter;  
22. import org.apache.hadoop.util.Tool;  
23. import org.apache.hadoop.util.ToolRunner;  
24.   
25. public class WordCount extends Configured implements Tool {  
26.   
27.     public static class MapClass extends MapReduceBase implements  
28.             Mapper<LongWritable, Text, Text, IntWritable> {  
29.   
30.         private final static IntWritable one = new IntWritable(1);  
31.         private Text word = new Text();  
32.   
33.         public void map(LongWritable key, Text value,  
34.                 OutputCollector<Text, IntWritable> output, Reporter reporter)  
35.                 throws IOException {  
36.             String line = value.toString();  
37.             StringTokenizer itr = new StringTokenizer(line);  
38.             while (itr.hasMoreTokens()) {  
39.                 word.set(itr.nextToken());  
40.                 output.collect(word, one);  
41.             }  
42.         }  
43.     }  
44.   
45.     /** 
46.      * A reducer class that just emits the sum of the input values. 
47.      */  
48.     public static class Reduce extends MapReduceBase implements  
49.             Reducer<Text, IntWritable, Text, IntWritable> {  
50.   
51.         public void reduce(Text key, Iterator<IntWritable> values,  
52.                 OutputCollector<Text, IntWritable> output, Reporter reporter)  
53.                 throws IOException {  
54.             int sum = 0;  
55.             while (values.hasNext()) {  
56.                 sum += values.next().get();  
57.             }  
58.             output.collect(key, new IntWritable(sum));  
59.         }  
60.     }  
61.   
62.     static int printUsage() {  
63.         System.out.println("wordcount [-m <maps>] [-r <reduces>] <input> <output>");  
64.         ToolRunner.printGenericCommandUsage(System.out);  
65.         return -1;  
66.     }  
67.   
68.     /** 
69.      * The main driver for word count map/reduce program. Invoke this method to 
70.      * submit the map/reduce job. 
71.      * 
72.      * @throws IOException 
73.      *             When there is communication problems with the job tracker. 
74.      */  
75.     public int run(String[] args) throws Exception {  
76.         JobConf conf = new JobConf(getConf(), WordCount.class);  
77.         conf.setJobName("wordcount");  
78.   
79.         // the keys are words (strings)  
80.         conf.setOutputKeyClass(Text.class);  
81.         // the values are counts (ints)  
82.         conf.setOutputValueClass(IntWritable.class);  
83.   
84.         conf.setMapperClass(MapClass.class);  
85.         conf.setCombinerClass(Reduce.class);  
86.         conf.setReducerClass(Reduce.class);  
87.   
88.         List<String> other_args = new ArrayList<String>();  
89.         for (int i = 0; i < args.length; ++i) {  
90.             try {  
91.                 if ("-m".equals(args[i])) {  
92.                     conf.setNumMapTasks(Integer.parseInt(args[++i]));  
93.                 } else if ("-r".equals(args[i])) {  
94.                     conf.setNumReduceTasks(Integer.parseInt(args[++i]));  
95.                 } else {  
96.                     other_args.add(args[i]);  
97.                 }  
98.             } catch (NumberFormatException except) {  
99.                 System.out.println("ERROR: Integer expected instead of "  
100.                         + args[i]);  
101.                 return printUsage();  
102.             } catch (ArrayIndexOutOfBoundsException except) {  
103.                 System.out.println("ERROR: Required parameter missing from "  
104.                         + args[i - 1]);  
105.                 return printUsage();  
106.             }  
107.         }  
108.   
109.         // Make sure there are exactly 2 parameters left.  
110.         if (other_args.size() != 2) {  
111.             System.out.println("ERROR: Wrong number of parameters: "  
112.                     + other_args.size() + " instead of 2.");  
113.             return printUsage();  
114.         }  
115.         FileInputFormat.setInputPaths(conf, other_args.get(0));  
116.         FileOutputFormat.setOutputPath(conf, new Path(other_args.get(1)));  
117.   
118.         JobClient.runJob(conf);  
119.         return 0;  
120.     }  
121.   
122.     public static void main(String[] args) throws Exception {  
123.         int res = ToolRunner.run(new Configuration(), new WordCount(), args);  
124.         System.exit(res);  
125.     }  
126.   
127. }  

 

摘自http://samwalt.iteye.com/blog/1026170