Hadoop: The Definitive Guide【PDF版】
输入数据（第一列child，第二列 parent）
Tom Lucy
Tom Jack
Jone Lucy
Jone Jack
Lucy Mary
Lucy Ben
Jack Alice
Jack Jesse
Terry Alice
Terry Jesse
Philip Terry
Philip Alma
Mark Terry
Mark Alma
输出数据(第一列grandchild,第二列grandparents)
Tom Jesse
Tom Alice
Jone Jesse
Jone Alice
Jone Ben
Jone Mary
Tom Ben
Tom Mary
Philip Alice
Philip Jesse
Mark Alice
Mark Jesse
不知到是"Hadoop 实战" 还是那个虾皮工作室写的解法，把这个问题说成是单表联接。个人觉得其实没有这么深奥，说白了非常的简单，map输入：如上所说，key是child's name, value是parent's name； map的输出：key是名字，value是“标识+名字”，其中标识'0'后面接他的child's name，'1'接他的parent's name。比如第一个记录Tom Lucy，那么key是"Tom", value是"1Lucy"，还有key是lucy，value是"0Tom" 。 reducer的输入：对于每一个人名(key)，它的一系列values是他的children或者parents name(通过标识来识别是children还是parents)，这样子得到key的children集合和parents集合（说白了就是知道了key的所有children和parents的名字）。然后求这两个集合的乘积（或者称为笛卡尔积）即可。。
具体代码如下：
import java.util.*;
import java.io.*;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapred.TextInputFormat;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
import org.apache.hadoop.mapreduce.lib.input.KeyValueTextInputFormat;
public class oneTableConnect {
public static class tableMapper extends Mapper<Text, Text, Text, Text>
{
@Override
public void map(Text key, Text value, Context context) throws IOException, InterruptedException
{
context.write(key, new Text( "1" + value.toString()));
context.write(value, new Text("0" + key.toString()));
}
}
public static class tableReducer extends Reducer<Text, Text, Text, Text>
{
@Override
public void reduce(Text key, Iterable<Text> values, Context context) throws IOException, InterruptedException
{
String[] gChildren = new String[10];
String[] gParents = new String[10];
int cNum = 0;
int pNum = 0;
for(Text val : values)
{
if(val.toString().charAt(0) == '0')// the key's child.
{
gChildren[cNum++] = val.toString().substring(1);
}
else//the key's parent.
{
gParents[pNum++] = val.toString().substring(1);
}
}
for(int i=0; i<cNum; i++)
for(int j=0;j<pNum;j++)
{
context.write(new Text(gChildren[i]), new Text(gParents[j]));
}
}
}
public static void main(String[] args) throws Exception
{
Configuration conf = new Configuration();
conf.set("mapreduce.input.keyvaluelinerecordreader.key.value.separator", " ");
Job job = new Job(conf, "tableConnect");
job.setJarByClass(oneTableConnect.class);
job.setMapperClass(tableMapper.class);
job.setReducerClass(tableReducer.class);
job.setOutputKeyClass(Text.class);
job.setOutputValueClass(Text.class);
job.setInputFormatClass(KeyValueTextInputFormat.class);
job.setOutputFormatClass(TextOutputFormat.class);
FileInputFormat.addInputPath(job, new Path(args[0]));
FileOutputFormat.setOutputPath(job, new Path(args[1]));
System.exit( job.waitForCompletion(true) ? 0 : 1);
}
}