图文精华

About云-梭伦科技»专题 › 技术学习(版主发帖区) › 大数据学习 › Mapreduce › hadoop mapreduce join代码示例及讲解
分享

hadoop mapreduce join代码示例及讲解

desehawk 发表于 2015-3-19 11:36:55 [显示全部楼层] 回帖奖励 阅读模式 关闭右栏 5 30646
本帖最后由 desehawk 于 2015-3-19 11:48 编辑
问题导读

1.在Reudce端进行连接，reduce端的主要工作有哪些？
2.在Map端进行连接，使用场景是什么？
3.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]+"\t"+valueItems[2]+"\t"+valueItems[3]+"\t"+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]+"\t"+valueItems[1]+"\t"+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+ "\t" + 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阶段出现大量的数据传输，效率很低。
2、在Map端进行连接。
使用场景：一张表十分小、一张表很大。
用法:在提交作业的时候先将小表文件放到该作业的DistributedCache中，然后从DistributeCache中取出该小表进行join key / value解释分割放到内存中（可以放大Hash Map等等容器中）。然后扫描大表，看大表中的每条记录的join key /value值是否能够在内存中找到相同join key的记录，如果有则直接输出结果。
直接上代码，比较简单：
  1. package com.mr.mapSideJoin;   

  2. import java.io.BufferedReader;   

  3. import java.io.FileReader;   

  4. import java.io.IOException;   

  5. import java.util.HashMap;   

  6. import org.apache.hadoop.conf.Configuration;   

  7. import org.apache.hadoop.conf.Configured;   

  8. import org.apache.hadoop.filecache.DistributedCache;   

  9. import org.apache.hadoop.fs.Path;   

  10. import org.apache.hadoop.io.Text;   

  11. import org.apache.hadoop.mapreduce.Job;   

  12. import org.apache.hadoop.mapreduce.Mapper;   

  13. import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;   

  14. import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;   

  15. import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;   

  16. import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;   

  17. import org.apache.hadoop.util.Tool;   

  18. import org.apache.hadoop.util.ToolRunner;   

  19. import org.slf4j.Logger;   

  20. import org.slf4j.LoggerFactory;   

  21. /**   

  22.  * @author zengzhaozheng   

  23.  *   

  24.  * 用途说明：   

  25.  * Map side join中的left outer join   

  26.  * 左连接，两个文件分别代表2个表,连接字段table1的id字段和table2的cityID字段   

  27.  * table1(左表):tb_dim_city(id int,name string,orderid int,city_code,is_show)，   

  28.  * 假设tb_dim_city文件记录数很少，tb_dim_city.dat文件内容,分隔符为"|"：   

  29.  * id     name  orderid  city_code  is_show   

  30.  * 0       其他        9999     9999         0   

  31.  * 1       长春        1        901          1   

  32.  * 2       吉林        2        902          1   

  33.  * 3       四平        3        903          1   

  34.  * 4       松原        4        904          1   

  35.  * 5       通化        5        905          1   

  36.  * 6       辽源        6        906          1   

  37.  * 7       白城        7        907          1   

  38.  * 8       白山        8        908          1   

  39.  * 9       延吉        9        909          1   

  40.  * -------------------------风骚的分割线-------------------------------   

  41.  * table2(右表)：tb_user_profiles(userID int,userName string,network string,double flow,cityID int)   

  42.  * tb_user_profiles.dat文件内容,分隔符为"|"：   

  43.  * userID   network     flow    cityID   

  44.  * 1           2G       123      1   

  45.  * 2           3G       333      2   

  46.  * 3           3G       555      1   

  47.  * 4           2G       777      3   

  48.  * 5           3G       666      4   

  49.  * -------------------------风骚的分割线-------------------------------   

  50.  *  结果：   

  51.  *  1   长春  1   901 1   1   2G  123   

  52.  *  1   长春  1   901 1   3   3G  555   

  53.  *  2   吉林  2   902 1   2   3G  333   

  54.  *  3   四平  3   903 1   4   2G  777   

  55.  *  4   松原  4   904 1   5   3G  666   

  56.  */ 

  57. public class MapSideJoinMain extends Configured implements Tool{   

  58.     private static final Logger logger = LoggerFactory.getLogger(MapSideJoinMain.class);   

  59.     public static class LeftOutJoinMapper extends Mapper<Object, Text, Text, Text> {

  60.  

  61.         private HashMap<String,String> city_info = new HashMap<String, String>();   

  62.         private Text outPutKey = new Text();   

  63.         private Text outPutValue = new Text();   

  64.         private String mapInputStr = null;   

  65.         private String mapInputSpit[] = null;   

  66.         private String city_secondPart = null;   

  67.         /**   

  68.          * 此方法在每个task开始之前执行，这里主要用作从DistributedCache   

  69.          * 中取到tb_dim_city文件，并将里边记录取出放到内存中。   

  70.          */ 

  71.         @Override 

  72.         protected void setup(Context context)   

  73.                 throws IOException, InterruptedException {   

  74.             BufferedReader br = null;   

  75.             //获得当前作业的DistributedCache相关文件   

  76.             Path[] distributePaths = DistributedCache.getLocalCacheFiles(context.getConfiguration());   

  77.             String cityInfo = null;   

  78.             for(Path p : distributePaths){   

  79.                 if(p.toString().endsWith("tb_dim_city.dat")){   

  80.                     //读缓存文件，并放到mem中   

  81.                     br = new BufferedReader(new FileReader(p.toString()));   

  82.                     while(null!=(cityInfo=br.readLine())){   

  83.                         String[] cityPart = cityInfo.split("\\|",5);   

  84.                         if(cityPart.length ==5){   

  85.                             city_info.put(cityPart[0], cityPart[1]+"\t"+cityPart[2]+"\t"+cityPart[3]+"\t"+cityPart[4]);   

  86.                         }   

  87.                     }   

  88.                 }   

  89.             }   

  90.         }

  91.  

  92.         /**   

  93.          * Map端的实现相当简单，直接判断tb_user_profiles.dat中的   

  94.          * cityID是否存在我的map中就ok了，这样就可以实现Map Join了   

  95.          */ 

  96.         @Override 

  97.         protected void map(Object key, Text value, Context context)   

  98.                 throws IOException, InterruptedException {   

  99.             //排掉空行   

  100.             if(value == null || value.toString().equals("")){   

  101.                 return;   

  102.             }   

  103.             mapInputStr = value.toString();   

  104.             mapInputSpit = mapInputStr.split("\\|",4);   

  105.             //过滤非法记录   

  106.             if(mapInputSpit.length != 4){   

  107.                 return;   

  108.             }   

  109.             //判断链接字段是否在map中存在   

  110.             city_secondPart = city_info.get(mapInputSpit[3]);   

  111.             if(city_secondPart != null){   

  112.                 this.outPutKey.set(mapInputSpit[3]);   

  113.                 this.outPutValue.set(city_secondPart+"\t"+mapInputSpit[0]+"\t"+mapInputSpit[1]+"\t"+mapInputSpit[2]);   

  114.                 context.write(outPutKey, outPutValue);   

  115.             }   

  116.         }   

  117.     }   

  118.     @Override 

  119.     public int run(String[] args) throws Exception {   

  120.             Configuration conf=getConf(); //获得配置文件对象   

  121.             DistributedCache.addCacheFile(new Path(args[1]).toUri(), conf);//为该job添加缓存文件   

  122.             Job job=new Job(conf,"MapJoinMR");   

  123.             job.setNumReduceTasks(0);

  124.  

  125.             FileInputFormat.addInputPath(job, new Path(args[0])); //设置map输入文件路径   

  126.             FileOutputFormat.setOutputPath(job, new Path(args[2])); //设置reduce输出文件路径

  127.  

  128.             job.setJarByClass(MapSideJoinMain.class);   

  129.             job.setMapperClass(LeftOutJoinMapper.class);

  130.  

  131.             job.setInputFormatClass(TextInputFormat.class); //设置文件输入格式   

  132.             job.setOutputFormatClass(TextOutputFormat.class);//使用默认的output格式

  133.  

  134.             //设置map的输出key和value类型   

  135.             job.setMapOutputKeyClass(Text.class);

  136.  

  137.             //设置reduce的输出key和value类型   

  138.             job.setOutputKeyClass(Text.class);   

  139.             job.setOutputValueClass(Text.class);   

  140.             job.waitForCompletion(true);   

  141.             return job.isSuccessful()?0:1;   

  142.     }   

  143.     public static void main(String[] args) throws IOException,   

  144.             ClassNotFoundException, InterruptedException {   

  145.         try {   

  146.             int returnCode =  ToolRunner.run(new MapSideJoinMain(),args);   

  147.             System.exit(returnCode);   

  148.         } catch (Exception e) {   

  149.             // TODO Auto-generated catch block   

  150.             logger.error(e.getMessage());   

  151.         }   

  152.     }   

  153. } 
