前言：通过MR不同方式的Join编程是为了更加熟悉join的实现过程以及不同方式的优缺点，切记，生产中要杜绝写MR，本文只供学习参考

1.需求

有两张表，分表是产品信息数据以及用户页面点击日志数据如下：

  1. #产品信息数据:product_info.txt
  2. #c1=产品ID（id）,c2=产品名称(name),c3=价格(privce),c4=生产国家(country)
  3. p0001,华为,8000,中国
  4. p0002,小米,3000,中国
  5. p0003,苹果,1500,美国
  6. p0004,三星,10000,韩国
  8. #用户页面点击日志数据:page_click_log.txt
  9. #c1=用户ID（id）,c2=产品id(prod_id),c3=点击时间(click_time),c4=动作发生地区(area)
  10. u0001,p0001,20190301040123,华中
  11. u0002,p0002,20190302040124,华北
  12. u0003,p0003,20190303040124,华南
  13. u0004,p0004,20190304040124,华南

由于点击日志的数据量过去庞大，数据是存在HDFS上，故需要使用MR来实现如下的逻辑:

  1. select b.id,b.name,b.privce,b.country,a.id,a.click_time,a.area 
  2. from page_click_log a join product_info b on a.prod_id=b.id

2.Map端实现Join

2.1思路分析

可以将小表数据分发到所有的map节点，然后可以与在本所读到的大表数据进行join并输出最终结果
优缺点：大大提高了jion的并发，速度快

2.2编程实现

数据封装类Info.java

  1. package com.wsk.bigdata.pojo;
  3. import org.apache.hadoop.io.Writable;
  5. import java.io.DataInput;
  6. import java.io.DataOutput;
  7. import java.io.IOException;
  8. import java.util.ArrayList;
  10. public class Info implements Writable {
  12.     /**
  13.      * 产品唯一标识id
  14.      */
  15.     private String pId;
  17.     /**
  18.      * 产品名称
  19.      */
  20.     private String pName;
  22.     /**
  23.      * 产品价格
  24.      */
  25.     private float price;
  27.     /**
  28.      * 产品生产地区
  29.      */
  30.     private String produceArea;
  32.     /**
  33.      * 用户Id
  34.      */
  35.     private String uId;
  37.     /**
  38.      * 用户点击时间戳：yyyyMMddHHmmss
  39.      */
  40.     private String dateStr;
  42.     /**
  43.      * 用户点击发生地区
  44.      */
  45.     private String clickArea;
  48.     /**
  49.      * flag=0，表示封装用户点击日志数据
  50.      * flag=1，表示封装产品信息
  51.      */
  52.     private String flag;
  54.     public String getpId() {
  55.         return pId;
  56.     }
  58.     public void setpId(String pId) {
  59.         this.pId = pId;
  60.     }
  62.     public String getpName() {
  63.         return pName;
  64.     }
  66.     public void setpName(String pName) {
  67.         this.pName = pName;
  68.     }
  70.     public float getPrice() {
  71.         return price;
  72.     }
  74.     public void setPrice(float price) {
  75.         this.price = price;
  76.     }
  78.     public String getProduceArea() {
  79.         return produceArea;
  80.     }
  82.     public void setProduceArea(String produceArea) {
  83.         this.produceArea = produceArea;
  84.     }
  86.     public String getuId() {
  87.         return uId;
  88.     }
  90.     public void setuId(String uId) {
  91.         this.uId = uId;
  92.     }
  94.     public String getDateStr() {
  95.         return dateStr;
  96.     }
  98.     public void setDateStr(String dateStr) {
  99.         this.dateStr = dateStr;
  100.     }
  102.     public String getClickArea() {
  103.         return clickArea;
  104.     }
  106.     public void setClickArea(String clickArea) {
  107.         this.clickArea = clickArea;
  108.     }
  110.     public String getFlag() {
  111.         return flag;
  112.     }
  114.     public void setFlag(String flag) {
  115.         this.flag = flag;
  116.     }
  118.     public Info(String pId, String pName, float price, String produceArea, String uId, String dateStr, String clickArea, String flag) {
  119.         this.pId = pId;
  120.         this.pName = pName;
  121.         this.price = price;
  122.         this.produceArea = produceArea;
  123.         this.uId = uId;
  124.         this.dateStr = dateStr;
  125.         this.clickArea = clickArea;
  126.         this.flag = flag;
  127.     }
  129.     public Info() {
  130.     }
  132.     @Override
  133.     public String toString() {
  134.         String[] fileds = {this.pId,};
  135.         return "pid=" + this.pId + ",pName=" + this.pName + ",price=" + this.price
  136.                 + ",produceArea=" + this.produceArea
  137.                 + ",uId=" + this.uId + ",clickDate=" + this.dateStr + ",clickArea=" + this.clickArea;
  138.     }
  140.     @Override
  141.     public void write(DataOutput out) throws IOException {
  143.         out.writeUTF(this.pId);
  144.         out.writeUTF(this.pName);
  145.         out.writeFloat(this.price);
  146.         out.writeUTF(this.produceArea);
  147.         out.writeUTF(this.uId);
  148.         out.writeUTF(this.dateStr);
  149.         out.writeUTF(this.clickArea);
  150.         out.writeUTF(this.flag);
  152.     }
  154.     @Override
  155.     public void readFields(DataInput in) throws IOException {
  156.         this.pId = in.readUTF();
  157.         this.pName = in.readUTF();
  158.         this.price = in.readFloat();
  159.         this.produceArea = in.readUTF();
  160.         this.uId = in.readUTF();
  161.         this.dateStr = in.readUTF();
  162.         this.clickArea = in.readUTF();
  163.         this.flag= in.readUTF();
  165.     }
  166. }

