"单表关联"要求从给出的数据中寻找所关心的数据，它是对原始数据所包含信息的挖掘。

实例描述 给出child-parent（孩子——父母）表，要求输出grandchild-grandparent（孙子——爷奶）表。

输入　　　　　　　　　　　　　　　　　　　　输出：

　　　　　　　　　　　　　　

 

问题分析：

　　这个实例需要进行单表连接，连接的是左表的parent列和右表的child列，且左表和右表是同一个表。连接结果中除去连接的两列就是所需要的结果——"grandchild--grandparent"表。要用MapReduce解决这个实例，首先应该考虑如何实现表的自连接；其次就是连接列的设置；最后是结果的整理 MapReduce的shuffle过程会将相同的key会连接在一起，所以可以将map结果的key设置成待连接的列，然后列中相同的值就自然会连接在一起了。

实现步骤：

1.map阶段将读入数据分割成child和parent之后，将parent设置成key，child设置成value进行输出，并作为左表；再将同一对child和parent中的child设置成key，parent设置成value进行输出，作为右表。

2.为了区分输出中的左右表，需要在输出的value中再加上左右表的信息，比如在value的String最开始处加上字符1表示左表，加上字符2表示右表。

3. reduce接收到连接的结果，其中每个key的value-list就包含了"grandchild--grandparent"关系。取出每个key的value-list进行解析，将左表中的child放入一个数组，右表中的parent放入一个数组，然后对两个数组求笛卡尔积就是最后的结果了。

代码：

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.output.FileOutputFormat;

public class connection {

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();

protected void map(Object key,Object value,Context context) throws IOException,InterruptedException{

String[] tokens = value.toString().split(",");

output_key.set(tokens[0]);

output_value.set(2+","+value);

context.write(output_key, output_value);

output_key.set(tokens[1]);

output_value.set(1+","+value);

context.write(output_key,output_value);

}

}

static class MyReduce extends Reducer<Text,Text,Text,Text> {

Text output_key=new Text();

Text output_value=new Text();

protected void reduce(Text key, Iterable<Text> values,Context context) throws IOException,InterruptedException{

List<String> childs= new ArrayList<String>();

List<String> grands= new ArrayList<String>();

for(Text line:values){

String[] tokens=line.toString().split(",");

if(tokens[0].equals("1")){

childs.add(tokens[1]); //tom

//System.out.println(1+"=="+tokens[1]);

}

else if(tokens[0].equals("2")){

grands.add(tokens[2]);

// System.out.println(2+"=="+tokens[2]);

}

}

for(String c:childs)

for(String g:grands){

output_key.set(c);

output_value.set(g);

context.write(output_key, output_value);

}

}

}

public static void main(String[] args) throws Exception{

Path outputpath = new Path(OUTPUT_PATH);

Configuration conf = new Configuration();

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);

job.setReducerClass(MyReduce.class);

// job.setMapOutputKeyClass(MyNewKey.class);

// job.setMapOutputValueClass(NullWritable.class);

job.setOutputKeyClass(Text.class);

job.setOutputValueClass(Text.class);

job.waitForCompletion(true);

}

}

最后就可以得出结论。