MapReduce之Map Join

时间：2024-09-23 12:34:02

一 介绍

之所以存在Reduce Join，是因为在map阶段不能获取所有需要的join字段，即：同一个key对应的字段可能位于不同map中。Reduce side join是非常低效的，因为shuffle阶段要进行大量的数据传输。

Map Join是针对以下场景进行的优化：两个待连接表中，有一个表非常大，而另一个表非常小，以至于小表可以直接存放到内存中。这样，我们可以将小表复制多份，让每个map task内存中存在一份（比如存放到hash table中），然后只扫描大表：对于大表中的每一条记录key/value，在hash table中查找是否有相同的key的记录，如果有，则连接后输出即可。

为了支持文件的共享，Hadoop用到了分布式缓存的概念，在MapReduce中称为DistributedCache（目前已被标注为弃用，分布式缓存的API可在Job类本身调用），它可以方便Map Task之间或Reduce Task之间共享一些信息，同时也可以将第三方Jar包添加到其Classpass路径中。Hadoop会将缓存数据分发到集群中所有准备启动的节点上，复制到mapreduce.temp.dir中的配置目录。

使用该类的方法如下：

job.addArchiveToClassPath(archive); //缓存jar包到task运行节点的classpath中

ob.addCacheArchive(uri); //缓存压缩包到task运行节点的工作目录

job.addFileToClassPath(file); //缓存普通文件到task运行节点的classpath中

job.addCacheFile(url); //将产品表文件缓存到task工作节点的工作目录中去

传参格式：hdfs://namenode:9000/home/XXX/file，即Jar包、压缩包、普通文件所在hdfs路径。

同时DistributedCache（分布式缓存）可用来解决join算法实现中的数据倾斜问题，例如两张表：订单表和产品表。

订单表：

订单号 时间 商品id 购买数量

1001,20170710,P0001,1

1002,20170710,P0001,3

1003,20170710,P0002,3

1004,20170710,P0002,4

产品表：　

商品id 商品名称

P0001,xiaomi

P0002,huawei

需求就是根据外键商品id来将两张表信息合并，拼接成 ：

1001 ,20170710,P0001,1 xiaomi

1002,20170710,P0001,3 xiaomi

1003,20170710,P0002,3,huawei

1004,20170710,P0002,4,huawei

考虑问题：在mapreduce程序中，如果某些产品非常畅销，肯定会产生很多订单，但是刚好这些订单信息都传到了一个reduce中（分区默认就是使用hashcode%reducetask数量，所以这种情况是正常的）。那么这个reducetask压力就很大了，而其他的reducetask处理的信息就很小，有的甚至就处理几条数据，这就出现了数据倾斜问题。

解决方案：一般来说订单表的数据远远多于产品表数据，毕竟产品的种类就那些，所以我们可以把产品信息都交给Map Task就行了逻辑都让Map Task来处理，也就是说不使用Reduce了，而让每个Map Task持有个product.data（存储产品信息的文件）即可。那么maptask怎么获得这个文件呢？刚好hadoop提供了DistributedCache，我们将文件交给这个分布式缓存，它会将我们的文件放到Map Task的工作目录中，那么Map 端可以直接从工作目录中去拿。

二 代码部分

 package mapreduce.DistributedCache;

 import java.io.BufferedReader;

 import java.io.FileInputStream;

 import java.io.IOException;

 import java.io.InputStream;

 import java.io.InputStreamReader;

 import java.net.URI;

 import java.util.HashMap;

 import java.util.Map;

 import org.apache.hadoop.conf.Configuration;

 import org.apache.hadoop.conf.Configured;

 import org.apache.hadoop.fs.Path;

 import org.apache.hadoop.io.LongWritable;

 import org.apache.hadoop.io.NullWritable;

 import org.apache.hadoop.io.Text;

 import org.apache.hadoop.mapreduce.Job;

 import org.apache.hadoop.mapreduce.Mapper;

 import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;

 import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

 import org.apache.hadoop.util.Tool;

 import org.apache.hadoop.util.ToolRunner;

 public class MapJoin extends Configured implements Tool{

     static class MapJoinMapper extends Mapper<LongWritable, Text, NullWritable, Text>{

         //用来缓存小文件（商品文件中的数据）

         Map<String, String> produceMap = new HashMap<String,String>();

         Text k = new Text();

         /*

          * 源码中能看到在循环执行map()之前会执行一次setUp方法,可以用来做初始化

          */

         @Override

         protected void setup(Context context)

                 throws IOException, InterruptedException {

             //将商品文件中的数据写到缓存中

             FileInputStream fileInput = new FileInputStream("product.data");

             //read data

             InputStreamReader readFile = new InputStreamReader(fileInput );

             BufferedReader br = new BufferedReader(readFile);

             String line = null;

             while((line=br.readLine())!=null){

                 //一行数据格式为P0001,xiaomi（商品id，商品名称）

                 String[] fields = line.split(",");

                 produceMap.put(fields[0], fields[1]);

             }

         }

         @Override

         protected void map(LongWritable key, Text value, Context context)

                 throws IOException, InterruptedException {

             //一行订单数据    格式为 1001,20170710,P0001,1（订单id，创建时间，商品id，购买商品数量）

             String line = value.toString();

             String[] fields = line.split(",");

             //根据订单数据中商品id在缓存中找出来对应商品信息(商品名称)，进行串接

             String productName = produceMap.get(fields[2]);

             k.set(line+","+productName);

             context.write(NullWritable.get(), k );

         }

     }

     public int run(String[] args) throws Exception {

     // step 1:get configuration

     Configuration conf = this.getConf();

         //set job

     Job job = Job.getInstance(conf);

         job.setJarByClass(MapJoin.class);

         job.setMapperClass(MapJoinMapper.class);

         job.setMapOutputKeyClass(Text.class);

         job.setMapOutputValueClass(NullWritable.class);

         //设置最终输出类型

         job.setOutputKeyClass(Text.class);

         job.setOutputValueClass(NullWritable.class);

         //将产品表文件缓存到task工作节点的工作目录中去

         //缓存普通文件到task运行节点的工作目录(hadoop帮我们完成)

         job.addCacheFile(new URI("hdfs://beifeng01:8020/user/beifeng01/mapreduce/input/mapjoin/product.data"));

         //不需要reduce，那么也就没有了shuffle过程

         job.setNumReduceTasks(0);

         FileInputFormat.setInputPaths(job, new Path(args[0]));

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

         boolean isSuccess = job.waitForCompletion(true);

         return isSuccess ? 0 : 1;

     }

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

         args = new String[]{

                 "hdfs://beifeng01:8020/user/beifeng01/mapreduce/input/mapjoin/orderid.data",

                 "hdfs://beifeng01:8020/user/beifeng01/mapreduce/output4"

         };

         Configuration conf = new Configuration();

         // run mapreduce

         int status = ToolRunner.run(conf, new MapJoin(), args);

         // exit program

         System.exit(status);

     }

 }

运行代码后查看输出结果

[hadoop@beifeng01 hadoop-2.5.0-cdh5.3.6]$ bin/hdfs dfs -text /user/beifeng01/mapreduce/output4/p*

1001,20170710,P0001,1,xiaomi

1002,20170710,P0001,3,xiaomi

1003,20170710,P0002,3,huawei

1004,20170710,P0002,4,huawei

　　

相关文章