map实现类FileMapJoinMapper.java

  1. package com.wsk.bigdata.mapreduce.mapper;
  3. import com.wsk.bigdata.pojo.Info;
  4. import org.apache.commons.lang3.StringUtils;
  5. import org.apache.hadoop.io.LongWritable;
  6. import org.apache.hadoop.io.NullWritable;
  7. import org.apache.hadoop.io.Text;
  8. import org.apache.hadoop.mapreduce.Mapper;
  10. import java.io.BufferedReader;
  11. import java.io.FileInputStream;
  12. import java.io.IOException;
  13. import java.io.InputStreamReader;
  14. import java.util.HashMap;
  15. import java.util.Map;
  17. /**
  18.  * 文件间的Mapjoin
  19.  */
  20. public class FileMapJoinMapper extends Mapper<LongWritable, Text, Info, NullWritable> {
  22.     /**
  23.      * 产品信息信息集合，key=产品ID，value=产品信息
  24.      */
  25.     private Map<String, Info> infos = new HashMap<>();
  28.     /**
  29.      * 执行Map方法前会调用一次setup方法，我们可以用于
  30.      * 初始化读取产品信息加到到内存中
  31.      *
  32.      */
  33.     @Override
  35.     protected void setup(Context context) throws IOException, InterruptedException {
  36.         System.out.println("--------MAP初始化：加载产品信息数据到内存------");
  37.         BufferedReader br = new BufferedReader(new InputStreamReader(new FileInputStream(System.getProperty("product.info.dir"))));
  38.         String line;
  40.         while (StringUtils.isNotEmpty(line = br.readLine())) {
  41.             String[] fields = line.split(",");
  42.             if (fields != null && fields.length == 4) {
  43.                 Info info = new Info(fields[0], fields[1], Float.parseFloat(fields[2]), fields[3], "", "", "", "1");
  44.                 infos.put(fields[0], info);
  45.             }
  46.         }
  47.         br.close();
  48.         System.out.println("--------MAP初始化：共加载了" + infos.size() + "条产品信息数据------");
  50.     }
  52.     @Override
  53.     protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
  54.         String line = value.toString();
  55.         String[] fields = line.split(",");
  56.         if (fields != null && fields.length == 4) {
  57.             String pid = fields[1];
  58.             Info produceInfo = infos.get(pid);
  59.             if (produceInfo == null) {
  60.                 return;
  61.             }
  62.             Info info = new Info(produceInfo.getpId(), produceInfo.getpName(), produceInfo.getPrice(), produceInfo.getProduceArea()
  63.                     , fields[0], fields[2], fields[3], null);
  64.             context.write(info, NullWritable.get());
  66.         }
  68.     }
  70. }

