MapReduce 之单表关联设计思路与源码分析

比眉伴天荒 2022-07-24 05:09 140阅读 0赞

本文通过一个简单的样例child-parent 表来介绍单表关联，在介绍这部分内容的时候主要包括两部分：1、描述清楚每一部分设计的思路；2、基于Mapreduce 进行了实现。

## 1. 样例child-parent表的内容和期望的输出 ##

我们得到一个child-parent 表，然后希望能够通过挖掘到两列之间的关系，得到一个新的表 grandchild-grandparent 。

为了方便说明问题，首先给出一个简单的家庭人物关系图。

![这里写图片描述][20160515100455578]

我们已知的表是表示孩子 和父母之间的对应关系，然后求出孩子和爷爷奶奶(姥姥姥爷)的对应关系。因此我们的输入如下图所示。

输入：

![这里写图片描述][20160515101023674]

这里输入包括两列，第一列为child列，第二列为parent列。  
然后我们通过连接的方式找到孩子父母的父母。所以它的输出结果如下：

![这里写图片描述][20160515101628442]

## 2. 设计思路 ##

现在对这个实例进行分析，已知给出child-parent 表，要想得到grandchild-grandparent表，那么就需要进行单表连接。即对左表的child列 和 右表的parent 列进行连接，且左表 和 右表 是同一个表。

### 2.1 Map过程的设计 ###

*  对读入的一行文本进行解析，得到两部分内容： child 和parent 。
 *  因为要实现单表关联，实际上就是两张表的连接，那我们希望Map过程的输出是两张表的内容，因此Map的输出应该是两类：  
    
    
     *  第一类输出的< key,value>中child作为key,parent 作为value
     *  第二类输出的< key,value>要把parent 作为key ,child 作为value。
     *  这样在map过程处理完所有行之后，它的输出既输出了左表child-parent,又输出了右表parent-child。
     *  另外 ，还要对这两类进行区别，比如第一类的输出的value里加入”@”,第二类输出的value里加入’&’。

经过Map过程之后，会得到如下结果：

![这里写图片描述][20160515103017011]

Map结果输出之后，MapReduce框架自动实现Shuffle 过程，结果如下：

![这里写图片描述][20160515103420345]

从Shuffle结果可以看出，输出的key为家庭各个成员，输出的value 为每个成员的父母或者子女，并且可以通过字符@或者&进行区分。  
那么我们可以得出，若一个成员的value列表中仅含有父母或者仅含有子女，则不符合我们的输出；当同时包含父母和子女时，才符合我们的输出要求。

### 2.2 Reduce过程的设计 ###

*  建立两个数组，一个数组用于存放父母，一个数组用于存放子女。
 *  对values列表进行分析，通过每个value的第一个字符是@还是&来区分父母或子女。  
    
    
     *  若是父母，则将其存入父母数组中
     *  若是子女，则将其存入子女数组中
 *  对values列表分析完毕后，首先进行一个判断：父母数组或者子女数组是否 为空（有一个为空即不符合输出要求）。当均不为空时，将子女数组的值逐个作为key,将父母数组的值逐个作为value,就得到了grandchild–grandparent 这样 的输出结果了。

## 3. 源码分析 ##

现在看一看完整的mapreduce 实现单表连接的代码。

package org.apache.hadoop.examples;
    import java.io.IOException;
    import java.util.StringTokenizer;
    import org.apache.hadoop.conf.Configuration;
    import org.apache.hadoop.fs.Path;
    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 SingleLink {
           
    
        /*Map过程*/
        public static class lxnmapper extends Mapper<Object,Text,Text,Text>{
                
            public void map(Object key,Text value,Context context)
                throws IOException,InterruptedException{
                StringTokenizer itr = new StringTokenizer(value.toString());//对value进行解析
                String child = itr.nextToken();
                String parent = itr.nextToken();
                context.write(new Text(child), new Text("@"+parent));//第一类输出
                context.write(new Text(parent), new Text("&"+child));//第二类输出
            }       
        }
    
        /*Reduce过程*/
        public static class lxnreduce extends Reducer<Text,Text,Text,Text>{
             
            public void reduce(Text key,Iterable<Text> values,Context context)
                throws IOException,InterruptedException{
                String[] childs = new String[10];//建立两个数组
                String[] parents = new String[10];
                int childnum = 0;//初始数组个数
                int parentnum = 0;
                //对values列表的值逐个分析，根据字符判断是父母或子女，放入对应的数组里
                for(Text val :values){
                    String vals = val.toString();
                    if(vals.substring(0,1).equals("@")){
                        parents[parentnum]=vals.substring(1);
                        parentnum++;
                    }
                    else if(vals.substring(0,1).equals("&")){
                        childs[childnum]=vals.substring(1);
                        childnum++;
                    }               
                }
                //if条件是筛选掉成员只有父母或只有子女的情况
                if(parentnum!=0 && childnum!=0){
                    for(int i=0;i<childnum;i++){
                        for(int j=0;j<parentnum;j++){
                            context.write(new Text(childs[i]), new Text(parents[j]));
                        }
                    }   
                }                   
            }
        }
    
        public static void main(String[] args) throws Exception{
            Configuration conf = new Configuration();
            conf.set("mapred.job.tracker", "172.16.10.15:9001");
            String pathIn1 = "hdfs://172.16.10.15:9000/user/hadoop/singlelinkinput";
            String pathOut="hdfs://172.16.10.15:9000/user/hadoop/singlelinkoutput";
    
            Job job = new Job(conf,"single link");
            job.setJarByClass(SingleLink.class);
            job.setMapperClass(lxnmapper.class);        
            job.setReducerClass(lxnreduce.class);
            job.setOutputKeyClass(Text.class);
            job.setOutputValueClass(Text.class);
            FileInputFormat.addInputPath(job, new Path(pathIn1));
            FileOutputFormat.setOutputPath(job, new Path(pathOut));
            System.exit(job.waitForCompletion(true)?0:1);
        }
    }

输出结果呈现在上面第一部分的期望输出中。

## 参考文献 ##

1.  MapReduce实现单表关联 [http://blog.csdn.net/jtlyuan/article/details/7937428][http_blog.csdn.net_jtlyuan_article_details_7937428]
2.  Hadoop集群（第9期）\_MapReduce初级案例 [http://www.cnblogs.com/xia520pi/archive/2012/06/04/2534533.html][http_www.cnblogs.com_xia520pi_archive_2012_06_04_2534533.html]

[20160515100455578]: /images/20220722/bedd4c343881463aa869bd2521c1638b.png
[20160515101023674]: /images/20220722/989b6d5c5bab4c98808676f2fcb41f87.png
[20160515101628442]: /images/20220722/b4e96544af7e4c909e1686d2504fa938.png
[20160515103017011]: /images/20220722/2ecb06918b944987b24a754f14d99da7.png
[20160515103420345]: /images/20220722/75b12d28da6547dbab4fa3bc41110baa.png
[http_blog.csdn.net_jtlyuan_article_details_7937428]: http://blog.csdn.net/jtlyuan/article/details/7937428
[http_www.cnblogs.com_xia520pi_archive_2012_06_04_2534533.html]: http://www.cnblogs.com/xia520pi/archive/2012/06/04/2534533.html