复制代码
这里说说DistributedCache。DistributedCache是分布式缓存的一种实现，它在整个MapReduce框架中起着相当重要的作用，他可以支撑我们写一些相当复杂高效的分布式程序。说回到这里，JobTracker在作业启动之前会获取到DistributedCache的资源uri列表，并将对应的文件分发到各个涉及到该作业的任务的TaskTracker上。另外，关于DistributedCache和作业的关系，比如权限、存储路径区分、public和private等属性，接下来有用再整理研究一下写一篇blog，这里就不详细说了。
另外还有一种比较变态的Map Join方式，就是结合HBase来做Map Join操作。这种方式完全可以突破内存的控制，使你毫无忌惮的使用Map Join，而且效率也非常不错。
3、SemiJoin。
SemiJoin就是所谓的半连接，其实仔细一看就是reduce join的一个变种，就是在map端过滤掉一些数据，在网络中只传输参与连接的数据不参与连接的数据不必在网络中进行传输，从而减少了shuffle的网络传输量，使整体效率得到提高，其他思想和reduce join是一模一样的。说得更加接地气一点就是将小表中参与join的key单独抽出来通过DistributedCach分发到相关节点，然后将其取出放到内存中（可以放到HashSet中），在map阶段扫描连接表，将join key不在内存HashSet中的记录过滤掉，让那些参与join的记录通过shuffle传输到reduce端进行join操作，其他的和reduce join都是一样的。看代码：
  1. package com.mr.SemiJoin;   

  2. import java.io.BufferedReader;   

  3. import java.io.FileReader;   

  4. import java.io.IOException;   

  5. import java.util.ArrayList;   

  6. import java.util.HashSet;   

  7. import org.apache.hadoop.conf.Configuration;   

  8. import org.apache.hadoop.conf.Configured;   

  9. import org.apache.hadoop.filecache.DistributedCache;   

  10. import org.apache.hadoop.fs.Path;   

  11. import org.apache.hadoop.io.Text;   

  12. import org.apache.hadoop.mapreduce.Job;   

  13. import org.apache.hadoop.mapreduce.Mapper;   

  14. import org.apache.hadoop.mapreduce.Reducer;   

  15. import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;   

  16. import org.apache.hadoop.mapreduce.lib.input.FileSplit;   

  17. import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;   

  18. import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;   

  19. import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;   

  20. import org.apache.hadoop.util.Tool;   

  21. import org.apache.hadoop.util.ToolRunner;   

  22. import org.slf4j.Logger;   

  23. import org.slf4j.LoggerFactory;   

  24. /**   

  25.  * @author zengzhaozheng   

  26.  *   

  27.  * 用途说明：   

  28.  * reudce side join中的left outer join   

  29.  * 左连接，两个文件分别代表2个表,连接字段table1的id字段和table2的cityID字段   

  30.  * table1(左表):tb_dim_city(id int,name string,orderid int,city_code,is_show)   

  31.  * tb_dim_city.dat文件内容,分隔符为"|"：   

  32.  * id     name  orderid  city_code  is_show   

  33.  * 0       其他        9999     9999         0   

  34.  * 1       长春        1        901          1   

  35.  * 2       吉林        2        902          1   

  36.  * 3       四平        3        903          1   

  37.  * 4       松原        4        904          1   

  38.  * 5       通化        5        905          1   

  39.  * 6       辽源        6        906          1   

  40.  * 7       白城        7        907          1   

  41.  * 8       白山        8        908          1   

  42.  * 9       延吉        9        909          1   

  43.  * -------------------------风骚的分割线-------------------------------   

  44.  * table2(右表)：tb_user_profiles(userID int,userName string,network string,double flow,cityID int)   

  45.  * tb_user_profiles.dat文件内容,分隔符为"|"：   

  46.  * userID   network     flow    cityID   

  47.  * 1           2G       123      1   

  48.  * 2           3G       333      2   

  49.  * 3           3G       555      1   

  50.  * 4           2G       777      3   

  51.  * 5           3G       666      4   

  52.  * -------------------------风骚的分割线-------------------------------   

  53.  * joinKey.dat内容：   

  54.  * city_code   

  55.  * 1   

  56.  * 2   

  57.  * 3   

  58.  * 4   

  59.  * -------------------------风骚的分割线-------------------------------   

  60.  *  结果：   

  61.  *  1   长春  1   901 1   1   2G  123   

  62.  *  1   长春  1   901 1   3   3G  555   

  63.  *  2   吉林  2   902 1   2   3G  333   

  64.  *  3   四平  3   903 1   4   2G  777   

  65.  *  4   松原  4   904 1   5   3G  666   

  66.  */ 

  67. public class SemiJoin extends Configured implements Tool{   

  68.     private static final Logger logger = LoggerFactory.getLogger(SemiJoin.class);   

  69.     public static class SemiJoinMapper extends Mapper<Object, Text, Text, CombineValues> {   

  70.         private CombineValues combineValues = new CombineValues();   

  71.         private HashSet<String> joinKeySet = new HashSet<String>();   

  72.         private Text flag = new Text();   

  73.         private Text joinKey = new Text();   

  74.         private Text secondPart = new Text();   

  75.         /**   

  76.          * 将参加join的key从DistributedCache取出放到内存中，以便在map端将要参加join的key过滤出来。b   

  77.          */ 

  78.         @Override 

  79.         protected void setup(Context context)   

  80.                 throws IOException, InterruptedException {   

  81.             BufferedReader br = null;   

  82.             //获得当前作业的DistributedCache相关文件   

  83.             Path[] distributePaths = DistributedCache.getLocalCacheFiles(context.getConfiguration());   

  84.             String joinKeyStr = null;   

  85.             for(Path p : distributePaths){   

  86.                 if(p.toString().endsWith("joinKey.dat")){   

  87.                     //读缓存文件，并放到mem中   

  88.                     br = new BufferedReader(new FileReader(p.toString()));   

  89.                     while(null!=(joinKeyStr=br.readLine())){   

  90.                         joinKeySet.add(joinKeyStr);   

  91.                     }   

  92.                 }   

  93.             }   

  94.         }   

  95.         @Override 

  96.         protected void map(Object key, Text value, Context context)   

  97.                 throws IOException, InterruptedException {   

  98.             //获得文件输入路径   

  99.             String pathName = ((FileSplit) context.getInputSplit()).getPath().toString();   

  100.             //数据来自tb_dim_city.dat文件,标志即为"0"   

  101.             if(pathName.endsWith("tb_dim_city.dat")){   

  102.                 String[] valueItems = value.toString().split("\\|");   

  103.                 //过滤格式错误的记录   

  104.                 if(valueItems.length != 5){   

  105.                     return;   

  106.                 }   

  107.                 //过滤掉不需要参加join的记录   

  108.                 if(joinKeySet.contains(valueItems[0])){   

  109.                     flag.set("0");   

  110.                     joinKey.set(valueItems[0]);   

  111.                     secondPart.set(valueItems[1]+"\t"+valueItems[2]+"\t"+valueItems[3]+"\t"+valueItems[4]);   

  112.                     combineValues.setFlag(flag);   

  113.                     combineValues.setJoinKey(joinKey);   

  114.                     combineValues.setSecondPart(secondPart);   

  115.                     context.write(combineValues.getJoinKey(), combineValues);   

  116.                 }else{   

  117.                     return ;   

  118.                 }   

  119.             }//数据来自于tb_user_profiles.dat，标志即为"1"   

  120.             else if(pathName.endsWith("tb_user_profiles.dat")){   

  121.                 String[] valueItems = value.toString().split("\\|");   

  122.                 //过滤格式错误的记录   

  123.                 if(valueItems.length != 4){   

  124.                     return;   

  125.                 }   

  126.                 //过滤掉不需要参加join的记录   

  127.                 if(joinKeySet.contains(valueItems[3])){   

  128.                     flag.set("1");   

  129.                     joinKey.set(valueItems[3]);   

  130.                     secondPart.set(valueItems[0]+"\t"+valueItems[1]+"\t"+valueItems[2]);   

  131.                     combineValues.setFlag(flag);   

  132.                     combineValues.setJoinKey(joinKey);   

  133.                     combineValues.setSecondPart(secondPart);   

  134.                     context.write(combineValues.getJoinKey(), combineValues);   

  135.                 }else{   

  136.                     return ;   

  137.                 }   

  138.             }   

  139.         }   

  140.     }   

  141.     public static class SemiJoinReducer extends Reducer<Text, CombineValues, Text, Text> {   

  142.         //存储一个分组中的左表信息   

  143.         private ArrayList<Text> leftTable = new ArrayList<Text>();   

  144.         //存储一个分组中的右表信息   

  145.         private ArrayList<Text> rightTable = new ArrayList<Text>();   

  146.         private Text secondPar = null;   

  147.         private Text output = new Text();   

  148.         /**   

  149.          * 一个分组调用一次reduce函数   

  150.          */ 

  151.         @Override 

  152.         protected void reduce(Text key, Iterable<CombineValues> value, Context context)   

  153.                 throws IOException, InterruptedException {   

  154.             leftTable.clear();   

  155.             rightTable.clear();   

  156.             /**   

  157.              * 将分组中的元素按照文件分别进行存放   

  158.              * 这种方法要注意的问题：   

  159.              * 如果一个分组内的元素太多的话，可能会导致在reduce阶段出现OOM，   

  160.              * 在处理分布式问题之前最好先了解数据的分布情况，根据不同的分布采取最   

  161.              * 适当的处理方法，这样可以有效的防止导致OOM和数据过度倾斜问题。   

  162.              */ 

  163.             for(CombineValues cv : value){   

  164.                 secondPar = new Text(cv.getSecondPart().toString());   

  165.                 //左表tb_dim_city   

  166.                 if("0".equals(cv.getFlag().toString().trim())){   

  167.                     leftTable.add(secondPar);   

  168.                 }   

  169.                 //右表tb_user_profiles   

  170.                 else if("1".equals(cv.getFlag().toString().trim())){   

  171.                     rightTable.add(secondPar);   

  172.                 }   

  173.             }   

  174.             logger.info("tb_dim_city:"+leftTable.toString());   

  175.             logger.info("tb_user_profiles:"+rightTable.toString());   

  176.             for(Text leftPart : leftTable){   

  177.                 for(Text rightPart : rightTable){   

  178.                     output.set(leftPart+ "\t" + rightPart);   

  179.                     context.write(key, output);   

  180.                 }   

  181.             }   

  182.         }   

  183.     }   

  184.     @Override 

  185.     public int run(String[] args) throws Exception {   

  186.             Configuration conf=getConf(); //获得配置文件对象   

  187.             DistributedCache.addCacheFile(new Path(args[2]).toUri(), conf);

  188.             Job job=new Job(conf,"LeftOutJoinMR");   

  189.             job.setJarByClass(SemiJoin.class);

  190.  

  191.             FileInputFormat.addInputPath(job, new Path(args[0])); //设置map输入文件路径   

  192.             FileOutputFormat.setOutputPath(job, new Path(args[1])); //设置reduce输出文件路径

  193.  

  194.             job.setMapperClass(SemiJoinMapper.class);   

  195.             job.setReducerClass(SemiJoinReducer.class);

  196.  

  197.             job.setInputFormatClass(TextInputFormat.class); //设置文件输入格式   

  198.             job.setOutputFormatClass(TextOutputFormat.class);//使用默认的output格式

  199.  

  200.             //设置map的输出key和value类型   

  201.             job.setMapOutputKeyClass(Text.class);   

  202.             job.setMapOutputValueClass(CombineValues.class);

  203.  

  204.             //设置reduce的输出key和value类型   

  205.             job.setOutputKeyClass(Text.class);   

  206.             job.setOutputValueClass(Text.class);   

  207.             job.waitForCompletion(true);   

  208.             return job.isSuccessful()?0:1;   

  209.     }   

  210.     public static void main(String[] args) throws IOException,   

  211.             ClassNotFoundException, InterruptedException {   

  212.         try {   

  213.             int returnCode =  ToolRunner.run(new SemiJoin(),args);   

  214.             System.exit(returnCode);   

  215.         } catch (Exception e) {   

  216.             logger.error(e.getMessage());   

  217.         }   

  218.     }   

  219. } 