程序入口类MapJoinDriver.java

  1. package com.wsk.bigdata.mapreduce.driver;
  3. import com.wsk.bigdata.mapreduce.mapper.FileMapJoinMapper;
  4. import com.wsk.bigdata.pojo.Info;
  5. import org.apache.hadoop.conf.Configuration;
  6. import org.apache.hadoop.fs.FileSystem;
  7. import org.apache.hadoop.fs.Path;
  8. import org.apache.hadoop.io.NullWritable;
  9. import org.apache.hadoop.mapreduce.Job;
  10. import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
  11. import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
  13. import java.io.IOException;
  15. public class MapJoinDriver {
  17.     public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {
  19.         if(args.length != 3 ) {
  20.             System.err.println("please input 3 params: product_File page_click_file output_mapjoin directory");
  21.             System.exit(0);
  22.         }
  23.         String productInfo = args[0];
  24.         String input = args[1];
  25.         String output = args[2];
  26.         System.setProperty("hadoop.home.dir", "D:\\appanzhuang\\cdh\\hadoop-2.6.0-cdh5.7.0");
  27.         System.setProperty("product.info.dir",productInfo);
  28.         Configuration conf = new Configuration();
  29.         // 写代码：死去活来法
  30.         FileSystem fs = FileSystem.get(conf);
  31.         Path outputPath = new Path(output);
  32.         if(!fs.exists(new Path(productInfo))){
  33.             System.err.println("not found File "+productInfo);
  34.             System.exit(0);
  35.         }
  36.         if(fs.exists(outputPath)) {
  37.             fs.delete(outputPath, true);
  38.         }
  40.         Job job = Job.getInstance(conf);
  41.         job.setJarByClass(MapJoinDriver.class);
  42.         job.setMapperClass(FileMapJoinMapper.class);
  43.         // 指定mapper输出数据的kv类型
  44.         job.setMapOutputKeyClass(Info.class);
  45.         job.setMapOutputValueClass(NullWritable.class);
  47.         FileInputFormat.setInputPaths(job, new Path(input));
  48.         FileOutputFormat.setOutputPath(job, new Path(output));
  50.         // map端join的逻辑不需要reduce阶段，设置reducetask数量为0
  51.         job.setNumReduceTasks(0);
  53.         boolean res = job.waitForCompletion(true);
  54.     }
  55. }

程序运行参数，分别是产品信息文件路径、页面点击日志数据路径、输出结果路径
[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-78w2BgKy-1608566731456)(https://s2.ax1x.com/2019/04/28/EMZCv9.md.png)]

3.Reduce端实现Join

3.1思路

通过将关联的条件作为map输出的key，将两表满足join条件的数据并携带数据所来源的文件信息，发往同一个reduce task，在reduce中进行数据的串联
优缺点：这种方式中，join的操作是在reduce阶段完成，reduce端的处理压力太大，map节点的运算负载则很低，资源利用率不高，且在reduce阶段极易产生数据倾斜

3.2编程实现

map实现类FileReduceJoinMapper.java

  1. package com.wsk.bigdata.mapreduce.mapper;
  3. import com.wsk.bigdata.pojo.Info;
  4. import org.apache.hadoop.io.LongWritable;
  5. import org.apache.hadoop.io.Text;
  6. import org.apache.hadoop.mapreduce.Mapper;
  7. import org.apache.hadoop.mapreduce.lib.input.FileSplit;
  9. import java.io.IOException;
  11. public class FileReduceJoinMapper extends Mapper<LongWritable, Text, Text, Info> {
  14.     Text k = new Text();
  16.     @Override
  17.     protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
  19.         String line = value.toString();
  20.         String[] fields = line.split(",");
  21.         String pid = "";
  22.         Info info = null;
  23.         // 通过文件名判断是哪种数据
  24.         FileSplit inputSplit = (FileSplit) context.getInputSplit();
  25.         String name = inputSplit.getPath().getName();
  26.         if (name.startsWith("product")) {
  27.             pid=fields[0];
  28.             info = new Info(pid,fields[1],Float.parseFloat(fields[2]),fields[3],"","","","1");
  29.         } else {
  30.             pid=fields[1];
  31.             info = new Info(pid,"",0,"",fields[0],fields[2],fields[3],"0");
  32.         }
  33.         if(info==null){
  34.             return;
  35.         }
  37.         k.set(pid);
  38.         System.out.println("map 输出"+info.toString());
  39.         context.write(k, info);
  40.     }
  41. }

