1 | import java.io.IOException; |
---|
2 | import java.util.*; |
---|
3 | |
---|
4 | import org.apache.hadoop.fs.Path; |
---|
5 | import org.apache.hadoop.conf.*; |
---|
6 | import org.apache.hadoop.io.*; |
---|
7 | import org.apache.hadoop.mapred.*; |
---|
8 | import org.apache.hadoop.util.*; |
---|
9 | |
---|
10 | public class WordCount { |
---|
11 | |
---|
12 | public static class Map extends MapReduceBase implements Mapper<LongWritable, Text, Text, IntWritable> { |
---|
13 | private final static IntWritable one = new IntWritable(1); |
---|
14 | private Text word = new Text(); |
---|
15 | |
---|
16 | public void map(LongWritable key, Text value, OutputCollector<Text, IntWritable> output, Reporter reporter) throws IOException { |
---|
17 | String line = value.toString(); |
---|
18 | StringTokenizer tokenizer = new StringTokenizer(line); |
---|
19 | while (tokenizer.hasMoreTokens()) { |
---|
20 | word.set(tokenizer.nextToken()); |
---|
21 | output.collect(word, one); |
---|
22 | } |
---|
23 | } |
---|
24 | } |
---|
25 | |
---|
26 | public static class Reduce extends MapReduceBase implements Reducer<Text, IntWritable, Text, IntWritable> { |
---|
27 | public void reduce(Text key, Iterator<IntWritable> values, OutputCollector<Text, IntWritable> output, Reporter reporter) throws IOException { |
---|
28 | int sum = 0; |
---|
29 | while (values.hasNext()) { |
---|
30 | sum += values.next().get(); |
---|
31 | } |
---|
32 | output.collect(key, new IntWritable(sum)); |
---|
33 | } |
---|
34 | } |
---|
35 | |
---|
36 | public static void main(String[] args) throws Exception { |
---|
37 | JobConf conf = new JobConf(WordCount.class); |
---|
38 | conf.setJobName("wordcount"); |
---|
39 | |
---|
40 | conf.setOutputKeyClass(Text.class); |
---|
41 | conf.setOutputValueClass(IntWritable.class); |
---|
42 | |
---|
43 | conf.setMapperClass(Map.class); |
---|
44 | conf.setCombinerClass(Reduce.class); |
---|
45 | conf.setReducerClass(Reduce.class); |
---|
46 | |
---|
47 | conf.setInputFormat(TextInputFormat.class); |
---|
48 | conf.setOutputFormat(TextOutputFormat.class); |
---|
49 | |
---|
50 | FileInputFormat.setInputPaths(conf, new Path(args[0])); |
---|
51 | FileOutputFormat.setOutputPath(conf, new Path(args[1])); |
---|
52 | |
---|
53 | JobClient.runJob(conf); |
---|
54 | } |
---|
55 | } |
---|