snippet-code/TopQuery.java at master · intergret/snippet-code · GitHub

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
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
importjava.io.IOException;
importjava.util.Random;
importorg.apache.hadoop.conf.Configuration;
importorg.apache.hadoop.fs.FileSystem;
importorg.apache.hadoop.fs.Path;
importorg.apache.hadoop.io.IntWritable;
importorg.apache.hadoop.io.Text;
importorg.apache.hadoop.io.WritableComparable;
importorg.apache.hadoop.mapreduce.Job;
importorg.apache.hadoop.mapreduce.Mapper;
importorg.apache.hadoop.mapreduce.Reducer;
importorg.apache.hadoop.mapreduce.lib.input.FileInputFormat;
importorg.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
importorg.apache.hadoop.util.GenericOptionsParser;


publicclassTopQuery{

publicstaticclassCountMapperextendsMapper<Object, Text, Text, IntWritable>{

privatefinalstaticIntWritableone = newIntWritable(1);
privateTextquery = newText();

publicvoidmap(Objectkey, Textvalue, Contextcontext) throwsIOException, InterruptedException {

Stringeachline=value.toString();
String [] eachterm=eachline.split("#");
//每行以#分隔的第二个字段为具体的query字面值
query.set(eachterm[1]);
context.write(query, one);
 }
	}


publicstaticclassCountReducerextendsReducer<Text,IntWritable,Text,IntWritable> {

privateIntWritabletotal = newIntWritable();

publicvoidreduce(Textkey, Iterable<IntWritable> values, Contextcontext) throwsIOException, InterruptedException {

intsum = 0;
for (IntWritableval : values) {
sum += val.get();
 }
total.set(sum);
context.write(key,total);
 }
	}


publicstaticclassSortMapperextendsMapper<Object, Text, IntWritable,Text>{

IntWritabletimes = newIntWritable();
Textquery = newText();

publicvoidmap(Objectkey, Textvalue, Contextcontext) throwsIOException, InterruptedException {
Stringeachline=value.toString();
String[] eachterm =eachline.split("	");

query.set(eachterm[0]);
times.set(Integer.parseInt(eachterm[1]));
context.write(times,query);
 }
	}


publicstaticclassSortReducerextendsReducer<IntWritable,Text,IntWritable,Text> {

privateTextquery = newText();

publicvoidreduce(IntWritablekey,Iterable<Text> values, Contextcontext) throwsIOException, InterruptedException {
//不同的query可能出现相同的次数
for (Textval : values) {
query.set(val);
context.write(key,query);
 }
 }
	}


privatestaticclassIntDecreasingComparatorextendsIntWritable.Comparator {
publicintcompare(WritableComparablea, WritableComparableb) {
return -super.compare(a, b);
 }

publicintcompare(byte[] b1, ints1, intl1, byte[] b2, ints2, intl2) {
return -super.compare(b1, s1, l1, b2, s2, l2);
 }
	}


publicstaticvoidmain(String[] args) throwsException {

Configurationconf = newConfiguration();
String[] otherArgs = newGenericOptionsParser(conf, args).getRemainingArgs();
if (otherArgs.length != 2) {
System.err.println("Usage: query <in> <out>");
System.exit(2);
 }

Jobjob = newJob(conf, "query");
job.setJarByClass(TopQuery.class);
job.setMapperClass(CountMapper.class);
job.setCombinerClass(CountReducer.class);
job.setReducerClass(CountReducer.class);

job.setOutputKeyClass(Text.class);
job.setOutputValueClass(IntWritable.class);

//定义一个临时目录，先将query词频统计任务的输出结果写到临时目录中, 下一个排序任务以临时目录为输入目录。
FileInputFormat.addInputPath(job, newPath(otherArgs[0]));
PathtempDir = newPath("query-temp-" + Integer.toString(newRandom().nextInt(Integer.MAX_VALUE)));
FileOutputFormat.setOutputPath(job, tempDir);


if(job.waitForCompletion(true))
 {
JobsortJob = newJob(conf, "querysort");
sortJob.setJarByClass(TopQuery.class);

FileInputFormat.addInputPath(sortJob, tempDir);

sortJob.setMapperClass(SortMapper.class);
FileOutputFormat.setOutputPath(sortJob, newPath(otherArgs[1]));

sortJob.setOutputKeyClass(IntWritable.class);
sortJob.setOutputValueClass(Text.class);

//Map、Reduce端的排序都以query频数降序为准
sortJob.setSortComparatorClass(IntDecreasingComparator.class);

FileSystem.get(conf).deleteOnExit(tempDir);

System.exit(sortJob.waitForCompletion(true) ? 0 : 1);
 }
System.exit(job.waitForCompletion(true) ? 0 : 1);
	}

}