ITPub博客
首页 博客 专栏 学院 社区 IT新媒体评选
发博

首页 > 大数据 > Hadoop > 细节决定成败 MapReduce任务实战 Map Join

细节决定成败 MapReduce任务实战 Map Join

原创 Hadoop 作者：self_control 时间：2016-06-14 14:46:44 0 删除 编辑

一、任务描述

在hdfs 上有两个文件

学生信息文件：hdfs://***.***.***:8020/user/train/joinjob/student.txt

以逗号分隔，第一列为学号，学号每个学生唯一，第二列为学生姓名。

点击(此处)折叠或打开

  1. 2016001,Join
  2. 2016002,Abigail
  3. 2016003,Abby
  4. 2016004,Alexandra
  5. 2016005,Cathy
  6. 2016006,Katherine

学生成绩信息文件：hdfs://***.***.***:8020/user/train/joinjob/student_score.txt

以逗号分隔，第一列为学号，学号每个学生唯一，第二列为课程代码，第三列为分数。

点击(此处)折叠或打开

  1. 2016001,YY,60
  2. 2016001,SX,88
  3. 2016001,YW,91
  4. 2016002,SX,77
  5. 2016002,YW,33
  6. .............

期望的结果文件形式如下

点击(此处)折叠或打开

  1. 2016001,Join,YY,60
  2. 2016001,Join,SX,88
  3. 2016001,Join,YW,91
  4. 2016002,Abigail,SX,77
  5. 2016002,Abigail,YW,33
  6. 2016002,Abigail,YY,56
  7. 2016003,Abby,YY,34
  8. 2016003,Abby,SX,84
  9. 2016003,Abby,YW,69
  10. 2016004,Alexandra,YY,89
  11. 2016004,Alexandra,SX,84
  12. .......

二、任务分析

  1. 这是一个两个数据集关联的任务。关联有map端join ，和reduce端join。 通常在有一个数据集比较小，可在全部放到内存中时，选择map端join；当两个数据集都很大时，使用reduce端join，这时可以使用Bloom过滤器来增加效率。 本次任务我们使用map端join实现。
  2. 此任务关联到直接输出即可，不需要reduce过程。
  3. 本任务会使用到的技术有：分布式缓存DistributedCache、mapreduce中的本地文件读取

三、实现思路

  1. 把一个学生信息使用分布式缓存分发到各个任务节点。
  2. 学生成绩信息当做map的input。
  3. 在mapper 的setup函数中读取已缓存到本的学生信息，并保存到一个HashMap中，key为学号，value为姓名。
  4. 在mapper函数中对每一个学生成绩信息，根据学号去上一步的map中去取姓名信息。
  5. 封装到Text对象中输出。

三、实现代码

