MapReduce多种join实现实例分析（一）

一、概述 

 
 
对于RDBMS中的join操作大伙一定非常熟悉，写sql的时候要十分注意细节，稍有差池就会耗时巨久造成很大的性能瓶颈，而在Hadoop中使用MapReduce框架进行join的操作时同样耗时，但是由于hadoop的分布式设计理念的特殊性，因此对于这种join操作同样也具备了一定的特殊性。本文主要对MapReduce框架对表之间的join操作的几种实现方式进行详细分析，并且根据我在实际开发过程中遇到的实际例子来进行进一步的说明。
 
 
二、实现原理



1、在Reudce端进行连接。
在Reudce端进行连接是MapReduce框架进行表之间join操作最为常见的模式，其具体的实现原理如下：
Map端的主要工作：为来自不同表（文件）的key/value对打标签以区别不同来源的记录。然后用连接字段作为key，其余部分和新加的标志作为value，最后进行输出。
reduce端的主要工作：在reduce端以连接字段作为key的分组已经完成，我们只需要在每一个分组当中将那些来源于不同文件的记录（在map阶段已经打标志）分开，最后进行笛卡尔只就ok了。原理非常简单，下面来看一个实例：
(1)自定义一个value返回类型:

1. package com.mr.reduceSizeJoin;   
2. import java.io.DataInput;   
3. import java.io.DataOutput;   
4. import java.io.IOException;   
5. import org.apache.hadoop.io.Text;   
6. import org.apache.hadoop.io.WritableComparable;   
7. public class CombineValues implements WritableComparable<CombineValues>{   
8.     //private static final Logger logger = LoggerFactory.getLogger(CombineValues.class); 
9.     private Text joinKey;//链接关键字 
10.     private Text flag;//文件来源标志 
11.     private Text secondPart;//除了链接键外的其他部分 
12.     public void setJoinKey(Text joinKey) {   
13.         this.joinKey = joinKey;   
14.     }   
15.     public void setFlag(Text flag) {   
16.         this.flag = flag;   
17.     }   
18.     public void setSecondPart(Text secondPart) {   
19.         this.secondPart = secondPart;   
20.     }   
21.     public Text getFlag() {   
22.         return flag;   
23.     }   
24.     public Text getSecondPart() {   
25.         return secondPart;   
26.     }   
27.     public Text getJoinKey() {   
28.         return joinKey;   
29.     }   
30.     public CombineValues() {   
31.         this.joinKey =  new Text();   
32.         this.flag = new Text();   
33.         this.secondPart = new Text();   
34.     }
35.  
36.     @Override
37.     public void write(DataOutput out) throws IOException {   
38.         this.joinKey.write(out);   
39.         this.flag.write(out);   
40.         this.secondPart.write(out);   
41.     }   
42.     @Override
43.     public void readFields(DataInput in) throws IOException {   
44.         this.joinKey.readFields(in);   
45.         this.flag.readFields(in);   
46.         this.secondPart.readFields(in);   
47.     }   
48.     @Override
49.     public int compareTo(CombineValues o) {   
50.         return this.joinKey.compareTo(o.getJoinKey());   
51.     }   
52.     @Override
53.     public String toString() {   
54.         // TODO Auto-generated method stub 
55.         return "[flag="+this.flag.toString()+",joinKey="+this.joinKey.toString()+",secondPart="+this.secondPart.toString()+"]";   
56.     }   
57. } 

 
(2) map、reduce主体代码:

