一、例項描述

資料去重是利用並行化思想來對資料進行有意義的篩選。統計大資料集上的資料種類個數、從網站日誌中計算訪問等這些看似龐大的任務都會涉及資料去重。

比如，輸入檔案
file1.txt，其內容如下：
2017-12-9 a
2017-12-10 b
2017-12-11 c
2017-12-12 d
2017-12-13 a
2017-12-14 b
2017-12-15 c
2017-12-11 c

file2.txt，其內容如下：
2017-12-9 b
2017-12-10 a
2017-12-11 b
2017-12-12 d
2017-12-13 a
2017-12-14 c
2017-12-15 d
2017-12-11 c

對應上面給出的輸入樣例，其輸出樣例為：
2017-12-9 a
2017-12-9 b
2017-12-10 a
2017-12-10 b
2017-12-11 b
2017-12-11 c
2017-12-12 d
2017-12-13 a
2017-12-14 b
2017-12-14 c
2017-12-15 c
2017-12-15 d

二、設計思路

由於要去除重複的資料，我們可以考慮直接將一行資料作為Map和Reduce函式處理後的key值。

1. job的處理過程如圖所示
（1）Map函式設計
Map函式的實現目的：
<1, 2017-12-9 a> ——> <2017-12-9 a, “ ”>

輸入的每一行的資料都當作key，value賦空格即可，因此Map函式的設計如下：

public static class DedupCleanMapper extends Mapper<LongWritable, Text, Text, Text> {

        private static Text line = new Text();

        @Override
        protected void map(LongWritable key, Text value, Mapper<LongWritable, Text, Text, Text>.Context context)
                throws IOException, InterruptedException {
            line = value;
            context.write(line, new
 
 Text(""));
        }
    }

（2）Reduce函式設計
Reduce函式的實現目的：

由於重複的資料需要剔除，於是對於同樣的key不需進行匯聚操作，直接儲存key值即可，因此Reduce函式的設計如下：

public static class DedupCleanReducer extends Reducer<Text, Text, Text, Text> {
        @Override
        protected void reduce(Text key, Iterable<Text> values, Reducer<Text, Text, Text, Text>.Context context)
                throws IOException, InterruptedException {
            context.write(key, new Text(""));
        }
    }

三、完整程式碼

package com.walker.mrdemo;

import java.io.IOException;
import java.net.URI;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
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 DedupClean {

    /*
     * Map函式
     */
    public static class DedupCleanMapper extends Mapper<LongWritable, Text, Text, Text> {

        private static Text line = new Text();

        @Override
        protected void map(LongWritable key, Text value, Mapper<LongWritable, Text, Text, Text>.Context context)
                throws IOException, InterruptedException {
            line = value;
            context.write(line, new Text(""));
        }
    }

    /*
     * Reduce函式
     */
    public static class DedupCleanReducer extends Reducer<Text, Text, Text, Text> {
        @Override
        protected void reduce(Text key, Iterable<Text> values, Reducer<Text, Text, Text, Text>.Context context)
                throws IOException, InterruptedException {
            context.write(key, new Text(""));
        }
    }

    // 輸入輸出路徑設定
    private static final String FILE_IN_PATH = "hdfs://192.168.50.130:9000/mrdemo/DedupClean/input";
    private static final String FILE_OUT_PATH = "hdfs://192.168.50.130:9000/mrdemo/DedupClean/output";

    public static void main(String[] args) throws Exception {

        Configuration conf = new Configuration();

        // 刪除已存在的輸出目錄
        FileSystem fileSystem = FileSystem.get(new URI(FILE_OUT_PATH), conf);
        if (fileSystem.exists(new Path(FILE_OUT_PATH))) {
            fileSystem.delete(new Path(FILE_OUT_PATH), true);
        }

        Job job = Job.getInstance(conf, "DedupClean");

        job.setJarByClass(DedupClean.class);
        job.setMapperClass(DedupCleanMapper.class);
        job.setReducerClass(DedupCleanReducer.class);

        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(Text.class);

        FileInputFormat.addInputPath(job, new Path(FILE_IN_PATH));
        FileOutputFormat.setOutputPath(job, new Path(FILE_OUT_PATH));

        job.waitForCompletion(true);
    }
}