MapReduce去重

Posted 2020-06-12

一：背景

很多数据源中的数据都是含有大量重复的，为此我们需要将重复的数据去掉，这也称为数据的清洗，MapReduce从Map端到Reduce端的Shuffle过程天生就有去重的功能，但是这是对输出的Key作为参照进行去重的。所以我们可以将Map端读入Value作为Key输出，就可以很方便的实现去重了。

 

二：技术实现

#需求 有两个文件file0和file1。将两个文件中的内容合并去重。

#file0的内容如下：

 

[java] view plain copy

 

1. 1  
2. 1  
3. 2  
4. 2  
5. 3  
6. 3  
7. 4  
8. 4  
9. 5  
10. 5  
11. 6  
12. 6  
13. 7  
14. 8  
15. 9  

file1的内容如下：

 

 

[java] view plain copy

 

1. 1  
2. 9  
3. 9  
4. 8  
5. 8  
6. 7  
7. 7  
8. 6  
9. 6  
10. 5  
11. 5  
12. 4  
13. 4  
14. 2  
15. 1  
16. 2  

代码实现：

 

 

[java] view plain copy

 

1. public class DistinctTest {  
2.         // 定义输入路径  
3.         private static final String INPUT_PATH = "hdfs://liaozhongmin:9000/distinct_file/*";  
4.         // 定义输出路径  
5.         private static final String OUT_PATH = "hdfs://liaozhongmin:9000/out";  
6.   
7.         public static void main(String[] args) {  
8.   
9.             try {  
10.                 // 创建配置信息  
11.                 Configuration conf = new Configuration();  
12.                   
13.   
14.                 // 创建文件系统  
15.                 FileSystem fileSystem = FileSystem.get(new URI(OUT_PATH), conf);  
16.                 // 如果输出目录存在，我们就删除  
17.                 if (fileSystem.exists(new Path(OUT_PATH))) {  
18.                     fileSystem.delete(new Path(OUT_PATH), true);  
19.                 }  
20.   
21.                 // 创建任务  
22.                 Job job = new Job(conf, DistinctTest.class.getName());  
23.   
24.                 //1.1   设置输入目录和设置输入数据格式化的类  
25.                 FileInputFormat.setInputPaths(job, INPUT_PATH);  
26.                 job.setInputFormatClass(TextInputFormat.class);  
27.   
28.                 //1.2   设置自定义Mapper类和设置map函数输出数据的key和value的类型  
29.                 job.setMapperClass(DistinctMapper.class);  
30.                 job.setMapOutputKeyClass(Text.class);  
31.                 job.setMapOutputValueClass(Text.class);  
32.   
33.                 //1.3   设置分区和reduce数量(reduce的数量，和分区的数量对应，因为分区为一个，所以reduce的数量也是一个)  
34.                 job.setPartitionerClass(HashPartitioner.class);  
35.                 job.setNumReduceTasks(1);  
36.   
37.                 //1.4   排序  
38.                 //1.5   归约  
39.                 job.setCombinerClass(DistinctReducer.class);  
40.                 //2.1   Shuffle把数据从Map端拷贝到Reduce端。  
41.                 //2.2   指定Reducer类和输出key和value的类型  
42.                 job.setReducerClass(DistinctReducer.class);  
43.                 job.setOutputKeyClass(Text.class);  
44.                 job.setOutputValueClass(Text.class);  
45.   
46.                 //2.3   指定输出的路径和设置输出的格式化类  
47.                 FileOutputFormat.setOutputPath(job, new Path(OUT_PATH));  
48.                 job.setOutputFormatClass(TextOutputFormat.class);  
49.   
50.   
51.                 // 提交作业 退出  
52.                 System.exit(job.waitForCompletion(true) ? 0 : 1);  
53.               
54.             } catch (Exception e) {  
55.                 e.printStackTrace();  
56.             }  
57.         }  
58.   
59.     public static class DistinctMapper extends Mapper<LongWritable, Text, Text, Text>{  
60.         //定义写出去的key和value  
61.         private Text outKey = new Text();  
62.         private Text outValue = new Text("");  
63.         @Override  
64.         protected void map(LongWritable key, Text value, Mapper<LongWritable, Text, Text, Text>.Context context) throws IOException, InterruptedException {  
65.             //把输入的key作为value输出(因为)  
66.             outKey = value;  
67.               
68.             //把结果写出去  
69.             context.write(outKey, outValue);  
70.         }  
71.     }  
72.       
73.     public static class DistinctReducer extends Reducer<Text, Text, Text, Text>{  
74.         @Override  
75.         protected void reduce(Text key, Iterable<Text> value, Reducer<Text, Text, Text, Text>.Context context) throws IOException, InterruptedException {  
76.             //直接把key写出去  
77.             context.write(key, new Text(""));  
78.         }  
79.     }  
80. }  

程序运行的结果：