图文精华

About云-梭伦科技»专题 › 交流区 › 技术交流 › MapReduce › hadoop 多数据源连接之DataJoin
分享

hadoop 多数据源连接之DataJoin

InSight 发表于 2015-5-4 16:38:02 [显示全部楼层] 回帖奖励 阅读模式 关闭右栏 0 18404
About云VIP帮助找工作和提升
问题导读：
1.Hadoop可以采用哪几种数据连接方式？
2.Redurce处理过程是怎样的？
3.实例操作？








一个MapReduce任务很可能访问和处理两个甚至多个数据集，在关系型数据库中，这将是两个或者多个表的连接，但是Hadoop系统没有关系型数据库中那样强大的连接处理功能，因此处理复杂一些。一般来讲，hadoop可以采用这几种数据连接方式：

        1 采用DataJoin类库实现Reduce端连接的方法

        2 用全局文件复制实现Map端连接方法

        3 带Map端过滤的Reduce端连接方法

   Hadoop的Mapreduce框架提供了一种较为通用  的多数据源连接方法，该方法用DataJoin类库为程序员提供了完成数据连接所需的编程框架和接口，其处理方法如下：

         为了完成不用数据源的连接操作，我们必须给每个数据源制定一个标签（tag），用来区分数据，就像关系型数据库中表名一样，这里我们需要实现 Text generateInputTag(String inputFile)方法；

         另外，为了进行连接操作，我们必须知道连接的主键是什么，类似于关系型数据库中的key，因此我们需要指定groupKey，这里我们需要实现 Text generateGroupKey(TaggedMapOutput aRecord)

         然后在Map端我们需要把原始数据包装成为一个带标签的数据记录，方便shuffle和Reduce端执行笛卡尔积，所以我们需要实现 TaggedMapOutput generateTaggedMapOutput(Object value);



总结一下

Map处理过程：

   Datajoin类库首先提哦功能管理一个抽象基类DataJoinMapperBase，该基类实现了map()方法，帮助程序员对每个数据源下的记录生成一个代标签的数据记录对象。Map端处理过程中，需要指定标签tag和Groupkey,然后包装成为带标签的数据记录对象，在shuffle过程中，这些GroupKey相同的记录被分到同一个Reduce节点上。

Reduce处理过程：

      Reduce节点收到这些带标签的数据记录后，Reduce过程将这些带不同的数据源标签的记录执行笛卡尔积，自动生成所有不同的叉积组合，由程序员实现一个combine()方法，根据应用程序需求将这些具有相同的Groupkey的数据记录进行适当的合并处理，以此完成类似于关系型数据库中不同实体数据记录之间的连接。

     在Reduce阶段我们需要继承DataJoinReduceBase,该基类实现了reduce()方法，我们只是需要实现combine()方法即可，另外我们还是需要继承TaggedMapOutput类，它描述了一个标签化的数据记录，实现了getTag(),setTag()方法，作为Mapper的key_value输出value类型，由于需要I/O,我们需要继承并且实现Writable接口，并且实现getData()方法用以读取记录数据

   下面是数据源：

        user.txt文件：

1,张三,135xxxxxxxx
2,李四,136xxxxxxxx
3,王五,137xxxxxxxx
4,赵六,138xxxxxxxx

order.txt文件：

3,A,13,2013-02-12
1,B,23,2013-02-14
2,C,16,2013-02-17
3,D,25,2013-03-12
这其中需要注意很多小细节，因为没有要求程序员实现Map和reduce方法，所以我们会很容易忽略很多东西，需要注意的东西我在下面一一注释了：