1. package com.mr.reduceSizeJoin;   
2. import java.io.IOException;   
3. import java.util.ArrayList;   
4. import org.apache.hadoop.conf.Configuration;   
5. import org.apache.hadoop.conf.Configured;   
6. import org.apache.hadoop.fs.Path;   
7. import org.apache.hadoop.io.Text;   
8. import org.apache.hadoop.mapreduce.Job;   
9. import org.apache.hadoop.mapreduce.Mapper;   
10. import org.apache.hadoop.mapreduce.Reducer;   
11. import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;   
12. import org.apache.hadoop.mapreduce.lib.input.FileSplit;   
13. import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;   
14. import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;   
15. import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;   
16. import org.apache.hadoop.util.Tool;   
17. import org.apache.hadoop.util.ToolRunner;   
18. import org.slf4j.Logger;   
19. import org.slf4j.LoggerFactory;   
20. /** 
21.  * @author zengzhaozheng 
22.  * 用途说明： 
23.  * reudce side join中的left outer join 
24.  * 左连接，两个文件分别代表2个表,连接字段table1的id字段和table2的cityID字段 
25.  * table1(左表):tb_dim_city(id int,name string,orderid int,city_code,is_show) 
26.  * tb_dim_city.dat文件内容,分隔符为"|"： 
27.  * id     name  orderid  city_code  is_show 
28.  * 0       其他        9999     9999         0 
29.  * 1       长春        1        901          1 
30.  * 2       吉林        2        902          1 
31.  * 3       四平        3        903          1 
32.  * 4       松原        4        904          1 
33.  * 5       通化        5        905          1 
34.  * 6       辽源        6        906          1 
35.  * 7       白城        7        907          1 
36.  * 8       白山        8        908          1 
37.  * 9       延吉        9        909          1 
38.  * -------------------------风骚的分割线------------------------------- 
39.  * table2(右表)：tb_user_profiles(userID int,userName string,network string,double flow,cityID int) 
40.  * tb_user_profiles.dat文件内容,分隔符为"|"： 
41.  * userID   network     flow    cityID 
42.  * 1           2G       123      1 
43.  * 2           3G       333      2 
44.  * 3           3G       555      1 
45.  * 4           2G       777      3 
46.  * 5           3G       666      4 
47.  * 
48.  * -------------------------风骚的分割线------------------------------- 
49.  *  结果： 
50.  *  1   长春  1   901 1   1   2G  123 
51.  *  1   长春  1   901 1   3   3G  555 
52.  *  2   吉林  2   902 1   2   3G  333 
53.  *  3   四平  3   903 1   4   2G  777 
54.  *  4   松原  4   904 1   5   3G  666 
55.  */
56. public class ReduceSideJoin_LeftOuterJoin extends Configured implements Tool{   
57.     private static final Logger logger = LoggerFactory.getLogger(ReduceSideJoin_LeftOuterJoin.class);   
58.     public static class LeftOutJoinMapper extends Mapper<Object, Text, Text, CombineValues> {   
59.         private CombineValues combineValues = new CombineValues();   
60.         private Text flag = new Text();   
61.         private Text joinKey = new Text();   
62.         private Text secondPart = new Text();   
63.         @Override
64.         protected void map(Object key, Text value, Context context)   
65.                 throws IOException, InterruptedException {   
66.             //获得文件输入路径 
67.             String pathName = ((FileSplit) context.getInputSplit()).getPath().toString();   
68.             //数据来自tb_dim_city.dat文件,标志即为"0" 
69.             if(pathName.endsWith("tb_dim_city.dat")){   
70.                 String[] valueItems = value.toString().split("\|");   
71.                 //过滤格式错误的记录 
72.                 if(valueItems.length != 5){   
73.                     return;   
74.                 }   
75.                 flag.set("0");   
76.                 joinKey.set(valueItems[0]);   
77.                 secondPart.set(valueItems[1]+" "+valueItems[2]+" "+valueItems[3]+" "+valueItems[4]);   
78.                 combineValues.setFlag(flag);   
79.                 combineValues.setJoinKey(joinKey);   
80.                 combineValues.setSecondPart(secondPart);   
81.                 context.write(combineValues.getJoinKey(), combineValues);
82.  
83.                 }//数据来自于tb_user_profiles.dat，标志即为"1" 
84.             else if(pathName.endsWith("tb_user_profiles.dat")){   
85.                 String[] valueItems = value.toString().split("\|");   
86.                 //过滤格式错误的记录 
87.                 if(valueItems.length != 4){   
88.                     return;   
89.                 }   
90.                 flag.set("1");   
91.                 joinKey.set(valueItems[3]);   
92.                 secondPart.set(valueItems[0]+" "+valueItems[1]+" "+valueItems[2]);   
93.                 combineValues.setFlag(flag);   
94.                 combineValues.setJoinKey(joinKey);   
95.                 combineValues.setSecondPart(secondPart);   
96.                 context.write(combineValues.getJoinKey(), combineValues);   
97.             }   
98.         }   
99.     }   
100.     public static class LeftOutJoinReducer extends Reducer<Text, CombineValues, Text, Text> {   
101.         //存储一个分组中的左表信息 
102.         private ArrayList<Text> leftTable = new ArrayList<Text>();   
103.         //存储一个分组中的右表信息 
104.         private ArrayList<Text> rightTable = new ArrayList<Text>();   
105.         private Text secondPar = null;   
106.         private Text output = new Text();   
107.         /** 
108.          * 一个分组调用一次reduce函数 
109.          */
110.         @Override
111.         protected void reduce(Text key, Iterable<CombineValues> value, Context context)   
112.                 throws IOException, InterruptedException {   
113.             leftTable.clear();   
114.             rightTable.clear();   
115.             /** 
116.              * 将分组中的元素按照文件分别进行存放 
117.              * 这种方法要注意的问题： 
118.              * 如果一个分组内的元素太多的话，可能会导致在reduce阶段出现OOM， 
119.              * 在处理分布式问题之前最好先了解数据的分布情况，根据不同的分布采取最 
120.              * 适当的处理方法，这样可以有效的防止导致OOM和数据过度倾斜问题。 
121.              */
122.             for(CombineValues cv : value){   
123.                 secondPar = new Text(cv.getSecondPart().toString());   
124.                 //左表tb_dim_city 
125.                 if("0".equals(cv.getFlag().toString().trim())){   
126.                     leftTable.add(secondPar);   
127.                 }   
128.                 //右表tb_user_profiles 
129.                 else if("1".equals(cv.getFlag().toString().trim())){   
130.                     rightTable.add(secondPar);   
131.                 }   
132.             }   
133.             logger.info("tb_dim_city:"+leftTable.toString());   
134.             logger.info("tb_user_profiles:"+rightTable.toString());   
135.             for(Text leftPart : leftTable){   
136.                 for(Text rightPart : rightTable){   
137.                     output.set(leftPart+ " " + rightPart);   
138.                     context.write(key, output);   
139.                 }   
140.             }   
141.         }   
142.     }   
143.     @Override
144.     public int run(String[] args) throws Exception {   
145.           Configuration conf=getConf(); //获得配置文件对象 
146.             Job job=new Job(conf,"LeftOutJoinMR");   
147.             job.setJarByClass(ReduceSideJoin_LeftOuterJoin.class);
148.             FileInputFormat.addInputPath(job, new Path(args[0])); //设置map输入文件路径 
149.             FileOutputFormat.setOutputPath(job, new Path(args[1])); //设置reduce输出文件路径
150.             job.setMapperClass(LeftOutJoinMapper.class);   
151.             job.setReducerClass(LeftOutJoinReducer.class);
152.             job.setInputFormatClass(TextInputFormat.class); //设置文件输入格式 
153.             job.setOutputFormatClass(TextOutputFormat.class);//使用默认的output格格式
154.  
155.             //设置map的输出key和value类型 
156.             job.setMapOutputKeyClass(Text.class);   
157.             job.setMapOutputValueClass(CombineValues.class);
158.  
159.             //设置reduce的输出key和value类型 
160.             job.setOutputKeyClass(Text.class);   
161.             job.setOutputValueClass(Text.class);   
162.             job.waitForCompletion(true);   
163.             return job.isSuccessful()?0:1;   
164.     }   
165.     public static void main(String[] args) throws IOException,   
166.             ClassNotFoundException, InterruptedException {   
167.         try {   
168.             int returnCode =  ToolRunner.run(new ReduceSideJoin_LeftOuterJoin(),args);   
169.             System.exit(returnCode);   
170.         } catch (Exception e) {   
171.             // TODO Auto-generated catch block 
172.             logger.error(e.getMessage());   
173.         }   
174.     }   
175. } 

其中具体的分析以及数据的输出输入请看代码中的注释已经写得比较清楚了，这里主要分析一下reduce join的一些不足。之所以会存在reduce join这种方式，我们可以很明显的看出原：因为整体数据被分割了，每个map task只处理一部分数据而不能够获取到所有需要的join字段，因此我们需要在讲join key作为reduce端的分组将所有join key相同的记录集中起来进行处理，所以reduce join这种方式就出现了。这种方式的缺点很明显就是会造成map和reduce端也就是shuffle阶段出现大量的数据传输，效率很低。

相关阅读：
《Hadoop伪分布式搭建操作步骤指南》；
《HADOOP的本地库（NATIVE LIBRARIES）简介》；
《基于Hadoop大数据分析应用场景与项目实战演练》