MapReduce 之数据去重

ゝ一纸荒年。 2022-08-22 03:54 253阅读 0赞

最近要更新一些MapReduce实例应用的系列文章。文章结构上包括以下几个部分：1.、问题背景；2、基于MapReduce的解决思路；3、代码实现；4、运行结果。

## 1. 问题背景 ##

随着数据信息量的急速增长，越来越多的人开始关注存储数据的数据去重技术。数据去重是指对所有的数据进行一个有意义的筛选，去除掉冗余的数据。例如统计大数据集上的数据种类个数、从网站日志中计算访问地等都会涉及到数据去重。下面通过一个例子具体分析。  
样例输入：  
1）file1:  
![这里写图片描述][20160511150508205]  
2）file2:  
![这里写图片描述][20160511150557753]  
要实现的输出：

## 2. 解决思路 ##

首先对于一个问题，我们要知道希望输出的文件是什么样的。基于这个想要的输出来设计reduce函数和map函数。  
对于数据去重，它希望去除掉数据中重复出现的值，也即是希望reduce的输出< key,value>中的key为期望输出的一行数据，value 为空值。那么reduce 输入的< key,list(value)>中的key应该和输出的key保存一致，而对于输入的list(value)，不管是什么，都可以在方法中将其设置为空值。  
那么在map阶段，map方法的输出key也应该是数据，value 任意值都可以。而map的输入key和value 是已知的，key为每行文本首地址相对于整个文本首地址的偏移量，value 是每行文本，因此在map方法从输入到输出的过程中，只需要将输入的value 赋给输出的key即可。  
在mapreduce整个框架下，在map输出和reduce输入之间还有一个shuffule过程，这个是系统自动完成的，但我们要知道它的作用：对每个map的输出结果组合起来并排序，它把所有相同的key的value 放在一个列表中。

## 3. 代码实现 ##

package org.apache.hadoop.examples;
    import java.io.IOException;
    import org.apache.hadoop.conf.Configuration;
    import org.apache.hadoop.fs.Path;
    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;
    
    
    public class Deduplication {
    
      public static class MyMapper 
           extends Mapper<Object, Text, Text, Text>{
    
        public void map(Object key, Text value, Context context
                        ) throws IOException, InterruptedException {
          context.write(value, new Text(""));
        }
      }
    
      public static class MyReducer 
           extends Reducer<Text,Text,Text,Text> {
    
        public void reduce(Text key, Iterable<Text> values, 
                           Context context
                           ) throws IOException, InterruptedException {      
          context.write(key, new Text(""));
        }
      }
    
      public static void main(String[] args) throws Exception {
        Configuration conf = new Configuration();
        conf.set("mapred.job.tracker", "172.16.10.15:9001");
        String input = "hdfs://172.16.10.15:9000/user/hadoop/deduplicationinput";
        String output = "hdfs://172.16.10.15:9000/user/hadoop/deduplicationoutput2";
    
        Job job = new Job(conf, "word count");
        job.setJarByClass(Deduplication.class);
        job.setMapperClass(MyMapper.class);
        job.setCombinerClass(MyReducer.class);
        job.setReducerClass(MyReducer.class);
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(Text.class);
        FileInputFormat.addInputPath(job, new Path(input));
        FileOutputFormat.setOutputPath(job, new Path(output));
        System.exit(job.waitForCompletion(true) ? 0 : 1);
      }
    }

## 4. 运行结果 ##

[20160511150508205]: /images/20220724/602c83560e804b8598fcae33abacfcc5.png
[20160511150557753]: /images/20220724/3cc521261e2f41e1abf20727b570f595.png