007_hadoop中MapReduce应用案例_1_数据去重

stbyruby

　　前面介绍了复杂的MapReduce Job流在实际中的应用方法：006_hadoop中MapReduce详解_3
　　这节主要是通过实例来分析MapReduce在实际中的应用，从中得到一些启发，在项目开发中，设计MapReduce往往是比较复杂的。我们先通过简单的实例入手后面慢慢加深。
　　先简单说一下后面还会出现什么实例吧：
　　1.数据排序-->partition
　　2.找隔代关系-->单表关联
　　3.学生课程-->多表关联
　　4.好友推荐
　　5.PageRank
　　6.倒排序索引
　　7.最优路径
　　基本上通过上面的7+1+1=9个实例，我们应该可以基本掌握MapReduce的设计过程。
　　好了，废话不多说，开始我们这个简单的数据去重的例子
　　实例描述：有以下两个文件，文件中表示某天，某IP访问了我们的系统这样一个日志。我们当时间和IP相同时，我们将这种相同的数据去掉，只留下一个。文件如下：

2014-10-3 10.3.5.19
2014-10-3 10.3.5.19
2014-10-3 10.3.5.18
2014-10-3 10.3.51.19
2014-10-3 10.3.02.19

2014-10-3 10.3.5.19
2014-10-4 10.3.5.19
2014-10-3 10.3.5.18
2014-10-5 10.3.51.19
2014-10-5 10.3.02.19
　　 实例分析：
　　数据去重，我们知道在Map到Reduce阶段，相同的key会合并在一起。既然如此，我们把整个文件的一行当作Map输出的Key。这样在Map输出进入Reduce事，所有重复的key不就自动合并了吗。这不就达到了我们去重的目标了
　　实例代码：

package com.sxt.mrtest;
import java.io.IOException;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
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;

/**
*
* @title Uniq
* @description 去掉重复行
* @author hadoop
* @version
* @copyright (c) SINOSOFT
*
*/
public class Uniq {
public static class UniqMap extends Mapper<LongWritable, Text, Text, Text>{
@Override
public void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
context.write(value, new Text(""));
}
}
public static class UniqReduce extends Reducer<Text, Text, Text, Text>{
@Override
public void reduce(Text arg0, Iterable<Text> arg1, Context arg2) throws IOException, InterruptedException {
arg2.write(arg0, new Text(""));
}
}
public static void main(String[] args) throws Exception {
Configuration conf = new Configuration();
//配置作业1
Job job = new Job(conf, "Uniq");
job.setJarByClass(Uniq.class);
job.setMapperClass(UniqMap.class);
job.setCombinerClass(UniqReduce.class);
job.setReducerClass(UniqReduce.class);
job.setOutputKeyClass(Text.class);
job.setOutputValueClass(Text.class);
FileInputFormat.addInputPath(job, new Path("/user/helloMR/demo1_datadel")); //Map的输入
FileOutputFormat.setOutputPath(job, new Path("/user/helloMR/success"));//Reduce的输出
System.exit(job.waitForCompletion(true) ? 0 : 1);
}
}

　　实例数据流程图：
　　

 
　　查看JobTracker状态：http://masterIP:50030/jobtracker.jsp
　　查看NameNode状态：http://masterIP:50070/dfshealth.jsp