MapReduce初级案例（1）：使用MapReduce去重

问题导读

1.map与reduce中key与value有什么不同？
2.shuffle的作用是什么？




1、数据去重
"数据去重"主要是为了掌握和利用并行化思想来对数据进行有意义的筛选。统计大数据集上的数据种类个数、从网站日志中计算访问地等这些看似庞杂的任务都会涉及数据去重。下面就进入这个实例的MapReduce程序设计。



1.1 实例描述


对数据文件中的数据进行去重。数据文件中的每行都是一个数据。
样例输入如下所示：
1）file1：

2012-3-1 a

2012-3-2 b

2012-3-3 c

2012-3-4 d

2012-3-5 a

2012-3-6 b

2012-3-7 c

2012-3-3 c
复制代码

2）file2：

2012-3-1 b

2012-3-2 a

2012-3-3 b

2012-3-4 d

2012-3-5 a

2012-3-6 c

2012-3-7 d

2012-3-3 c
复制代码

样例输出如下所示：

2012-3-1 a

2012-3-1 b

2012-3-2 a

2012-3-2 b

2012-3-3 b

2012-3-3 c

2012-3-4 d

2012-3-5 a

2012-3-6 b

2012-3-6 c

2012-3-7 c

2012-3-7 d
复制代码

2、设计思路

　数据去重的最终目标是让原始数据中出现次数超过一次的数据在输出文件中只出现一次。我们自然而然会想到将同一个数据的所有记录都交给一台reduce机器，无论这个数据出现多少次，只要在最终结果中输出一次就可以了。具体就是reduce的输入应该以数据作为key，而对value-list则没有要求。当reduce接收到一个<key，value-list>时就直接将key复制到输出的key中，并将value设置成空值。（reduce中的key表示的是我们要统计的数据例如2012-3-7 c，另外的value可以理解为一个序号，没有太大的作用，可理解为无意义数据）


　　在MapReduce流程中，map的输出<key，value>经过shuffle过程聚集成<key，value-list>后会交给reduce。
      shuffle是MapReduce的关键，也是mapreduce的难点，明白了shuffle，mapreduce就没有什么内容了。而这里的value-list则是shuffle的难点。value-list可以理解是用来标识有效数据的。但是其本身没有太大意义。




所以从设计好的reduce输入可以反推出map的输出key应为数据，value任意。继续反推，map输出数据的key为数据，而在这个实例中每个数据代表输入文件中的一行内容，所以map阶段要完成的任务就是在采用Hadoop默认的作业输入方式之后，将value设置为key，并直接输出（输出中的value任意）。map中的结果经过shuffle过程之后交给reduce。reduce阶段不会管每个key有多少个value，它直接将输入的key复制为输出的key，并输出就可以了（输出中的value被设置成空了）。


       上面是不是有点难以理解那，这里以后在进行总结，输出中的value被设置成空了，初学者还是比较困惑的。这里举个例子：
下面的1，2，3也就是被置为空的值，可能不是1，2，3....,也可能是其他值。这里只要明白key被置为空的key是什么意思就够了。

1. 2012-3-1 a

2. 2012-3-1 b

3. 2012-3-2 a

4. 2012-3-2 b

5. 2012-3-3 b

6. 2012-3-3 c

7.2012-3-4 d

8.2012-3-5 a

...............
复制代码

3、程序代码


程序代码如下所示：

package com.hebut.mr;[indent]
import java.io.IOException;[/indent][indent]
import org.apache.hadoop.conf.Configuration;

import org.apache.hadoop.fs.Path;

import org.apache.hadoop.io.IntWritable;

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;

import org.apache.hadoop.util.GenericOptionsParser;

public class Dedup {

    //map将输入中的value复制到输出数据的key上，并直接输出

    public static class Map extends Mapper<Object,Text,Text,Text>{

        private static Text line=new Text();//每行数据

        //实现map函数

        public void map(Object key,Text value,Context context)

                throws IOException,InterruptedException{

            line=value;

            context.write(line, new Text(""));[/indent]
        }
    }

   

    //reduce将输入中的key复制到输出数据的key上，并直接输出

    public static class Reduce extends Reducer<Text,Text,Text,Text>{

        //实现reduce函数

        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", "192.168.1.2:9001");

       

