Hadoop_MapReduce_数据去重示例

比眉伴天荒 2023-03-13 12:19 21阅读 0赞

**Hadoop\_MapReduce\_数据去重**

示例：删除输入文件中的重复数据，重复的数据只保留一个

输入文件1：D:\\data\\distinct\\file1.txt

zhangsan 500 450 jan
    lisi 200 150 jan
    jerry 200 150 feb
    amy 200 150 feb

输入文件2：D:\\data\\distinct\\file2.txt

zhangsan 500 450 jan
    lisi 200 150 jan
    jack 150 160 jan
    tom 500 500 feb
    jerry 200 150 feb
    zhangsan 500 450 jan
    lisi 200 150 jan

1.  DistinctMapper

package com.blu.distinct;
    
    import java.io.IOException;
    
    import org.apache.hadoop.io.LongWritable;
    import org.apache.hadoop.io.NullWritable;
    import org.apache.hadoop.io.Text;
    import org.apache.hadoop.mapreduce.Mapper;
    
    public class DistinctMapper extends Mapper<LongWritable, Text, Text, NullWritable> { 
    	@Override
    	protected void map(LongWritable key, Text value, Mapper<LongWritable, Text, Text, NullWritable>.Context context)
    			throws IOException, InterruptedException { 
    		context.write(value, NullWritable.get());
    	}
    }

1.  DistinctReducer

package com.blu.distinct;
    
    import java.io.IOException;
    
    import org.apache.hadoop.io.NullWritable;
    import org.apache.hadoop.io.Text;
    import org.apache.hadoop.mapreduce.Reducer;
    
    public class DistinctReducer extends Reducer<Text, NullWritable, Text, NullWritable>{ 
    	@Override
    	protected void reduce(Text key, Iterable<NullWritable> value,
    			Reducer<Text, NullWritable, Text, NullWritable>.Context context) throws IOException, InterruptedException { 
    		context.write(key, NullWritable.get());
    	}
    }

1.  DistinctJob

package com.blu.distinct;
    
    import org.apache.hadoop.fs.Path;
    import org.apache.hadoop.io.NullWritable;
    import org.apache.hadoop.io.Text;
    import org.apache.hadoop.mapreduce.Job;
    import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
    import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
    
    public class DistinctJob { 
    
    	public static void main(String[] args) throws Exception { 
    		Job job = Job.getInstance();
    		job.setJarByClass(DistinctJob.class);
    		job.setMapperClass(DistinctMapper.class);
    		job.setReducerClass(DistinctReducer.class);
    		job.setMapOutputKeyClass(Text.class);
    		job.setMapOutputValueClass(NullWritable.class);
    		job.setOutputKeyClass(Text.class);
    		job.setOutputValueClass(NullWritable.class);
    		FileInputFormat.addInputPath(job, new Path(args[0]));
    		FileOutputFormat.setOutputPath(job, new Path(args[1]));
    		boolean flag = job.waitForCompletion(true);
    		System.exit(flag ? 0 : 1);
    
    	}
    
    }

1.  运行参数：

D:\data\distinct\ D:\data\output

1.  运行结果：

amy 200 150 feb
    jack 150 160 jan
    jerry 200 150 feb
    lisi 200 150 jan
    tom 500 500 feb
    zhangsan 500 450 jan