我们必须使用Jobconf 来声明一个job,同时使用JobClient来run job,另外我们在继承TaggedMapOutput的时候默认的无参构造方法中需要初始化data

  1. package joinTest;

  2. 

  3. import java.io.DataInput;

  4. import java.io.DataOutput;

  5. import java.io.IOException;

  6. 

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

  8. import org.apache.hadoop.contrib.utils.join.DataJoinMapperBase;

  9. import org.apache.hadoop.contrib.utils.join.DataJoinReducerBase;

  10. import org.apache.hadoop.contrib.utils.join.TaggedMapOutput;

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

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

  13. import org.apache.hadoop.io.Writable;

  14. import org.apache.hadoop.mapred.FileInputFormat;

  15. import org.apache.hadoop.mapred.FileOutputFormat;

  16. import org.apache.hadoop.mapred.JobClient;

  17. import org.apache.hadoop.mapred.JobConf;

  18. import org.apache.hadoop.mapred.TextInputFormat;

  19. import org.apache.hadoop.mapred.TextOutputFormat;

  20. 

  21. public class DataJoin{

  22. 

  23.         public static class MyTaggedMapOutput extends TaggedMapOutput{

  24.                 private Writable data;

  25.             

  26.                 public MyTaggedMapOutput(){

  27.                         //一定要new一下，不然反序列化时会报空指针异常

  28.                         this.data=new Text();

  29.                 }

  30.                 

  31.                 public MyTaggedMapOutput(Writable data){

  32.                         this.tag=new Text("");

  33.                         this.data=data;

  34.                 }

  35. 

  36.                 public void readFields(DataInput in) throws IOException {

  37.                         this.tag.readFields(in);

  38.                         this.data.readFields(in);

  39.                 }

  40. 

  41.                 public void write(DataOutput output) throws IOException {

  42.                         this.tag.write(output);

  43.                         //this.tag.write(output);     //大问题，粗心写成了this.tag.write(output); 结果一直报错

  44.                         this.data.write(output);

  45.                 }

  46. 

  47.                 @Override

  48.                 public Writable getData() {

  49.                         return data;

  50.                 }

  51.         }

  52.         

  53.         

  54.         public static class DataJoinMapper extends DataJoinMapperBase{

  55.                 

  56.                 protected Text generateInputTag(String inputFile){

  57.                                 //String datasource=inputFile.substring(inputFile.lastIndexOf("/")+1).split("\\.")[0];

  58.                             String datasource = inputFile.split("-")[0];

  59.                                 System.out.println("datasource:"+datasource);

  60.                                 return new Text(datasource);

  61.                 }

  62.                 

  63.                 protected TaggedMapOutput generateTaggedMapOutput(Object value){

  64.                         TaggedMapOutput tm=new MyTaggedMapOutput((Text)value);

  65.                         tm.setTag(this.inputTag);

  66.                         return tm;

  67.                 }

  68. 

  69.                 @Override

  70.                 protected Text generateGroupKey(TaggedMapOutput aRecord) {

  71.                         String line=aRecord.getData().toString();

  72.                         //String groupkey=line.split("\\s")[0];

  73.                         String groupkey=line.split(",")[0];

  74.                         return new Text(groupkey);

  75.                 }

  76.         }

  77.         

  78.         public static class DataJoinReducer extends DataJoinReducerBase{

  79. 

  80.                 @Override

  81.                 protected TaggedMapOutput combine(Object[] tags, Object[] values) {

  82.                         if(tags.length<2){

  83.                                 return null;

  84.                         }

  85.                         String output = "";

  86.                    /* for(int i=0;i<tags.length;i++){

  87.                             TaggedMapOutput tat=(MyTaggedMapOutput)values[i];System.out.println("tags:"+tags[i]+"    values:"+tat.getData().toString());

  88.                             if(i==0){

  89.                                  output=tat.getData().toString();//System.out.println("i==0  output:"+output);

  90.                             }else{

  91.                                     output+="\t";

  92.                                     String [] s=tat.getData().toString().split("\\s",2);

  93.                                     System.out.println("s.length:"+s.length);

  94.                                     output+=s[0];

  95.                             }*/

  96.                         for(int j=0;j<tags.length;j++){

  97.                                 //TaggedMapOutput taOutput=(TaggedMapOutput)tags[j];

  98.                                 TaggedMapOutput taggedMapOutput=(TaggedMapOutput)values[j];

  99.                                 System.out.println("tag:"+taggedMapOutput.getTag()+"  value:"+taggedMapOutput.getData().toString());

  100.                         }

  101.                              for(int i=0;i<values.length;i++){

  102.                                      TaggedMapOutput tat=(MyTaggedMapOutput)values[i];

  103.                                      String  recordLine=((Text)tat.getData()).toString();

  104.                                      String [] tokens=recordLine.split(",",2);System.out.println("data:"+recordLine);

  105.                                      if(i>0)

  106.                                              output+=",";

  107.                                     output+=tokens[1];

  108.                     }

  109.                     TaggedMapOutput tag=new MyTaggedMapOutput(new Text(output));

  110.                     tag.setTag((Text)tags[0]);

  111.                     return tag;

  112.                 }

  113.         }

  114.         

  115.         /*

  116.          * 这里一定要注意，FileInputFormat和FileOutputFormat一定要是org.apache.hadoop.mapred下面的包

  117.          */

  118.         public static int run(String args[]) throws IOException{

  119.                        Configuration conf=new Configuration();

  120.                                 //Configuration conf=getConf();

  121.                                 JobConf job=new JobConf(conf, DataJoin.class);

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

  123.                                 job.setJobName("DataJoin");

  124.                         //        job.setJarByClass(DataJoin.class);

  125.                                 

  126.                                 job.setMapperClass(DataJoinMapper.class);

  127.                                 job.setReducerClass(DataJoinReducer.class);

  128.                                 job.setInputFormat(TextInputFormat.class);

  129.                                 job.setOutputFormat(TextOutputFormat.class);

  130.                                 job.setMapOutputKeyClass(Text.class);

  131.                                 job.setMapOutputValueClass(MyTaggedMapOutput.class);

  132.                                 job.setOutputKeyClass(Text.class);

  133.                                 job.setOutputValueClass(Text.class);

  134.                                 

  135.                                 //job.set("mapred.textoutputformat.separator", "\t");

  136.                                 job.set("mapred.textoutputformat.separator", ",");

  137.                                 //FileInputFormat.addInputPath(job, new Path("/home/hadoop/test/mapReduce/DataJoinTest2"));

  138.                                 //FileInputFormat.addInputPath(job, new Path("hdfs://192.168.0.1:9000/user/hadoop/DataJoinTest2"));

  139.                                 //FileInputFormat.addInputPaths(job,  "/home/hadoop/test/DataJoinTest/province.txt");

  140.                                 //MultipleInputs.addInputPath(job, new Path("/home/hadoop/test/DataJoinTest/province.txt"), TextInputFormat.class);

  141.                                 FileInputFormat.addInputPaths(job, args[0]);

  142.                                 //FileOutputFormat.setOutputPath(job, new Path("hdfs://192.168.0.1:9000/user/hadoop/DataJoinTest2_Out"));

  143.                                 //FileOutputFormat.setOutputPath(job, new Path("/home/hadoop/test/mapReduce/DataJoinTest2_result"));

  144.                                 FileOutputFormat.setOutputPath(job, new Path(args[1]));  

  145.                                 JobClient.runJob(job);

  146.                                 return 0;

  147.         }

  148.         public static void main(String [] args) throws Exception{

  149.                 String[] arg = { "/home/hadoop/test/mapReduce/DataJoinTest2",

  150.                 "/home/hadoop/test/mapReduce/DataJoinTest2_result" };

  151.                 int res=run(arg);

  152.                 System.exit(res);

  153.                 

  154.         }

  155.         

  156. }

  157. 

  158. 

复制代码


转。
回复

使用道具 举报

自定义广告语

没找到任何评论，期待你打破沉寂

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

本版积分规则

发表新帖
InSight

中级会员

关注

  • 33

    主题

  • 141

    帖子

  • 11

    粉丝

TA的主题
关闭

推荐上一条 /2 下一条

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

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

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