复制代码
这里还说说SemiJoin也是有一定的适用范围的，其抽取出来进行join的key是要放到内存中的，所以不能够太大，容易在Map端造成OOM。
三、总结
blog介绍了三种join方式。这三种join方式适用于不同的场景，其处理效率上的相差还是蛮大的，其中主要导致因素是网络传输。Map join效率最高，其次是SemiJoin，最低的是reduce join。另外，写分布式大数据处理程序的时最好要对整体要处理的数据分布情况作一个了解，这可以提高我们代码的效率，使数据的倾斜度降到最低，使我们的代码倾向性更好。


相关内容：
hadoop mapreduce join原理、方法讲解


关注公众号，获取大数据、人工智能20套、区块链资源5阶段等资源，随时更新，获取最新技术资源
回复

使用道具 举报

自定义广告语

已有(5)人评论

跳转到指定楼层
正序浏览
hahaxixi 发表于 2015-3-19 17:52:36
不错，学习下~~~~~~~~
回复

使用道具 举报

tchangwei 发表于 2015-3-19 16:06:12
提示: 作者被禁止或删除 内容自动屏蔽
回复

使用道具 举报

arsenduan 发表于 2015-3-19 16:48:49
tchangwei 发表于 2015-3-19 16:06
在Reudce端进行连接       中，ReduceSideJoin_LeftOuterJoin 类中的第101行开始创建的两个arraylist 是不 ...

一样的，一个全局，一个局部
回复

使用道具 举报

sprite101 发表于 2015-5-13 19:03:46
回复

使用道具 举报

程序猿的无奈 发表于 2016-6-4 15:23:30
楼主，我现在要三个表同时join，是两个job完成吗？还是有其他的解决办法
回复

使用道具 举报

高级模式
B Color Image Link Quote Code Smilies
您需要登录后才可以回帖 登录 | 立即注册

本版积分规则

发表新帖
desehawk

实习版主

关注

  • 1058

    主题

  • 1959

    帖子

  • 191

    粉丝

TA的主题
关闭

推荐上一条 /2 下一条

关于我们· 联系我们· 加入我们· 小黑屋· 合作伙伴   

Copyright © 2001-2024 About云-梭伦科技 Powered by Discuz! X3.4 Licensed Discuz Team.

简书 / Smrz 京ICP备2020039040号 Wxb 简书网举报电话：021-34700000