reducer实现类FileReduceJoinReducer.java

  1. package com.wsk.bigdata.mapreduce.reduce;
  3. import com.wsk.bigdata.pojo.Info;
  4. import org.apache.commons.beanutils.BeanUtils;
  5. import org.apache.hadoop.io.NullWritable;
  6. import org.apache.hadoop.io.Text;
  7. import org.apache.hadoop.mapreduce.Reducer;
  9. import java.io.IOException;
  10. import java.lang.reflect.InvocationTargetException;
  11. import java.util.ArrayList;
  12. import java.util.Iterator;
  13. import java.util.List;
  15. public class FileReduceJoinReducer extends Reducer<Text, Info, Info, NullWritable> {
  17.     @Override
  18.     protected void reduce(Text key, Iterable<Info> values, Context context) throws IOException, InterruptedException {
  20.         Info pInfo = new Info();
  21.         List<Info> clickBeans = new ArrayList<Info>();
  23.         Iterator<Info> iterator = values.iterator();
  24.         while (iterator.hasNext()) {
  25.             Info bean = iterator.next();
  26.             System.out.println("reduce接收 "+bean);
  27.             if ("1".equals(bean.getFlag())) { //产品
  28.                 try {
  29.                     BeanUtils.copyProperties(pInfo, bean);
  30.                 } catch (IllegalAccessException | InvocationTargetException e) {
  31.                     e.printStackTrace();
  32.                 }
  33.             } else {
  34.                 Info clickBean = new Info();
  35.                 try {
  36.                     BeanUtils.copyProperties(clickBean, bean);
  37.                     clickBeans.add(clickBean);
  38.                 } catch (IllegalAccessException | InvocationTargetException e) {
  39.                     e.printStackTrace();
  40.                 }
  41.             }
  42.         }
  44.         // 拼接数据获取最终结果
  45.         for (Info bean : clickBeans) {
  46.             bean.setpName(pInfo.getpName());
  47.             bean.setPrice(pInfo.getPrice());
  48.             bean.setProduceArea(pInfo.getProduceArea());
  49.             System.out.println("reduce结果输出："+bean.toString());
  50.             context.write(bean, NullWritable.get());
  51.         }
  52.     }
  54. }

程序入口ReduceJoinDriver.java

  1. package com.wsk.bigdata.mapreduce.driver;
  3. import com.wsk.bigdata.mapreduce.mapper.FileReduceJoinMapper;
  4. import com.wsk.bigdata.mapreduce.reduce.FileReduceJoinReducer;
  5. import com.wsk.bigdata.pojo.Info;
  6. import org.apache.hadoop.conf.Configuration;
  7. import org.apache.hadoop.fs.FileSystem;
  8. import org.apache.hadoop.fs.Path;
  9. import org.apache.hadoop.io.NullWritable;
  10. import org.apache.hadoop.io.Text;
  11. import org.apache.hadoop.mapreduce.Job;
  12. import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
  13. import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
  15. import java.io.IOException;
  17. public class ReduceJoinDriver {
  18.     public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {
  19.         if (args.length != 2) {
  20.             System.err.println("please input 2 params: inpt_data output_mapjoin directory");
  21.             System.exit(0);
  22.         }
  23.         String input = args[0];
  24.         String output = args[1];
  25.         System.setProperty("hadoop.home.dir", "D:\\appanzhuang\\cdh\\hadoop-2.6.0-cdh5.7.0");
  26.         Configuration conf = new Configuration();
  27.         FileSystem fs = FileSystem.get(conf);
  28.         Path outputPath = new Path(output);
  29.         if (fs.exists(outputPath)) {
  30.             fs.delete(outputPath, true);
  31.         }
  34.         Job job = Job.getInstance(conf);
  35.         job.setJarByClass(ReduceJoinDriver.class);
  36.         job.setMapperClass(FileReduceJoinMapper.class);
  37.         job.setReducerClass(FileReduceJoinReducer.class);
  38.         // 指定mapper输出数据的kv类型
  39.         job.setMapOutputKeyClass(Text.class);
  40.         job.setMapOutputValueClass(Info.class);
  41.         //定义Reducer输出数据的kv类型
  42.         job.setOutputKeyClass(Info.class);
  43.         job.setOutputValueClass(NullWritable.class);
  45.         FileInputFormat.setInputPaths(job, new Path(input));
  46.         FileOutputFormat.setOutputPath(job, new Path(output));
  48.         boolean res = job.waitForCompletion(true);
  50.         if (!res) {
  51.             System.err.println("error：作业执行失败");
  52.         }
  53.     }
  54. }

程序运行的两个参数，第一是输入输入目录，第二个是输出数据目录
[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-Oo7xIw12-1608566731458)(https://s2.ax1x.com/2019/04/28/EMmjEt.md.png)]

备注踩坑：

  • 代码引包时要注意引入的类是否正确
  • domain类重写序列化方法，一定要包含所有的字段，不然会导致字段缺少值以及值串位