如何使用Hadoop的Reduce Side Join

上帝大脸

我们都知道在数据库里，多个表之间是可以根据某个链接键进行join的，这也是数据库的范式规范，通过主外键的关联，由此来减少数据冗余，提高性能。当然近几年，随着NOSQL的兴起，出现了基于列的的列式数据库，典型的有Hbase，MongonDB，Cassdran，等等，NOSQL数据库弱化了关联，直接将一整条数据，存入一列，以及去掉了数据库的部分事务特性，从而在海量数据面前显得游刃有余，当然，大部分的NOSQL不支持join操作，也没有绝对的必要支持，因为现在，我们完全是把一整条数据存在了一起，虽然多了许多冗余，但也换来了比较高检索性能，扩展性能，可靠性能。但某些业务场景下，我们仍然需要Join操作，这时候怎么办？
如果数据量比较大的情况下，我们可以使用Hadoop的MapReduce来完成大表join，尤其对Hbase的某些表进行join操作，当然我们也可以使用Hive或Pig来完成，其实质在后台还是运行的一个MR程序。
那么，散仙今天就来看下如何使用MapReduce来完成一个join操作，Hadoop的join分为很多种例如；Reduce链接，Map侧链接，半链接和Reduce侧链接+BloomFilter等等，各个链接都有自己特定的应用场景，没有绝对的谁好谁坏。

今天散仙要说的是，基于Reduce侧的链接，原理如下：
1、在Reudce端进行连接。
   在Reudce端进行连接是MapReduce框架进行表之间join操作最为常见的模式，其具体的实现原理如下：
Map端的主要工作：为来自不同表（文件）的key/value对打标签以区别不同来源的记录。然后用连接字段作为key，其余部分和新加的标志作为value，最后进行输出。
reduce端的主要工作：在reduce端以连接字段作为key的分组已经完成，我们只需要在每一个分组当中将那些来源于不同文件的记录（在map阶段已经打标志）分开，最后进行笛卡尔只就ok了。
本次的实现是基于hadoop的旧版API+contribu扩展包里的，DataJoin的工具类辅助来完成的，下篇博客，将会给出，基于新版API，独立来完成Reduce侧的连接示例。

现在看下散仙的两个文件的测试数据，一个是a.txt，另一个是b.txt

a文件的数据
1,三劫散仙,13575468248
2,凤舞九天,18965235874
3,忙忙碌碌,15986854789
4,少林寺方丈,15698745862



b文件的数据
3,A,99,2013-03-05
1,B,89,2013-02-05
2,C,69,2013-03-09
3,D,56,2013-06-07

源码如下：
package com.qin.reducejoin;

import java.io.DataInput;  
import java.io.DataOutput;  
import java.io.IOException;  
import java.util.Iterator;  
import org.apache.hadoop.conf.Configuration;  
import org.apache.hadoop.conf.Configured;  
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;  
import org.apache.hadoop.io.Text;  
import org.apache.hadoop.io.Writable;  
import org.apache.hadoop.mapred.FileInputFormat;  
import org.apache.hadoop.mapred.FileOutputFormat;  
import org.apache.hadoop.mapred.JobClient;  
import org.apache.hadoop.mapred.JobConf;  
import org.apache.hadoop.mapred.KeyValueTextInputFormat;  
import org.apache.hadoop.mapred.MapReduceBase;  
import org.apache.hadoop.mapred.Mapper;  
import org.apache.hadoop.mapred.OutputCollector;  
import org.apache.hadoop.mapred.Reducer;  
import org.apache.hadoop.mapred.Reporter;  
import org.apache.hadoop.mapred.TextInputFormat;  
import org.apache.hadoop.mapred.TextOutputFormat;  
import org.apache.hadoop.util.ReflectionUtils;
import org.apache.hadoop.util.Tool;  
import org.apache.hadoop.util.ToolRunner;  
import org.apache.hadoop.contrib.utils.join.DataJoinMapperBase;  
import org.apache.hadoop.contrib.utils.join.DataJoinReducerBase;  
import org.apache.hadoop.contrib.utils.join.TaggedMapOutput;  
import com.qin.joinreduceerror.JoinReduce;

