MapReduce去重

一：背景

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

程序运行的结果：