ITPub博客

首页 > 大数据 > Hadoop > 使用DataJoin包实现Join

使用DataJoin包实现Join

Hadoop 作者:navybean 时间:2014-02-12 16:08:54 0 删除 编辑

概念:

Hadoop有一个叫DataJoin的包为Data Join提供相应的框架。它的Jar包存在于contrib/datajoin/hadoop-*-datajoin。

为区别于其他的data join技术,我们称其为reduce-side join。(因为我们在reducer上作大多数的工作)

reduce-side join引入了一些术语及概念:

            1.Data Source:基本与关系数据库中的表相似,形式为:(例子中为CSV格式)

      Customers                  Orders
      1,Stephanie Leung,555-555-5555      3,A,12.95,02-Jun-2008
      2,Edward Kim,123-456-7890         1,B,88.25,20-May-2008
      3,Jose Madriz,281-330-8004         2,C,32.00,30-Nov-2007
      4,David Stork,408-555-0000          3,D,25.02,22-Jan-2009

            2.Tag:由于记录类型(Customers或Orders)与记录本身分离,标记一个Record会确保特殊元数据会一致存在于记录中。在这个目的下,我们将使用每个record自身的Data source名称标记每个record。

            3.Group Key:Group Key类似于关系数据库中的链接键(join key),在我们的例子中,group key就是Customer ID(第一列的3)。由于datajoin包允许用户自定义group key,所以其较之关系数据库中的join key更一般、平常。


流程:(详见《Hadoop in Action》Chapter 5.2)

Advanced MapReduce:



Joining Data from different sources:




利用datajoin包来实现join:

  Hadoop的datajoin包中有三个需要我们继承的类:DataJoinMapperBase,DataJoinReducerBase,TaggedMapOutput。正如其名字一样,我们的MapClass将会扩展DataJoinMapperBase,Reduce类会扩展DataJoinReducerBase。这个datajoin包已经实现了map()和reduce()方法,因此我们的子类只需要实现一些新方法来设置一些细节。

  

  在用DataJoinMapperBase和DataJoinReducerBase之前,我们需要弄清楚我们贯穿整个程序使用的新的虚数据类TaggedMapOutput。

  

  根据之前我们在图Advance MapReduce的数据流中所展示的那样,mapper输出一个包(由一个key和一个value(tagged record)组成)。datajoin包将key设置为Text类型,将value设置为TaggedMapOutput类型(TaggedMapOutput是一个将我们的记录使用一个Text类型的tag包装起来的数据类型)。它实现了getTag()和setTag(Text tag)方法。它还定义了一个getData()方法,我们的子类将实现这个方法来处理record记录。我们并没有明确地要求子类实现setData()方法,但我们最好还是实现这个方法以实现程序的对称性(或者在构造函数中实现)。作为Mapper的输出,TaggedMapOutput需要是Writable类型,因此的子类还需要实现readFields()和write()方法。


DataJoinMapperBase:

  回忆join数据流图,mapper的主要功能就是打包一个record使其能够和其他拥有相同group key的记录去向一个Reducer。DataJoinMapperBase完成所有的打包工作,这个类定义了三个虚类让我们的子类实现:

  protected abstract Text generateInputTag(String inputFile);

  protected abstract TaggedMapOutput generateTaggedMapOutut(Object value);

  protected abstract Text generateGroupKey(TaggedMapOutput aRecored);

  

  在一个map任务开始之前为所有这个map任务会处理的记录定义一个tag(Text),结果将保存到DataJoinMapperBase的inputTag变量中,我们也可以保存filename至inputFile变量中以待后用。


  在map任务初始化之后,DataJoinMapperBase的map()方法会对每一个记录执行。它调用了两个我们还没有实现的虚方法:generateTaggedMapOutput()以及generateGroupKey(aRecord);(详见代码)


DataJoinReducerBase:

DataJoinMapperBase将我们所需要做的工作以一个full outer join的方式简化。我们的Reducer子类只需要实现combine()方法来滤除掉我们不需要的组合来得到我们需要的(inner join, left outer join等)。同时我们也在combiner()中将我们的组合格式化为输出格式。

