【原创】MapReduce编程系列之表连接

• 问题描述

        需要连接的表如下：其中左边是child，右边是parent，我们要做的是找出grandchild和grandparent的对应关系，为此需要进行表的连接。

---

Tom Lucy
Tom Jim
Lucy David
Lucy Lili
Jim Lilei
Jim SuSan
Lily Green
Lily Bians
Green Well
Green MillShell
Havid James
James LiT
Richard Cheng
Cheng LiHua

• 思路分析

诚然，在写MR程序的时候要结合MR数据处理的一些特性。例如如果我们用默认的TextInputFormat来处理传入的文件数据，传入的格式是key为行号，value为这一行的值（如上例中的第一行，key为0，value为[Tom,Lucy]），在shuffle过程中，我们的值如果有相同的key，会merge到一起（这一点很重要！）。我们利用shuffle阶段的特性，merge到一组的数据够成一组关系，然后我们在这组关系中想办法区分晚辈和长辈，最后对merge里的value一一作处理，分离出grandchild和grandparent的关系。

例如，Tom Lucy传入处理后我们将其反转，成为Lucy Tom输出。当然，输出的时候，为了达到join的效果，我们要输出两份，因为join要两个表，一个表为L1：child parent，一个表为L2：child parent，为了达到关联的目的和利用shuffle阶段的特性，我们需要将L1反转，把parent放在前面，这样L1表中的parent和L2表中的child如果字段是相同的那么在shuffle阶段就能merge到一起。还有，为了区分merge到一起后如何区分child和parent，我们把L1表中反转后的child（未来的 grandchild）字段后面加一个1，L2表中parent（未来的grandparent）字段后加2。

package com.test.join;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Iterator;

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;

public class STJoin {

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

        @Override
        protected void map(Object key, Text value, Context context)
                throws IOException, InterruptedException {
            // TODO Auto-generated method stub
            String[] rela = value.toString().trim().split(" ",2);
            if(rela.length!=2)
                return;
            String child = rela[0];
            String parent = rela[1];
            context.write(new Text(parent), new Text((child+"1")));
            context.write(new Text(child), new Text((parent+"2")));
            
        }
        
    }
    public static class STJoinReducer extends Reducer<Text, Text, Text, Text>{

        @Override
        protected void reduce(Text arg0, Iterable<Text> arg1,Context context)
                throws IOException, InterruptedException {
            // TODO Auto-generated method stub
            ArrayList<String> grandParent = new ArrayList<>();
            ArrayList<String> grandChild = new ArrayList<>();
            Iterator<Text> iterator = arg1.iterator();
            while(iterator.hasNext()){
                String text = iterator.next().toString();
                if(text.endsWith("1"))
                    grandChild.add(text.substring(0, text.length()-1));
                if(text.endsWith("2"))
                    grandParent.add(text.substring(0, text.length()-1));
            }
            
            for(String grandparent:grandParent){
                for(String grandchild:grandChild){
                    context.write(new Text(grandchild), new Text(grandparent));
                }
            }
        }
    }
    
    
    public static void main(String args[]) throws IOException, ClassNotFoundException, InterruptedException {
        Configuration conf = new Configuration();
        Job job = new Job(conf,"STJoin");
        job.setMapperClass(STJoinMapper.class);
        job.setReducerClass(STJoinReducer.class);
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(Text.class);
        FileInputFormat.addInputPath(job, new Path("hdfs://localhost:9000/user/hadoop/STJoin/joinFile"));
        FileOutputFormat.setOutputPath(job, new Path("hdfs://localhost:9000/user/hadoop/STJoin/joinResult"));
        
        System.exit(job.waitForCompletion(true)?0:1);
    }
}

• 结果显示

 

Richard    LiHua
Lily    Well
Lily    MillShell
Havid    LiT
Tom    Lilei
Tom    SuSan
Tom    Lili
Tom    David

 以上代码在hadoop1.0.3平台实现