我修改了Azure的WordCount範例，以TreeMap做排序
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
| package org.apache.hadoop.examples;import java.io.IOException;import java.util.StringTokenizer;import org.apache.hadoop.conf.Configuration;import org.apache.hadoop.fs.Path;import org.apache.hadoop.io.IntWritable;import org.apache.hadoop.io.NullWritable;import org.apache.hadoop.io.Text;import org.apache.hadoop.mapreduce.Job;import org.apache.hadoop.mapreduce.Mapper;import org.apache.hadoop.mapreduce.Reducer;import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;import org.apache.hadoop.util.GenericOptionsParser;import java.util.Map;import java.util.TreeMap;public class WordCount { public static final int k = 10; public static class TokenizerMapper extends Mapper< Object, Text, NullWritable, Text>{ private final static IntWritable one = new IntWritable(1); private Text word = new Text(); private TreeMap< Long, Text> top = new TreeMap< Long, Text>(); public void map(Object key, Text value, Context context) throws IOException, InterruptedException { StringTokenizer itr = new StringTokenizer(value.toString()); while (itr.hasMoreTokens()) { top.put(Long.parseLong(itr.nextToken()), new Text(value)); if (top.size()>k) top.remove(top.firstKey()); } } @Override protected void cleanup(Context context) throws IOException, InterruptedException { for (Text t : top.values()) { context.write(NullWritable.get(), t); } } } public static class IntSumReducer extends Reducer< NullWritable, Text, NullWritable, Text> { private IntWritable result = new IntWritable(); private TreeMap< Long, Text> top = new TreeMap< Long, Text>(); public void reduce(NullWritable key, Iterable< Text> values, Context context) throws IOException, InterruptedException { int sum = 0; for (Text val : values) { String v[] = val.toString().split("\t"); Long w = Long.parseLong(v[1]); top.put(w, val); if (top.size()>k) top.remove(top.firstKey()); } for (Text t : top.descendingMap().values()) { context.write(NullWritable.get(), t); } } } public static void main(String[] args) throws Exception { Configuration conf = new Configuration(); String[] otherArgs = new GenericOptionsParser(conf, args).getRemainingArgs(); if (otherArgs.length != 2) { System.err.println("Usage: wordcount < in> < out>"); System.exit(2); } Job job = new Job(conf, "word count"); job.setJarByClass(WordCount.class); job.setMapperClass(TokenizerMapper.class); job.setReducerClass(IntSumReducer.class); job.setNumReduceTasks(1); job.setOutputKeyClass(NullWritable.class); job.setOutputValueClass(Text.class); FileInputFormat.addInputPath(job, new Path(otherArgs[0])); FileOutputFormat.setOutputPath(job, new Path(otherArgs[1])); System.exit(job.waitForCompletion(true) ? 0 : 1); }} |
