经典案例 - 数据去重

 1 package org.apache.hadoop.examples;
 2 
 3 import java.io.IOException;
 4 
 5 //导入各种Hadoop包
 6 import org.apache.hadoop.conf.Configuration;
 7 import org.apache.hadoop.fs.Path;
 8 import org.apache.hadoop.io.Text;
 9 import org.apache.hadoop.mapreduce.Job;
10 import org.apache.hadoop.mapreduce.Mapper;
11 import org.apache.hadoop.mapreduce.Reducer;
12 import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
13 import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
14 import org.apache.hadoop.util.GenericOptionsParser;
15 
16 // 主类
17 public class Dedup {
18         
19     // Mapper类
20     public static class Map extends Mapper<Object, Text, Text, Text>{
21         
22         // new一个值为空的Text对象
23         private static Text line = new Text();
24                 
25         // 实现map函数
26         public void map(Object key, Text value, Context context) throws IOException, InterruptedException {
27             
28             // 将切分后的value作为中间输出的key
29             line = value;
30             context.write(line, new Text(""));
31         }
32     }
33         
34     // Reducer类
35     public static class Reduce extends Reducer<Text,Text,Text,Text> {
36     
37         // 实现reduce函数
38         public void reduce(Text key, Iterable<Text> values, Context context) throws IOException, InterruptedException {
39                 
40             // 仅仅输出第一个Key
41             context.write(key, new Text(""));
42         }
43     }
44 
45     // 主函数
46     public static void main(String[] args) throws Exception {
47     
48         // 获取配置参数
49         Configuration conf = new Configuration();
50         String[] otherArgs = new GenericOptionsParser(conf, args).getRemainingArgs();
51                 
52         // 检查命令语法
53         if (otherArgs.length != 2) {
54             System.err.println("Usage: Dedup <in> <out>");
55             System.exit(2);
56         }
57 
58         // 定义作业对象
59         Job job = new Job(conf, "Dedup");
60         // 注册分布式类
61         job.setJarByClass(Dedup.class);
62         // 注册Mapper类
63         job.setMapperClass(Map.class);
64         // 注册合并类
65         job.setCombinerClass(Reduce.class);
66         // 注册Reducer类
67         job.setReducerClass(Reduce.class);
68         // 注册输出格式类
69         job.setOutputKeyClass(Text.class);
70         job.setOutputValueClass(Text.class);
71         // 设置输入输出路径
72         FileInputFormat.addInputPath(job, new Path(otherArgs[0]));
73         FileOutputFormat.setOutputPath(job, new Path(otherArgs[1]));
74                 
75         // 运行程序
76         System.exit(job.waitForCompletion(true) ? 0 : 1);
77     }
78 }