/***
*
* Hadoop1.2的版本，旧版本实现的Reduce侧连接
*
* @author qindongliang
*
*    大数据交流群：376932160
*  搜索技术交流群：324714439
*
*
*/
public class DataJoin extends Configured implements Tool {  

/**
*
* Map实现
*
* */
public static class MapClass extends DataJoinMapperBase {  
/**
* 读取输入的文件路径
*
* **/
protected Text generateInputTag(String inputFile) {  
//返回文件路径，做标记
return new Text(inputFile);  
}  

/***
* 分组的Key
*
* **/
protected Text generateGroupKey(TaggedMapOutput aRecord) {  
String line = ((Text) aRecord.getData()).toString();  
String[] tokens = line.split(",");  
String groupKey = tokens[0];  
return new Text(groupKey);  
}  


protected TaggedMapOutput generateTaggedMapOutput(Object value) {  
TaggedWritable retv = new TaggedWritable((Text) value);  
retv.setTag(this.inputTag);  
return retv;  
}  
}  
/**
*
* Reduce进行笛卡尔积
*
* **/
public static class Reduce extends DataJoinReducerBase {  

/***
* 笛卡尔积
*
* */
protected TaggedMapOutput combine(Object[] tags, Object[] values) {  
if (tags.length < 2) return null;   
String joinedStr = "";   
for (int i=0; i<values.length; i++) {  
if (i > 0) {joinedStr += ",";}  
TaggedWritable tw = (TaggedWritable) values;  
String line = ((Text) tw.getData()).toString();  
String[] tokens = line.split(",", 2);  
joinedStr += tokens[1];  
}  
TaggedWritable retv = new TaggedWritable(new Text(joinedStr));  
retv.setTag((Text) tags[0]);   
return retv;  
}  
}  
/**
*
* 自定义的输出类型
*
* ***/
public static class TaggedWritable extends TaggedMapOutput {  
private Writable data;  
/**
* 注意加上构造方法
*
* */
public TaggedWritable() {
// TODO Auto-generated constructor stub
}
public TaggedWritable(Writable data) {  
this.tag = new Text("");  
this.data = data;  
}  
public Writable getData() {  
return data;  
}  
public void write(DataOutput out) throws IOException {  
this.tag.write(out);  
//此行代码很重要
out.writeUTF(this.data.getClass().getName());
this.data.write(out);
}  
public void readFields(DataInput in) throws IOException {  
this.tag.readFields(in);  
//加入此部分代码，否则，可能报空指针异常
String temp=in.readUTF();
if (this.data == null|| !this.data.getClass().getName().equals(temp)) {
try {
this.data = (Writable) ReflectionUtils.newInstance(
Class.forName(temp), null);
} catch (ClassNotFoundException e) {
e.printStackTrace();
}
}
this.data.readFields(in);  
}  
}  
public int run(String[] args) throws Exception {  
Configuration conf = getConf();  
JobConf job = new JobConf(conf, DataJoin.class);  
job.set("mapred.job.tracker","192.168.75.130:9001");
////读取person中的数据字段
job.setJar("tt.jar");
job.setJarByClass(DataJoin.class);
System.out.println("模式：  "+job.get("mapred.job.tracker"));;


String path="hdfs://192.168.75.130:9000/root/outputjoindb";
FileSystem fs=FileSystem.get(conf);
Path p=new Path(path);
if(fs.exists(p)){
fs.delete(p, true);
System.out.println("输出路径存在，已删除！");
}

Path in = new Path("hdfs://192.168.75.130:9000/root/inputjoindb");  
//  Path out = new Path(args[1]);  
FileInputFormat.setInputPaths(job, in);  
FileOutputFormat.setOutputPath(job, p);  
job.setJobName("cee");  
job.setMapperClass(MapClass.class);  
job.setReducerClass(Reduce.class);  
job.setInputFormat(TextInputFormat.class);  
job.setOutputFormat(TextOutputFormat.class);  
job.setOutputKeyClass(Text.class);  
job.setOutputValueClass(TaggedWritable.class);  
job.set("mapred.textoutputformat.separator", ",");  
JobClient.runJob(job);   
return 0;  
}  
public static void main(String[] args) throws Exception {   

int res = ToolRunner.run(new Configuration(),  
new DataJoin(),  
args);  
System.exit(res);  
}  
}  

