`
weitao1026
  • 浏览: 996188 次
  • 性别: Icon_minigender_1
  • 来自: 上海
社区版块
存档分类
最新评论

使用Hadoop的MapReduce来完成大表join

hadoop 
阅读更多

我们都知道在数据库里，多个表之间是可以根据某个链接键进行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

Java代码 复制代码 收藏代码
  1. a文件的数据  
  2.   
  3. 1,三劫散仙,13575468248  
  4. 2,凤舞九天,18965235874  
  5. 3,忙忙碌碌,15986854789  
  6. 4,少林寺方丈,15698745862  
a文件的数据

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



Java代码 复制代码 收藏代码
  1. b文件的数据  
  2.   
  3. 3,A,99,2013-03-05  
  4. 1,B,89,2013-02-05  
  5. 2,C,69,2013-03-09  
  6. 3,D,56,2013-06-07  
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


源码如下：

Java代码 复制代码 收藏代码
  1. package com.qin.reducejoin;  
  2.   
  3.   
  4. import java.io.DataInput;    
  5. import java.io.DataOutput;    
  6. import java.io.IOException;    
  7. import java.util.Iterator;    
  8.     
  9. import org.apache.hadoop.conf.Configuration;    
  10. import org.apache.hadoop.conf.Configured;    
  11. import org.apache.hadoop.fs.FileSystem;  
  12. import org.apache.hadoop.fs.Path;    
  13. import org.apache.hadoop.io.Text;    
  14. import org.apache.hadoop.io.Writable;    
  15. import org.apache.hadoop.mapred.FileInputFormat;    
  16. import org.apache.hadoop.mapred.FileOutputFormat;    
  17. import org.apache.hadoop.mapred.JobClient;    
  18. import org.apache.hadoop.mapred.JobConf;    
  19. import org.apache.hadoop.mapred.KeyValueTextInputFormat;    
  20. import org.apache.hadoop.mapred.MapReduceBase;    
  21. import org.apache.hadoop.mapred.Mapper;    
  22. import org.apache.hadoop.mapred.OutputCollector;    
  23. import org.apache.hadoop.mapred.Reducer;    
  24. import org.apache.hadoop.mapred.Reporter;    
  25. import org.apache.hadoop.mapred.TextInputFormat;    
  26. import org.apache.hadoop.mapred.TextOutputFormat;    
  27. import org.apache.hadoop.util.ReflectionUtils;  
  28. import org.apache.hadoop.util.Tool;    
  29. import org.apache.hadoop.util.ToolRunner;    
  30.     
  31. import org.apache.hadoop.contrib.utils.join.DataJoinMapperBase;    
  32. import org.apache.hadoop.contrib.utils.join.DataJoinReducerBase;    
  33. import org.apache.hadoop.contrib.utils.join.TaggedMapOutput;    
  34.   
  35. import com.qin.joinreduceerror.JoinReduce;  
  36.     
  37.   
  38. /*** 
  39.  *  
  40.  * Hadoop1.2的版本，旧版本实现的Reduce侧连接 
  41.  *  
  42.  * @author qindongliang 
  43.  *  
  44.  *    大数据交流群：376932160 
  45.  *  搜索技术交流群：324714439 
  46.  *  
  47.  *  
  48.  */  
  49.   
  50. public class DataJoin extends Configured implements Tool {    
  51.         
  52.       
  53.     /** 
  54.      *  
  55.      * Map实现 
  56.      *  
  57.      * */  
  58.     public static class MapClass extends DataJoinMapperBase {    
  59.             
  60.         /** 
  61.          * 读取输入的文件路径 
  62.          *  
  63.          * **/  
  64.         protected Text generateInputTag(String inputFile) {    
  65.               
  66.             //返回文件路径，做标记  
  67.             return new Text(inputFile);    
  68.         }    
  69.             
  70.           
  71.         /*** 
  72.          * 分组的Key 
  73.          *  
  74.          * **/  
  75.           
  76.         protected Text generateGroupKey(TaggedMapOutput aRecord) {    
  77.             String line = ((Text) aRecord.getData()).toString();    
  78.             String[] tokens = line.split(",");    
  79.             String groupKey = tokens[0];    
  80.             return new Text(groupKey);    
  81.         }    
  82.             
  83.           
  84.           
  85.           
  86.         protected TaggedMapOutput generateTaggedMapOutput(Object value) {    
  87.             TaggedWritable retv = new TaggedWritable((Text) value);    
  88.             retv.setTag(this.inputTag);    
  89.             return retv;    
  90.         }    
  91.     }    
  92.         
  93.     /** 
  94.      *  
  95.      * Reduce进行笛卡尔积 
  96.      *  
  97.      * **/  
  98.     public static class Reduce extends DataJoinReducerBase {    
  99.             
  100.           
  101.         /*** 
  102.          * 笛卡尔积 
  103.          *  
  104.          * */  
  105.         protected TaggedMapOutput combine(Object[] tags, Object[] values) {    
  106.             if (tags.length < 2) return null;      
  107.             String joinedStr = "";     
  108.             for (int i=0; i<values.length; i++) {    
  109.                 if (i > 0) {joinedStr += ",";}    
  110.                 TaggedWritable tw = (TaggedWritable) values[i];    
  111.                 String line = ((Text) tw.getData()).toString();    
  112.                 String[] tokens = line.split(",", 2);    
  113.                 joinedStr += tokens[1];    
  114.             }    
  115.             TaggedWritable retv = new TaggedWritable(new Text(joinedStr));    
  116.             retv.setTag((Text) tags[0]);     
  117.             return retv;    
  118.         }    
  119.     }    
  120.         
  121.     /** 
  122.      *  
  123.      * 自定义的输出类型 
  124.      *  
  125.      * ***/  
  126.     public static class TaggedWritable extends TaggedMapOutput {    
  127.         
  128.         private Writable data;    
  129.             
  130.         /** 
  131.          * 注意加上构造方法 
  132.          *  
  133.          * */  
  134.         public TaggedWritable() {  
  135.             // TODO Auto-generated constructor stub  
  136.         }  
  137.           
  138.         public TaggedWritable(Writable data) {    
  139.             this.tag = new Text("");    
  140.             this.data = data;    
  141.         }    
  142.             
  143.         public Writable getData() {    
  144.             return data;    
  145.         }    
  146.             
  147.         public void write(DataOutput out) throws IOException {    
  148.             this.tag.write(out);    
  149.             //此行代码很重要  
  150.             out.writeUTF(this.data.getClass().getName());  
  151.              
  152.             this.data.write(out);   
  153.               
  154.         }    
  155.             
  156.         public void readFields(DataInput in) throws IOException {    
  157.             this.tag.readFields(in);    
  158.               //加入此部分代码，否则，可能报空指针异常  
  159.             String temp=in.readUTF();  
  160.             if (this.data == null|| !this.data.getClass().getName().equals(temp)) {  
  161.                     try {  
  162.                     this.data = (Writable) ReflectionUtils.newInstance(  
  163.                     Class.forName(temp), null);  
  164.                     } catch (ClassNotFoundException e) {  
  165.                     e.printStackTrace();  
  166.                     }  
  167.                     }  
  168.             this.data.readFields(in);    
  169.         }    
  170.     }    
  171.         
  172.     public int run(String[] args) throws Exception {    
  173.         Configuration conf = getConf();    
  174.             
  175.         JobConf job = new JobConf(conf, DataJoin.class);    
  176.           
  177.         job.set("mapred.job.tracker","192.168.75.130:9001");  
  178.             ////读取person中的数据字段  
  179.              job.setJar("tt.jar");  
  180.         job.setJarByClass(DataJoin.class);  
  181.             System.out.println("模式：  "+job.get("mapred.job.tracker"));;  
  182.                
  183.                
  184.                
  185.                
  186.         String path="hdfs://192.168.75.130:9000/root/outputjoindb";  
  187.         FileSystem fs=FileSystem.get(conf);  
  188.         Path p=new Path(path);  
  189.         if(fs.exists(p)){  
  190.             fs.delete(p, true);  
  191.             System.out.println("输出路径存在，已删除！");  
  192.         }  
  193.        
  194.           
  195.           
  196.         Path in = new Path("hdfs://192.168.75.130:9000/root/inputjoindb");    
  197.       //  Path out = new Path(args[1]);    
  198.         FileInputFormat.setInputPaths(job, in);    
  199.         FileOutputFormat.setOutputPath(job, p);    
  200.             
  201.         job.setJobName("cee");    
  202.         job.setMapperClass(MapClass.class);    
  203.         job.setReducerClass(Reduce.class);    
  204.             
  205.         job.setInputFormat(TextInputFormat.class);    
  206.         job.setOutputFormat(TextOutputFormat.class);    
  207.         job.setOutputKeyClass(Text.class);    
  208.         job.setOutputValueClass(TaggedWritable.class);    
  209.         job.set("mapred.textoutputformat.separator", ",");    
  210.             
  211.         JobClient.runJob(job);     
  212.         return 0;    
  213.     }    
  214.         
  215.     public static void main(String[] args) throws Exception {     
  216.           
  217.           
  218.           
  219.         int res = ToolRunner.run(new Configuration(),    
  220.                                  new DataJoin(),    
  221.                                  args);    
  222.             
  223.         System.exit(res);    
  224.     }    
  225. }    
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[i];  
                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);  
    }  
}




运行，日志

Java代码 复制代码 收藏代码
  1. 模式：  192.168.75.130:9001  
  2. 输出路径存在，已删除！  
  3. INFO - NativeCodeLoader.<clinit>(43) | Loaded the native-hadoop library  
  4. WARN - LoadSnappy.<clinit>(46) | Snappy native library not loaded  
  5. INFO - FileInputFormat.listStatus(199) | Total input paths to process : 2  
  6. INFO - JobClient.monitorAndPrintJob(1380) | Running job: job_201404222310_0025  
  7. INFO - JobClient.monitorAndPrintJob(1393) |  map 0% reduce 0%  
  8. INFO - JobClient.monitorAndPrintJob(1393) |  map 33% reduce 0%  
  9. INFO - JobClient.monitorAndPrintJob(1393) |  map 100% reduce 0%  
  10. INFO - JobClient.monitorAndPrintJob(1393) |  map 100% reduce 33%  
  11. INFO - JobClient.monitorAndPrintJob(1393) |  map 100% reduce 100%  
  12. INFO - JobClient.monitorAndPrintJob(1448) | Job complete: job_201404222310_0025  
  13. INFO - Counters.log(585) | Counters: 30  
  14. INFO - Counters.log(587) |   Job Counters   
  15. INFO - Counters.log(589) |     Launched reduce tasks=1  
  16. INFO - Counters.log(589) |     SLOTS_MILLIS_MAPS=14335  
  17. INFO - Counters.log(589) |     Total time spent by all reduces waiting after reserving slots (ms)=0  
  18. INFO - Counters.log(589) |     Total time spent by all maps waiting after reserving slots (ms)=0  
  19. INFO - Counters.log(589) |     Launched map tasks=3  
  20. INFO - Counters.log(589) |     Data-local map tasks=3  
  21. INFO - Counters.log(589) |     SLOTS_MILLIS_REDUCES=9868  
  22. INFO - Counters.log(587) |   File Input Format Counters   
  23. INFO - Counters.log(589) |     Bytes Read=207  
  24. INFO - Counters.log(587) |   File Output Format Counters   
  25. INFO - Counters.log(589) |     Bytes Written=172  
  26. INFO - Counters.log(587) |   FileSystemCounters  
  27. INFO - Counters.log(589) |     FILE_BYTES_READ=837  
  28. INFO - Counters.log(589) |     HDFS_BYTES_READ=513  
  29. INFO - Counters.log(589) |     FILE_BYTES_WRITTEN=221032  
  30. INFO - Counters.log(589) |     HDFS_BYTES_WRITTEN=172  
  31. INFO - Counters.log(587) |   Map-Reduce Framework  
  32. INFO - Counters.log(589) |     Map output materialized bytes=849  
  33. INFO - Counters.log(589) |     Map input records=8  
  34. INFO - Counters.log(589) |     Reduce shuffle bytes=849  
  35. INFO - Counters.log(589) |     Spilled Records=16  
  36. INFO - Counters.log(589) |     Map output bytes=815  
  37. INFO - Counters.log(589) |     Total committed heap usage (bytes)=496644096  
  38. INFO - Counters.log(589) |     CPU time spent (ms)=2080  
  39. INFO - Counters.log(589) |     Map input bytes=187  
  40. INFO - Counters.log(589) |     SPLIT_RAW_BYTES=306  
  41. INFO - Counters.log(589) |     Combine input records=0  
  42. INFO - Counters.log(589) |     Reduce input records=8  
  43. INFO - Counters.log(589) |     Reduce input groups=4  
  44. INFO - Counters.log(589) |     Combine output records=0  
  45. INFO - Counters.log(589) |     Physical memory (bytes) snapshot=623570944  
  46. INFO - Counters.log(589) |     Reduce output records=4  
  47. INFO - Counters.log(589) |     Virtual memory (bytes) snapshot=2908262400  
  48. INFO - Counters.log(589) |     Map output records=8  
模式：  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的测试集群上进行测试。

分享到：
| 在hadoop里面处理的数据，默认按输入内容的 ...
评论
发表评论

您还没有登录,请您登录后再发表评论

相关推荐

    hadoop mapreduce多表关联join多个job相互依赖传递参数

    mapreduce多表关联join多个job相互依赖传递参数

    hadoop0.23.9离线api

    org.apache.hadoop.mapreduce.lib.join org.apache.hadoop.mapreduce.lib.map org.apache.hadoop.mapreduce.lib.output org.apache.hadoop.mapreduce.lib.partition org.apache.hadoop.mapreduce.lib.reduce ...

    word源码java-hadoop-test:hadoop、mapreduce的一些练习

    包org.dan.mr.order_pro_mapjoin MapReduce实现订单信息和产品信息的join逻辑，在Mapper端实现，避免数据倾斜 包org.dan.mr.wordindex MapReduce单词索引 包org.dan.mr.shared_friends MapReduce查找共同好友 包org....

    MapReduce-Code:Hadoop平台下的MapReduce源码分析

    其他表示父目录下的.java文件的总称）：1.org.apache.hadoop.mapred（旧版MapReduceAPI）：( 1）.jobcontrol（job作业直接控制类）（2 ）.join ：(作业作业中用于模仿数据连接处理工具）（3）.lib（MapReduce所依赖...

    MapReduce之Join操作

    本文对Hadoop中最基本的join方法进行简单介绍，这也是其它许多方法和优化措施的基础。文中所采用的例子来自于《HadoopinAction》一书中的5.2节。假设两个表所在的文件分别为Customers和Orders，以CSV格式存储在HDFS...

    Hadoop实战(第2版)

    ·如何整合MapReduce和R前言 致谢关于本书 第1 部分　背景和基本原理1　跳跃中的Hadoop1．1　什么是Hadoop 1．1．1　Hadoop 的核心组件1．1．2　Hadoop 生态圈1．1．3　物理架构1．1．4　谁在使用Hadoop1．1．5　...

    史上最全面的hadoop入门视频教程

    手把手视频详细讲解项目开发全过程，需要的小伙伴自行百度网盘下载，链接见附件...第七章 MapReduce的运行机制和join操作 第八章 MapReduce的其他操作和yarn 第九章 数仓Hive基本操作 第十章 数仓Hive的其他操作和调优

    hadoop-2.2.0-x64.tar.gz part3

    [INFO] Apache Hadoop Data Join ........................... SUCCESS [5.463s] [INFO] Apache Hadoop Extras .............................. SUCCESS [3.394s] [INFO] Apache Hadoop Pipes ........................

    hadoop-2.2.0-x64.tar.gz part2

    [INFO] Apache Hadoop Data Join ........................... SUCCESS [5.463s] [INFO] Apache Hadoop Extras .............................. SUCCESS [3.394s] [INFO] Apache Hadoop Pipes ........................

    Hadoop开发者第四期

    Join 算子在Hadoop 中的实现； 配置Hive 元数据DB 为PostgreSQL； ZooKeeper 权限管理机制； ZooKeeper 服务器工作原理和流程； ZooKeeper 实现共享锁； Hadoop 最佳实践； 通过Hadoop 的API 管理Job； Hadoop 集群...

    MapReduce Design Pattern

    reorganize data to work with other systems, or to make MapReduce analysis easier Join patterns: analyze different datasets together to discover interesting relationships Metapatterns: piece together ...

    5堂Hadoop必修课，不会这些勿称高手

    大数据分布式集群搭建（HA），构建企业级MapReduce项目，hadoop和spark源码编译，Zookeeper, MapReduce高级Join操作等等，不会这些别说你是hadoop高手

    hadoop_join_aggregate:在hadoop中加入和聚合mapreduce算法

    Query4.java 使用两个 map-reduce 任务来完成。 此方法适用于减速器侧连接。 它需要一个 map-reduce 任务来加入 2 个数据集，将中间结果写入 HDFS，另一个 map-reduce 任务读回中间结果进行聚合。 Query4_1.java ...

    hadoop-2.2.0-x64.tar.gz part1

    自己编译的64bit的hadoop-2.2.0，自己lab安装使用。 [INFO] Reactor Summary: [INFO] [INFO] Apache Hadoop Main ................................ SUCCESS [1.834s] [INFO] Apache Hadoop Project POM .............

    Hadoop硬实战 [（美）霍姆斯著][电子工业出版社][2015.01]_PDF电子书下载 带书签目录 高清完整版.rar )

    技术点38　使用堆转储来查找未优化的用户代码 6．2．5　硬件性能问题 技术点39　查找硬件的失效 技术点40　CPU 竞争 ． 技术点41　内存交换 技术点42　磁盘健康 技术点43　网络 6．3　可视化 技术点44　提取并可视化...

    19、Join操作map side join 和 reduce side join

    19、Join操作map side join 和 reduce side join 网址：...本文介绍mapreduce的join操作。 本文前提是hadoop可以正常使用。 本文分为3个部分介绍，即join的介绍、map side join和reduce side join。

    Data-Intensive Text Processing with MapReduce Jimmy Lin and Chris Dyer

    2.6 Hadoop Cluster Architecture 2.7 Summary 3 MapReduce Algorithm Design 3.1 Local Aggregation 3.1.1 Combiners and In-Mapper Combining 3.1.2 Algorithmic Correctness with Local Aggregation 3.2 ...

Magicbox
Global site tag (gtag.js) - Google Analytics