MapReduce实现‘多表关联’

多表关联和单表关联相似，都类似于数据库中的自然连接。相比单表关联，多表关联的左右表和连接列更加清楚。所以可以采用和单表关联的相同的处理方式，map识别出输入的行属于哪个表之后，对其进行分割，将连接的列值保存在key中，另一列和左右表标识保存在value中，然后输出。reduce拿到连接结果之后，解析value内容，根据标志将左右表内容分开存放，然后求笛卡尔积，最后直接输出。

---

输入是两个文件，一个代表工厂表，包含工厂名列和地址编号列；

image.png

另一个代表地址表，包含地址编号列和地址名列。

image.png
期望输出： image.png

完整代码：

package mr;

import java.io.IOException;
import java.net.URI;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
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;   

public class MyAddress {
    
    
    
    static class MyAddressMapper  extends  Mapper<LongWritable, Text, Text, Text>{  
        
         public void map(LongWritable k1, Text v1, Context context) 
                         throws java.io.IOException, java.lang.InterruptedException
         {
            String[]  lines= v1.toString().split("\t");
            if(lines[0].equals("factoryname") || lines[0].equals("addressID")) return;
            String word1=lines[0];
            String word2=lines[1];
            
            if(word1.charAt(0)>='0'&&word1.charAt(0)<='9'){
                context.write(new Text(word1), new Text("1"+","+word1+","+word2));
            }
            else if(word2.charAt(0)>='0'&&word2.charAt(0)<='9'){
                context.write(new Text(word2), new Text("2"+","+word1+","+word2));
            }
            else return;
            
        System.out.println("map......"+word1+","+word2);
         }
        
    }
    
    static class  MyAddressReduce extends Reducer<Text, Text, Text, Text>{
        
        protected void setup(Context context) 
                throws java.io.IOException, java.lang.InterruptedException{
            context.write(new Text("factory\t"),new Text("address"));
        }
        
         public void reduce(Text key, Iterable<Text> values, Context context) throws java.io.IOException, java.lang.InterruptedException
         {
             List<String> fname=new ArrayList();
             List<String> aname=new ArrayList();
             
             Iterator<Text>  it=values.iterator();
             while(it.hasNext()){
                String lines=it.next().toString();
                String[] words=lines.split(",");
                if(words[0].equals("1")){
                    aname.add(words[2]);
                }
                else if(words[0].equals("2")){
                    fname.add(words[1]);
                }
                else return;
             }
             for(String fn:fname){
                 for(String an:aname){
                     context.write(new Text(fn+"\t"), new Text(an));
                 }
             }
                 
             
             System.out.println("reduce......");
         }
            
    }

    private static String INPUT_PATH="hdfs://master:9000/input/fname.txt";
    private static String INPUT_PATH2="hdfs://master:9000/input/aname.txt";
    private static String OUTPUT_PATH="hdfs://master:9000/output/MyAddressResult/";

    public static void main(String[] args) throws Exception {   
        
        Configuration  conf=new Configuration();
        FileSystem  fs=FileSystem.get(new URI(OUTPUT_PATH),conf);
     
        if(fs.exists(new Path(OUTPUT_PATH)))
                fs.delete(new Path(OUTPUT_PATH));
        
        Job  job=new Job(conf,"myjob");
        
        job.setJarByClass(MyAddress.class);
        job.setMapperClass(MyAddressMapper.class);
        job.setReducerClass(MyAddressReduce.class);
         
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(Text.class);
        
         
        
        FileInputFormat.addInputPath(job,new Path(INPUT_PATH));
        FileInputFormat.addInputPath(job,new Path(INPUT_PATH2));
        FileOutputFormat.setOutputPath(job, new Path(OUTPUT_PATH));
        
        job.waitForCompletion(true);

    }

}

代码理解参照《MapReduce实现‘单表关联’》