运行，日志
模式：  192.168.75.130:9001
输出路径存在，已删除！
INFO - NativeCodeLoader.<clinit>(43) | Loaded the native-hadoop library
WARN - LoadSnappy.<clinit>(46) | Snappy native library not loaded
INFO - FileInputFormat.listStatus(199) | Total input paths to process : 2
INFO - JobClient.monitorAndPrintJob(1380) | Running job: job_201404222310_0025
INFO - JobClient.monitorAndPrintJob(1393) |  map 0% reduce 0%
INFO - JobClient.monitorAndPrintJob(1393) |  map 33% reduce 0%
INFO - JobClient.monitorAndPrintJob(1393) |  map 100% reduce 0%
INFO - JobClient.monitorAndPrintJob(1393) |  map 100% reduce 33%
INFO - JobClient.monitorAndPrintJob(1393) |  map 100% reduce 100%
INFO - JobClient.monitorAndPrintJob(1448) | Job complete: job_201404222310_0025
INFO - Counters.log(585) | Counters: 30
INFO - Counters.log(587) |   Job Counters
INFO - Counters.log(589) |     Launched reduce tasks=1
INFO - Counters.log(589) |     SLOTS_MILLIS_MAPS=14335
INFO - Counters.log(589) |     Total time spent by all reduces waiting after reserving slots (ms)=0
INFO - Counters.log(589) |     Total time spent by all maps waiting after reserving slots (ms)=0
INFO - Counters.log(589) |     Launched map tasks=3
INFO - Counters.log(589) |     Data-local map tasks=3
INFO - Counters.log(589) |     SLOTS_MILLIS_REDUCES=9868
INFO - Counters.log(587) |   File Input Format Counters
INFO - Counters.log(589) |     Bytes Read=207
INFO - Counters.log(587) |   File Output Format Counters
INFO - Counters.log(589) |     Bytes Written=172
INFO - Counters.log(587) |   FileSystemCounters
INFO - Counters.log(589) |     FILE_BYTES_READ=837
INFO - Counters.log(589) |     HDFS_BYTES_READ=513
INFO - Counters.log(589) |     FILE_BYTES_WRITTEN=221032
INFO - Counters.log(589) |     HDFS_BYTES_WRITTEN=172
INFO - Counters.log(587) |   Map-Reduce Framework
INFO - Counters.log(589) |     Map output materialized bytes=849
INFO - Counters.log(589) |     Map input records=8
INFO - Counters.log(589) |     Reduce shuffle bytes=849
INFO - Counters.log(589) |     Spilled Records=16
INFO - Counters.log(589) |     Map output bytes=815
INFO - Counters.log(589) |     Total committed heap usage (bytes)=496644096
INFO - Counters.log(589) |     CPU time spent (ms)=2080
INFO - Counters.log(589) |     Map input bytes=187
INFO - Counters.log(589) |     SPLIT_RAW_BYTES=306
INFO - Counters.log(589) |     Combine input records=0
INFO - Counters.log(589) |     Reduce input records=8
INFO - Counters.log(589) |     Reduce input groups=4
INFO - Counters.log(589) |     Combine output records=0
INFO - Counters.log(589) |     Physical memory (bytes) snapshot=623570944
INFO - Counters.log(589) |     Reduce output records=4
INFO - Counters.log(589) |     Virtual memory (bytes) snapshot=2908262400
INFO - Counters.log(589) |     Map output records=8

运行结果，如下图所示：

可以看出，MR正确的完成了join操作，需要注意的是Reduce侧连接的不足之处：
之所以会存在reduce join这种方式，我们可以很明显的看出原：因为整体数据被分割了，每个map task只处理一部分数据而不能够获取到所有需要的join字段，因此我们需要在讲join key作为reduce端的分组将所有join key相同的记录集中起来进行处理，所以reduce join这种方式就出现了。这种方式的缺点很明显就是会造成map和reduce端也就是shuffle阶段出现大量的数据传输，效率很低。
另外一点需要注意的是，散仙在eclipse里进行调试，Local模式下会报异常，建议提交到hadoop的测试集群上进行测试。