-
Notifications
You must be signed in to change notification settings - Fork 0
/
Copy pathTask2.java
87 lines (64 loc) · 2.42 KB
/
Task2.java
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
package bigdata;
import java.io.IOException;
import java.util.StringTokenizer;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
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.Tool;
import org.apache.hadoop.util.ToolRunner;
public class Task2 extends Configured implements Tool{
public static void main(String args[]) throws Exception {
ToolRunner.run(new Task2(), args);
}
public int run(String[] args) throws Exception {
String input = args[0];
String output = args[1];
Job job = Job.getInstance(getConf());
job.setJarByClass(Task2.class);
job.setMapperClass(T2Map.class);
job.setReducerClass(T2Reduce.class);
job.setMapOutputKeyClass(IntWritable.class);
job.setMapOutputValueClass(IntWritable.class);
job.setOutputKeyClass(IntWritable.class);
job.setOutputValueClass(IntWritable.class);
FileInputFormat.addInputPath(job, new Path(input));
FileOutputFormat.setOutputPath(job, new Path(output));
job.waitForCompletion(true);
return 0;
}
public static class T2Map extends Mapper<Object, Text, IntWritable, IntWritable>{
IntWritable ou = new IntWritable();
IntWritable ov = new IntWritable();
@Override
protected void map(Object key, Text value,
Mapper<Object, Text, IntWritable, IntWritable>.Context context)
throws IOException, InterruptedException {
StringTokenizer st = new StringTokenizer(value.toString());
ou.set(Integer.parseInt(st.nextToken()));
ov.set(Integer.parseInt(st.nextToken()));
IntWritable one = new IntWritable(1);
context.write(ou, one);
context.write(ov, one);
}
}
public static class T2Reduce extends Reducer<IntWritable, IntWritable, IntWritable, IntWritable>{
IntWritable count = new IntWritable();
@Override
protected void reduce(IntWritable key, Iterable<IntWritable> values,
Reducer<IntWritable, IntWritable, IntWritable, IntWritable>.Context context)
throws IOException, InterruptedException {
Integer sum = 0;
for(IntWritable value : values) {
sum += value.get();
}
count.set(sum);
if(count.get() > 0) context.write(key, count);
}
}
}