        String[] ioArgs=new String[]{"dedup_in","dedup_out"};

     String[] otherArgs = new GenericOptionsParser(conf, ioArgs).getRemainingArgs();

     if (otherArgs.length != 2) {

     System.err.println("Usage: Data Deduplication <in> <out>");

     System.exit(2);

     }

     

     Job job = new Job(conf, "Data Deduplication");

     job.setJarByClass(Dedup.class);

     

     //设置Map、Combine和Reduce处理类

     job.setMapperClass(Map.class);

     job.setCombinerClass(Reduce.class);

     job.setReducerClass(Reduce.class);

     

     //设置输出类型

     job.setOutputKeyClass(Text.class);

     job.setOutputValueClass(Text.class);

     

     //设置输入和输出目录

     FileInputFormat.addInputPath(job, new Path(otherArgs[0]));

     FileOutputFormat.setOutputPath(job, new Path(otherArgs[1]));

     System.exit(job.waitForCompletion(true) ? 0 : 1);

     }

}
复制代码

4 代码结果

  1）准备测试数据

     通过Eclipse下面的"DFS Locations"在"/user/hadoop"目录下创建输入文件"dedup_in"文件夹（备注："dedup_out"不需要创建。）如图1.4-1所示，已经成功创建。

         

图1.4-1 创建"dedup_in"                                   图1.4.2 上传"file*.txt"

     然后在本地建立两个txt文件，通过Eclipse上传到"/user/hadoop/dedup_in"文件夹中，两个txt文件的内容如"实例描述"那两个文件一样。如图1.4-2所示，成功上传之后。

     从SecureCRT远处查看"Master.Hadoop"的也能证实我们上传的两个文件。

    查看两个文件的内容如图1.4-3所示：

图1.4-3 文件"file*.txt"内容

2）查看运行结果

     这时我们右击Eclipse的"DFS Locations"中"/user/hadoop"文件夹进行刷新，这时会发现多出一个"dedup_out"文件夹，且里面有3个文件，然后打开双其"part-r-00000"文件，会在Eclipse中间把内容显示出来。如图1.4-4所示。

图1.4-4 运行结果

那个输入和输出是在远程机器上吗？我的怎么在本地win7下对应的工程下面啊？

好了，终于搞定了，是设置的input和output在本地的缘故，非常感谢楼主，学习了

代码中不加

//这句话很关键

        conf.set("mapred.job.tracker", "192.168.1.2:9001");

也可以运行成功，我用的是apache hadoop 2.3.0

不错的小例子

我在1.X平台上跑出来的结果怎么没有去重啊，mr就是按帖子上的写的

C:\Users\Administrator\Desktop\1.jpg

xiaohao 发表于 2014-10-21 17:52
本帖最后由 howtodown 于 2014-10-21 18:58 编辑
我在1.X平台上跑出来的结果怎么没有去重啊，mr就是按帖 ...

根据例子，仔细检查下自己的数据和程序

bioger_hit 发表于 2014-10-21 19:04
根据例子，仔细检查下自己的数据和程序

不好意思，我重新检查了一下，发现就多写了，一个字母，坑死我啦。。已经成功了，多谢指教

楼主，我怎么运行你的例子老是抛类找不到的错误啊  
14/10/30 18:57:09 INFO mapred.JobClient: Task Id : attempt_201410300042_0021_m_000001_0, Status : FAILED
java.lang.RuntimeException: java.lang.ClassNotFoundException: cheng.Dedup$Map
        at org.apache.hadoop.conf.Configuration.getClass(Configuration.java:849)
        at org.apache.hadoop.mapreduce.JobContext.getMapperClass(JobContext.java:199)
        at org.apache.hadoop.mapred.MapTask.runNewMapper(MapTask.java:719)
        at org.apache.hadoop.mapred.MapTask.run(MapTask.java:370)
        at org.apache.hadoop.mapred.Child$4.run(Child.java:255)
        at java.security.AccessController.doPrivileged(Native Method)
        at javax.security.auth.Subject.doAs(Subject.java:396)
        at org.apache.hadoop.security.UserGroupInformation.doAs(UserGroupInformation.java:1149)
        at org.apache.hadoop.mapred.Child.main(Child.java:249)