现在假设有两个数据文件
| file1.txt | file2.txt |
|---|---|
| 2018-3-1 a 2018-3-2 b 2018-3-3 c 2018-3-4 d 2018-3-5 a 2018-3-6 b 2018-3-7 c 2018-3-3 c | 2018-3-1 b 2018-3-2 a 2018-3-3 b 2018-3-4 d 2018-3-5 a 2018-3-6 c 2018-3-7 d 2018-3-3 c |
上述文件 file1.txt 本身包含重复数据，并且与 file2.txt 同样出现重复数据，现要求使用 Hadoop 大数据相关技术对这两个文件进行去重操作，并最终将结果汇总到一个文件中。
一、MapReduce 的数据去重
二、案例实现
1、Map 阶段实现
DedupMapper.java
package com.itcast.dedup;//import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;import java.io.IOException;public class DedupMapper extends Mapper<LongWritable, Text,Text, NullWritable> {//重写Ctrl+o@Overrideprotected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
// <0,2018-3-1 a> <11,2018-3-2 b>
// NullWritable.get() 方法设置空值context.write(value, NullWritable.get());}
}
该代码的作用是为了读取数据集文件将 TextInputFormat 默认组件解析的类似 <0,2018-3-1 a> 键值对修改 <2018-3-1 a,null>
2、Reduce 阶段实现
DedupReducer.java
package com.itcast.dedup;//import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;import java.io.IOException;public class DedupReducer extends Reducer<Text, NullWritable,Text,NullWritable> {@Overrideprotected void reduce(Text key, Iterable<NullWritable> values, Context context) throws IOException, InterruptedException {//<2018-3-1 a,null> <11,2018-3-2 b,null> <11,2018-3-3 c,null>context.write(key,NullWritable.get());}
}
该代码的作用仅仅是接受 Map 阶段传递来的数据，根据 Shuffle 工作原理，键值 key 相同的数据就不会被合并，因此输出数据就不会出现重复数据了。
3、Dtuver 程序主类实现
DedupDriver.java
package com.itcast.dedup;import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
//import org.apache.hadoop.io.IntWritable;
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;import java.io.IOException;public class DedupDriver {public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {//通过 Job 来封装本次 MR 的相关信息Configuration conf = new Configuration();//System.setProperty("HADOOP_USER_NAME","root");//配置 MR 运行模式，使用 local 表示本地模式，可以省略
// conf.set("mapreduce.framework.name","local");Job job = Job.getInstance(conf);//指定 MR Job jar 包运行主类job.setJarByClass(DedupDriver.class);//指定本次 MR 所有的 Mapper Reducer 类job.setMapperClass(DedupMapper.class);job.setReducerClass(DedupReducer.class);//设置业务逻辑 Mapper 类的输出 key 和 value 的数据类型job.setMapOutputKeyClass(Text.class);job.setMapOutputValueClass(NullWritable.class);//设置业务逻辑 Reducer 类的输出 key 和 value 的数据类型job.setOutputKeyClass(Text.class);job.setOutputValueClass(NullWritable.class);//使用本地模式指定处理的数据所在的位置//{input2\*} 表示读取该路径下所有的文件FileInputFormat.setInputPaths(job,"D:\\homework2\\Hadoop\\mr\\{input2\\*}");//使用本地模式指定处理完成之后的结果所保存的位置FileOutputFormat.setOutputPath(job, new Path("D:\\homework2\\Hadoop\\mr\\output"));//提交程序并且监控打印程序执行情况boolean res = job.waitForCompletion(true);//执行成功输出 0 ，不成功输出 1System.exit(res ? 0 : 1);}
}
运行结果：
三、拓展
只要日期相同，就判定为相同，最后结果输出日期即可
只需要修改DedupMapper.java文件
package com.itcast.dedup;//import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;import java.io.IOException;public class DedupMapper extends Mapper<LongWritable, Text,Text, NullWritable> {//重写Ctrl+o@Overrideprotected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {//输出日期// 把 hadoop 类型转换为 java 类型(接收传入进来的一行文本，把数据类型转换为 String 类型)String line = value.toString();// 把字符串拆分为单词String[] words = line.split(" ");// 输出前面的内容String wo = words[0];context.write(new Text(wo), NullWritable.get());}
}
运行结果：