代码:

  1. import java.io.DataInput;  
  2. import java.io.DataOutput;  
  3. import java.io.IOException;  
  4. import java.util.Iterator;  
  5.   
  6. import org.apache.hadoop.conf.Configuration;  
  7. import org.apache.hadoop.conf.Configured;  
  8. import org.apache.hadoop.fs.Path;  
  9. import org.apache.hadoop.io.Text;  
  10. import org.apache.hadoop.io.Writable;  
  11. import org.apache.hadoop.mapred.FileInputFormat;  
  12. import org.apache.hadoop.mapred.FileOutputFormat;  
  13. import org.apache.hadoop.mapred.JobClient;  
  14. import org.apache.hadoop.mapred.JobConf;  
  15. import org.apache.hadoop.mapred.KeyValueTextInputFormat;  
  16. import org.apache.hadoop.mapred.MapReduceBase;  
  17. import org.apache.hadoop.mapred.Mapper;  
  18. import org.apache.hadoop.mapred.OutputCollector;  
  19. import org.apache.hadoop.mapred.Reducer;  
  20. import org.apache.hadoop.mapred.Reporter;  
  21. import org.apache.hadoop.mapred.TextInputFormat;  
  22. import org.apache.hadoop.mapred.TextOutputFormat;  
  23. import org.apache.hadoop.util.Tool;  
  24. import org.apache.hadoop.util.ToolRunner;  
  25.   
  26. import org.apache.hadoop.contrib.utils.join.DataJoinMapperBase;  
  27. import org.apache.hadoop.contrib.utils.join.DataJoinReducerBase;  
  28. import org.apache.hadoop.contrib.utils.join.TaggedMapOutput;  
  29.   
  30. public class DataJoin extends Configured implements Tool  
  31.       
  32.     public static class MapClass extends DataJoinMapperBase  
  33.           
  34.         protected Text generateInputTag(String inputFile)  
  35.             String datasource inputFile.split("-")[0];  
  36.             return new Text(datasource);  
  37.          
  38.           
  39.         protected Text generateGroupKey(TaggedMapOutput aRecord)  
  40.             String line ((Text) aRecord.getData()).toString();  
  41.             String[] tokens line.split(",");  
  42.             String groupKey tokens[0];  
  43.             return new Text(groupKey);  
  44.          
  45.           
  46.         protected TaggedMapOutput generateTaggedMapOutput(Object value)  
  47.             TaggedWritable retv new TaggedWritable((Text) value);  
  48.             retv.setTag(this.inputTag);  
  49.             return retv;  
  50.          
  51.      
  52.       
  53.     public static class Reduce extends DataJoinReducerBase  
  54.           
  55.         protected TaggedMapOutput combine(Object[] tags, Object[] values)  
  56.             if (tags.length 2return null   
  57.             String joinedStr ""  
  58.             for (int i=0i
  59.                 if (i 0joinedStr += "," 
  60.                 TaggedWritable tw (TaggedWritable) values[i];  
  61.                 String line ((Text) tw.getData()).toString();  
  62.                 String[] tokens line.split(","2);  
  63.                 joinedStr += tokens[1];  
  64.              
  65.             TaggedWritable retv new TaggedWritable(new Text(joinedStr));  
  66.             retv.setTag((Text) tags[0]);   
  67.             return retv;  
  68.          
  69.      
  70.       
  71.     public static class TaggedWritable extends TaggedMapOutput  
  72.       
  73.         private Writable data;  
  74.           
  75.         public TaggedWritable(Writable data)  
  76.             this.tag new Text("");  
  77.             this.data data;  
  78.          
  79.           
  80.         public Writable getData()  
  81.             return data;  
  82.          
  83.           
  84.         public void write(DataOutput out) throws IOException  
  85.             this.tag.write(out);  
  86.             this.data.write(out);  
  87.          
  88.           
  89.         public void readFields(DataInput in) throws IOException  
  90.             this.tag.readFields(in);  
  91.             this.data.readFields(in);  
  92.          
  93.      
  94.       
  95.     public int run(String[] args) throws Exception  
  96.         Configuration conf getConf();  
  97.           
  98.         JobConf job new JobConf(conf, DataJoin.class);  
  99.           
  100.         Path in new Path(args[0]);  
  101.         Path out new Path(args[1]);  
  102.         FileInputFormat.setInputPaths(job, in);  
  103.         FileOutputFormat.setOutputPath(job, out);  
  104.           
  105.         job.setJobName("DataJoin");  
  106.         job.setMapperClass(MapClass.class);  
  107.         job.setReducerClass(Reduce.class);  
  108.           
  109.         job.setInputFormat(TextInputFormat.class);  
  110.         job.setOutputFormat(TextOutputFormat.class);  
  111.         job.setOutputKeyClass(Text.class);  
  112.         job.setOutputValueClass(TaggedWritable.class);  
  113.         job.set("mapred.textoutputformat.separator"",");  
  114.           
  115.         JobClient.runJob(job);   
  116.         return 0 
  117.      
  118.       
  119.     public static void main(String[] args) throws Exception   
  120.         int res ToolRunner.run(new Configuration(),  
  121.                                  new DataJoin(),  
  122.                                  args);  
  123.           
  124.         System.exit(res);  
  125.      
  126.  
<!-- 正文结束 -->

来自 “ ITPUB博客 ” ,链接:http://blog.itpub.net/24200923/viewspace-1119730/,如需转载,请注明出处,否则将追究法律责任。

上一篇: 没有了~
下一篇: 没有了~
请登录后发表评论 登录
全部评论

注册时间:2010-07-06