【原創】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平臺實現