多表关联

多表关联和单表关联类似，它也是通过对原始数据进行一定的处理，从其中挖掘出关心的信息。

实例描述 输入是两个文件，一个代表工厂表，包含工厂名列和地址编号列；另一个代表地址表，包含地址名列和地址编号列。要求从输入数据中找出工厂名和地址名的对应关系，输出"工厂名——地址名"表：

样例输入如下所示：

工厂表：　　　　　　　　　　　　　　　　　　　　地址表：　　　　　　　　　　　　　　　　    输出： 

　　　　　　　　　　　　　　

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

代码：

import java.io.IOException;
import java.util.ArrayList;
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.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.input.FileSplit;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

public class DoubleTble { //word count
static String INPUT_PATH="hdfs://master:9000/qq";
static String OUTPUT_PATH="hdfs://master:9000/output";
static class MyMapper extends Mapper<Object,Object,Text,Text>{
Text output_key=new Text();
Text output_value=new Text();
String tableName="";
protected void setup(Context context) throws java.io.IOException, java.lang.InterruptedException{
FileSplit fs= (FileSplit) context.getInputSplit();
tableName=fs.getPath().getName();
System.out.println(tableName);
}
protected void map(Object key, Object value, Context context) throws IOException, InterruptedException{
//
String[] tokens=value.toString().split(","); //f name id
// a 1 name
if(tokens!=null && tokens.length==2){
//tom 2,tom,lucy lucy 1,tom,lucy
if(tableName.equals("123")){
output_key.set(tokens[1]);
output_value.set(1+","+tokens[0]+","+tokens[1]); // 1 1,Beijing Red Star,1
}
else if(tableName.equals("a")){
output_key.set(tokens[0]);
output_value.set(2+","+tokens[0]+","+tokens[1]);
}
context.write(output_key,output_value);
System.out.println(tokens[0]+" - "+tokens[1]);
}
}
}
static class MyReduce extends Reducer<Text,Text,Text,Text>{
Text output_key=new Text();
Text output_value=new Text();
//k=1 v={1,bj Beijing Red Star,1 }
protected void reduce(Text key, Iterable<Text> values, Context context)
throws IOException, InterruptedException{
List<String> factorys=new ArrayList();
List<String> addrs=new ArrayList();
for(Text line:values){
String[] tokens=line.toString().split(",");
if(tokens[0].equals("1")){
factorys.add(tokens[1]); //tom
System.out.println(1+"=="+tokens[1]);
}
else if(tokens[0].equals("2")){
addrs.add(tokens[2]); // 2=tNo,1=id,bj
System.out.println(2+"=="+tokens[2]);
}
}
for(String c:factorys)
for(String g:addrs){
output_key.set(c);
output_value.set(g);
context.write(output_key, output_value);
}
//key=hi values={1,1,1}--->hi 3,
}
}

public static void main(String[] args) throws Exception{
Path outputpath=new Path(OUTPUT_PATH);
Configuration conf=new Configuration();
// conf.set("fs.default.name", "hdfs://master:9000");
FileSystem fs=outputpath.getFileSystem(conf);
if(fs.exists(outputpath)){
fs.delete(outputpath, true);
}
Job job=Job.getInstance(conf);
FileInputFormat.setInputPaths(job, INPUT_PATH);
FileOutputFormat.setOutputPath(job, outputpath);
job.setMapperClass(MyMapper.class); //map
job.setReducerClass(MyReduce.class); //reduce
// job.setCombinerClass(MyReduce.class);
job.setOutputKeyClass(Text.class);
job.setOutputValueClass(Text.class);
// job.setNumReduceTasks(0);
job.waitForCompletion(true);

}

}

这样结果就可以得出。