点击(此处)折叠或打开

  1. package join;
  2. import java.io.BufferedReader;
  3. import java.io.IOException;
  4. import java.io.InputStreamReader;
  5. import java.net.URI;
  6. import java.util.HashMap;
  7. import java.util.Map;

  9. import org.apache.commons.lang.StringUtils;
  10. import org.apache.hadoop.conf.Configuration;
  11. import org.apache.hadoop.conf.Configured;
  12. import org.apache.hadoop.fs.FileSystem;
  13. import org.apache.hadoop.fs.Path;
  14. import org.apache.hadoop.io.LongWritable;
  15. import org.apache.hadoop.io.NullWritable;
  16. import org.apache.hadoop.io.Text;
  17. import org.apache.hadoop.mapreduce.Job;
  18. import org.apache.hadoop.mapreduce.Mapper;
  19. import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
  20. import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
  21. import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
  22. import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
  23. import org.apache.hadoop.util.Tool;
  24. import org.apache.hadoop.util.ToolRunner;
  25. import org.apache.log4j.Logger;


  28. public class MapJoinDirectOutPutJob extends Configured implements Tool {
  29.         private static String STUDENTNAME_PARAM ="STUDENTNAME_PARAM";
  30.         private static String INPATH_SCORE = "joinjob/student_score.txt";
  31.         private static String OUTPATH = "joinjob/output";
  32.         public static class MapJoinDirectOutPutMapper extends Mapper<LongWritable,Text,NullWritable,Text>{
  33.                 private BufferedReader br = null;
  34.                 private Map <String,String> map =new HashMap<String,String>();
  35.                 private Text newValue = new Text();
  36.                 @Override
  37.                 protected void map(LongWritable key, Text value, Context context)
  38.                                 throws IOException, InterruptedException {
  39.                         String [] words = StringUtils.split(value.toString(),',');
  40.                         String name = map.get(words[0]);
  41.                         newValue.set(words[0]+","+name +","+words[1]+","+words[2]);
  42.                         context.write(NullWritable.get(), newValue);
  43.                 }

  45.                 @Override
  46.                 protected void setup(Context context) throws IOException,
  47.                                 InterruptedException {
  48.                         Configuration conf = context.getConfiguration();
  49.                         FileSystem fs = FileSystem.getLocal(conf);

  51.                         br = new BufferedReader(new InputStreamReader(fs.open(new Path("studentLink.txt"))));
  52.                         String current;
  53.                         while((current= br.readLine())!=null){
  54.                                 String [] words = current.split(",");
  55.                                 map.put(words[0],words[1]);
  56.                         }
  57.                         br.close();
  58.                 }

  60.         }

  62.         @Override
  63.         public int run(String[] args) throws Exception {
  64.                 Job job = Job.getInstance(getConf(),"MapJoinDirectOutPutJob");
  65.                 job.setJarByClass(getClass());
  66.                 Configuration conf = job.getConfiguration();

  68.                 conf.set(STUDENTNAME_PARAM, args[0]);

  70.                 Path in = new Path(INPATH_SCORE);
  71.                 Path out = new Path(OUTPATH);
  72.                 FileSystem.get(conf).delete(out,true);
  73.                 FileInputFormat.setInputPaths(job, in);
  74.                 FileOutputFormat.setOutputPath(job, out);
  75.                 job.setInputFormatClass(TextInputFormat.class);
  76.                 job.setOutputFormatClass(TextOutputFormat.class);
  77.                 job.setMapperClass(MapJoinDirectOutPutMapper.class);
  78.                 job.setMapOutputKeyClass(NullWritable.class);
  79.                 job.setMapOutputValueClass(Text.class);

  81.                 job.setNumReduceTasks(0);
  82.                 URI uri = new URI("hdfs://***.***.***.***:8020/user/train/joinjob/student.txt#studentLink.txt");
  83.                 job.addCacheFile(uri);

  85.                 return job.waitForCompletion(true) ?0:1;
  86.         }

  88.         public static void main(String [] args){
  89.                 int r = 0;
  90.                 try{
  91.                         r = ToolRunner.run(new Configuration(),new MapJoinDirectOutPutJob(),args);
  92.                 }catch(Exception e){
  93.                         e.printStackTrace();
  94.                 }
  95.                 System.exit(r);
  96.         }

  98. }
本文地址：http://blog.itpub.net/30066956/viewspace-2120063/



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

2
0
上一篇： 一个MapReduce 程序示例 细节决定成败(九)：RawComparator
下一篇： 细节决定成败 MapReduce任务实战 Reduce Join
请登录后发表评论 登录
全部评论
self_control

注册时间：2014-12-16

  • 博文量
    49
  • 访问量
    406901

最新文章

支持我们 作者招募 用户协议 FAQ Contact Us

北京盛拓优讯信息技术有限公司. 版权所有  京ICP备16024965号-8 北京市公安局海淀分局网监中心备案编号：11010802021510 niuxiaotong@pcpop.com 17352615567 未成年人举报专区

广播电视节目制作经营许可证(京) 字第1234号 中国互联网协会会员

违法和不良信息举报电话:18101376593