一. 背景 在实际的数据库应用中，我们经常需要从多个数据表中读取数据，这时我们就可以使用SQL语句中的连接（JOIN），在两个或多个数据表中查询数据。
在使用MapReduce框架进行数据处理的过程中，也会涉及到从多个数据集读取数据，进行join关联的操作，只不过此时需要使用java代码并且根据MapReduce的编程规范进行业务的实现。
但是由于MapReduce的分布式设计理念的特殊性，因此对于MapReduce实现join操作具备了一定的特殊性。特殊主要体现在：究竟在MapReduce中的什么阶段进行数据集的关联操作，是mapper阶段还是reducer阶段，之间的区别又是什么？
整个MapReduce的join分为两类：map side join、reduce side join。
二. reduce side join 1. 概述
reduce side join，顾名思义，在reduce阶段执行join关联操作。这也是最容易想到和实现的join方式。因为通过shuffle过程就可以将相关的数据分到相同的分组中，这将为后面的join操作提供了便捷。

文章图片

基本上，reduce side join大致步骤如下：

1. mapper分别读取不同的数据集；
2. mapper的输出中，通常以join的字段作为输出的key；
3. 不同数据集的数据经过shuffle，key一样的会被分到同一分组处理；
4. 在reduce中根据业务需求把数据进行关联整合汇总，最终输出。

2. 弊端
reduce端join最大的问题是整个join的工作是在reduce阶段完成的，但是通常情况下MapReduce中reduce的并行度是极小的（默认是1个），这就使得所有的数据都挤压到reduce阶段处理，压力颇大。虽然可以设置reduce的并行度，但是又会导致最终结果被分散到多个不同文件中。
并且在数据从mapper到reducer的过程中，shuffle阶段十分繁琐，数据集大时成本极高。
三. MapReduce 分布式缓存 DistributedCache是hadoop框架提供的一种机制,可以将job指定的文件,在job执行前,先行分发到task执行的机器上,并有相关机制对cache文件进行管理。
DistributedCache能够缓存应用程序所需的文件 （包括文本，档案文件，jar文件等）。
Map-Redcue框架在作业所有任务执行之前会把必要的文件拷贝到slave节点上。 它运行高效是因为每个作业的文件只拷贝一次并且为那些没有文档的slave节点缓存文档。
1. 使用方式
1. 添加缓存文件 可以使用MapReduce的API添加需要缓存的文件。

//添加归档文件到分布式缓存中 job.addCacheArchive(URI uri); //添加普通文件到分布式缓存中 job.addCacheFile(URI uri);

注意：需要分发的文件,必须提前放到hdfs上.默认的路径前缀是hdfs://。
2. 程序中读取缓存文件 在Mapper类或者Reducer类的setup方法中，用输入流获取分布式缓存中的文件。

protected void setup(Context context) throw IOException,InterruptedException{ FileReader reader = new FileReader("myfile"); BufferReader br = new BufferedReader(reader); ...... }

四. map side join 1. 概述
map side join，其精髓就是在map阶段执行join关联操作,并且程序也没有了reduce阶段，避免了shuffle时候的繁琐。实现的关键是使用MapReduce的分布式缓存。

文章图片

尤其是涉及到一大一小数据集的处理场景时，map端的join将会发挥出得天独厚的优势。
map side join的大致思路如下：

1. 首先分析join处理的数据集，使用分布式缓存技术将小的数据集进行分布式缓存
2. MapReduce框架在执行的时候会自动将缓存的数据分发到各个maptask运行的机器上
3. 程序只运行mapper，在mapper初始化的时候从分布式缓存中读取小数据集数据，然后和自己读取的大数据集进行join关联，输出最终的结果。
4. 整个join的过程没有shuffle，没有reducer。

2. 优势
map端join最大的优势减少shuffle时候的数据传输成本。并且mapper的并行度可以根据输入数据量自动调整，充分发挥分布式计算的优势。
五. MapReduce join案例：订单商品处理 1. 需求
有两份结构化的数据文件：itheima_goods（商品信息表）、itheima_order_goods（订单信息表）,具体字段内容如下。
要求使用MapReduce统计出每笔订单中对应的具体的商品名称信息。比如107860商品对应着：AMAZFIT黑色硅胶腕带
数据结构:

1. itheima_goods
   字段：goodsId（商品id）、goodsSn（商品编号）、goodsName（商品名称）
   
   文章图片
   
2. itheima_order_goods
   字段： orderId（订单ID）、goodsId（商品ID）、payPrice（实际支付价格）
   
   文章图片
   

2. Reduce Side 实现
1. 分析 使用mapper处理订单数据和商品数据，输出的时候以goodsId商品编号作为key。相同goodsId的商品和订单会到同一个reduce的同一个分组，在分组中进行订单和商品信息的关联合并。在MapReduce程序中可以通过context获取到当前处理的切片所属的文件名称。根据文件名来判断当前处理的是订单数据还是商品数据，以此来进行不同逻辑的输出。
join处理完之后，最后可以再通过MapReduce程序排序功能，将属于同一笔订单的所有商品信息汇聚在一起。
2. 代码实现

1. Mapper
   

   package com.uuicon.sentiment_upload.join; import org.apache.hadoop.io.LongWritable; import org.apache.hadoop.io.Text; import org.apache.hadoop.mapreduce.Mapper; import org.apache.hadoop.mapreduce.lib.input.FileSplit; import java.io.IOException; public class ReduceJoinMapper extends Mapper { Text outKey = new Text(); Text outValue = https://www.it610.com/article/new Text(); StringBuffer sb = new StringBuffer(); String fileName = null; @Override protected void setup(Context context) throws IOException, InterruptedException { super.setup(context); FileSplit split = (FileSplit) context.getInputSplit(); fileName = split.getPath().getName(); System.out.println("当前文件----" + fileName); } @Override protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException { sb.setLength(0); String[] fields = value.toString().split("\\|"); if (fileName.contains("itheima_goods.txt")) { // 100101|155083444927602|四川果冻橙6个约180g/个 outKey.set(fields[0]); sb.append(fields[1] + "\t" + fields[2]); outValue.set(sb.insert(0, "goods#").toString()); context.write(outKey, outValue); } else { // 1|107860|7191 outKey.set(fields[1]); StringBuffer append = sb.append(fields[0]).append("\t").append(fields[1]).append("\t").append(fields[2]); outValue.set(sb.insert(0, "orders#").toString()); context.write(outKey, outValue); } } }

   
   
2. Reduce 类

package com.uuicon.sentiment_upload.join; import org.apache.hadoop.io.LongWritable; import org.apache.hadoop.io.Text; import org.apache.hadoop.mapreduce.Mapper; import org.apache.hadoop.mapreduce.lib.input.FileSplit; import java.io.IOException; public class ReduceJoinMapper extends Mapper { Text outKey = new Text(); Text outValue = https://www.it610.com/article/new Text(); StringBuffer sb = new StringBuffer(); String fileName = null; @Override protected void setup(Context context) throws IOException, InterruptedException { super.setup(context); FileSplit split = (FileSplit) context.getInputSplit(); fileName = split.getPath().getName(); System.out.println("当前文件----" + fileName); }@Override protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException { sb.setLength(0); String[] fields = value.toString().split("\\|"); if (fileName.contains("itheima_goods.txt")) { // 100101|155083444927602|四川果冻橙6个约180g/个 outKey.set(fields[0]); sb.append(fields[1] + "\t" + fields[2]); outValue.set(sb.insert(0, "goods#").toString()); context.write(outKey, outValue); } else { // 1|107860|7191 outKey.set(fields[1]); StringBuffer append = sb.append(fields[0]).append("\t").append(fields[1]).append("\t").append(fields[2]); outValue.set(sb.insert(0, "orders#").toString()); context.write(outKey, outValue); } } }

3. 驱动类
   

   package com.uuicon.sentiment_upload.join; import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.fs.FileSystem; import org.apache.hadoop.fs.Path; import org.apache.hadoop.io.Text; import org.apache.hadoop.mapreduce.Job; import org.apache.hadoop.mapreduce.lib.input.FileInputFormat; import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat; public class ReduceJoinDriver { public static void main(String[] args) throws Exception { // 配置文件对象 Configuration conf = new Configuration(); // 创建作业实例 Job job = Job.getInstance(conf, ReduceJoinDriver.class.getSimpleName()); // 设置作业驱动类 job.setJarByClass(ReduceJoinDriver.class); // 设置作业Mapper reduce 类 job.setMapperClass(ReduceJoinMapper.class); job.setReducerClass(ReduceJoinReducer.class); // 设置作业 mapper 阶段输出key value 数据类型, job.setMapOutputKeyClass(Text.class); job.setMapOutputValueClass(Text.class); // 设置作业reducer 阶段输出key value 数据类型,也就是程序最终输出的数据类型 job.setOutputKeyClass(Text.class); job.setOutputValueClass(Text.class); // 配置作业的输入数据路径 FileInputFormat.addInputPath(job, new Path(args[0])); // 配置作业的输出数据路径 FileOutputFormat.setOutputPath(job, new Path(args[1])); // 判断输出路径是否存在,如果存在,删除 FileSystem fs = FileSystem.get(conf); if (fs.exists(new Path(args[1]))) { fs.delete(new Path(args[1]), true); } boolean resultFlag = job.waitForCompletion(true); System.exit(resultFlag ? 0 : 1); } }

   
   
4. 结果排序

package com.uuicon.sentiment_upload.join; import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.fs.FileSystem; 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.Reducer; import org.apache.hadoop.mapreduce.lib.input.FileInputFormat; import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat; import java.io.IOException; public class ReduceJoinSort { public static class ReduceJoinSortMapper extends Mapper { Text outKey = new Text(); Text outValue = https://www.it610.com/article/new Text(); @Override protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException { //227810010138155083444927602四川果冻橙6个约180g/个 String[] fields = value.toString().split("\t"); outKey.set(fields[0]); outValue.set(fields[0] + "\t" + fields[1] + "\t" + fields[3] + "\t" + fields[4] + "\t" + fields[2]); context.write(outKey, outValue); } }public static class ReduceJoinSortReducer extends Reducer { @Override protected void reduce(Text key, Iterable values, Context context) throws IOException, InterruptedException { for (Text value : values) { context.write(value, NullWritable.get()); } } }public static void main(String[] args) throws Exception { // 配置文件对象 Configuration conf = new Configuration(); // 创建作业实例 Job job = Job.getInstance(conf, ReduceJoinSort.class.getSimpleName()); // 设置作业驱动类 job.setJarByClass(ReduceJoinSort.class); // 设置作业Mapper reduce 类 job.setMapperClass(ReduceJoinSortMapper.class); job.setReducerClass(ReduceJoinSortReducer.class); // 设置作业 mapper 阶段输出key value 数据类型, job.setMapOutputKeyClass(Text.class); job.setMapOutputValueClass(Text.class); // 设置作业reducer 阶段输出key value 数据类型,也就是程序最终输出的数据类型 job.setOutputKeyClass(Text.class); job.setOutputValueClass(NullWritable.class); // 配置作业的输入数据路径 FileInputFormat.addInputPath(job, new Path(args[0])); // 配置作业的输出数据路径 FileOutputFormat.setOutputPath(job, new Path(args[1])); // 判断输出路径是否存在,如果存在,删除 FileSystem fs = FileSystem.get(conf); if (fs.exists(new Path(args[1]))) { fs.delete(new Path(args[1]), true); } boolean resultFlag = job.waitForCompletion(true); System.exit(resultFlag ? 0 : 1); } }

5. 运行结果
   
   • reduce join的结果
   
   文章图片
   
   
   • 重新排序之后的结果
   
   文章图片
   
   

3. Map Side 实现
1. 分析 Map-side Join是指在Mapper任务中加载特定数据集，此案例中把商品数据进行分布式缓存，使用Mapper读取订单数据和缓存的商品数据进行连接。
通常为了方便使用，会在mapper的初始化方法setup中读取分布式缓存文件加载的程序的内存中，便于后续mapper处理数据。
因为在mapper阶段已经完成了数据的关联操作，因此程序不需要进行reduce。需要在job中将reducetask的个数设置为0,也就是mapper的输出就是程序最终的输出。
2. 代码实现

1. Mapper 类

package com.uuicon.sentiment_upload.cache; import org.apache.commons.collections.map.HashedMap; import org.apache.hadoop.io.LongWritable; import org.apache.hadoop.io.NullWritable; import org.apache.hadoop.io.Text; import org.apache.hadoop.mapreduce.Mapper; import java.io.BufferedReader; import java.io.FileReader; import java.io.IOException; import java.util.Map; public class ReduceCacheMapper extends Mapper { Map goodsMap = new HashedMap(); Text outKey = new Text(); @Override protected void setup(Context context) throws IOException, InterruptedException { //加载缓存文件 BufferedReader br = new BufferedReader(new FileReader("itheima_goods.txt")); String line = null; while ((line = br.readLine()) != null) { String[] fields = line.split("\\|"); goodsMap.put(fields[0], fields[1] + "\t" + fields[2]); } }@Override protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException { //56982|100917|1192 String[] fields = value.toString().split("\\|"); outKey.set(value.toString() + "\t" + goodsMap.get(fields[1])); context.write(outKey, NullWritable.get()); } }

2. 程序主类

package com.uuicon.sentiment_upload.cache; import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.fs.FileSystem; import org.apache.hadoop.fs.Path; import org.apache.hadoop.io.NullWritable; import org.apache.hadoop.io.Text; import org.apache.hadoop.mapreduce.Job; import org.apache.hadoop.mapreduce.lib.input.FileInputFormat; import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat; import java.net.URI; public class ReduceCacheDriver { public static void main(String[] args) throws Exception { // 配置文件对象 Configuration conf = new Configuration(); // 创建作业实例 Job job = Job.getInstance(conf, ReduceCacheDriver.class.getSimpleName()); // 设置作业驱动类 job.setJarByClass(ReduceCacheDriver.class); // 设置作业Mapper reduce 类 job.setMapperClass(ReduceCacheMapper.class); // 设置作业 mapper 阶段输出key value 数据类型, job.setMapOutputKeyClass(Text.class); job.setMapOutputValueClass(NullWritable.class); // 设置作业reducer 阶段输出key value 数据类型,也就是程序最终输出的数据类型 job.setOutputKeyClass(Text.class); job.setOutputValueClass(NullWritable.class); job.setNumReduceTasks(0); job.addCacheFile(new URI("/data/cache/itheima_goods.txt")); // 配置作业的输入数据路径 FileInputFormat.addInputPath(job, new Path(args[0])); // 配置作业的输出数据路径 FileOutputFormat.setOutputPath(job, new Path(args[1])); // 判断输出路径是否存在,如果存在,删除 FileSystem fs = FileSystem.get(conf); if (fs.exists(new Path(args[1]))) { fs.delete(new Path(args[1]), true); } boolean resultFlag = job.waitForCompletion(true); System.exit(resultFlag ? 0 : 1); } }

3. 提交运行
   
   • 在工程的pom.xml文件中指定程序运行的主类全路径；

org.apache.maven.plugins maven-jar-plugin 2.4 true lib/ com.uuicon.sentiment_upload.cache.ReduceCacheDriver 【Hadoop 入门笔记 二十二 : MapReduce实现Join关联操作】org.apache.maven.plugins maven-compiler-plugin 3.1 1.8 1.8 UTF-8

• 执行mvn package命令生成jar包；
  
  文章图片
  
• 将jar包上传到hadoop集群（任意节点上）；
• 执行命令（任意节点上）：hadoop jar xxxx.jar。注意保证yarn集群提前启动成功。

推荐阅读

• 

  祝福婚姻幸福的句子
• 

  aws_cli|aws_cli s3基本使用
• 

  慧定投怎么样？ 蚂蚁聚宝慧定投靠谱吗？
• 

  有哪些免费使用流程图模板的方法？
• 

  林内热水器跳到72度不烧什么意思，燃气热水器调到六十度不够热
• 

  前端|前端 JavaScript运行原理
• 

  夏季养生保健 预防阴暑更重要
• 

  day20，天道
• 

  种植碗莲需要注意什么 蔓莲养殖方法和注意事项
• 

  无症状感染者|陕西新增境外输入确诊病例6例、无症状感染者1例
• 

  公主连结6星需要多少角色碎片 公主连接六星角色碎片数量介绍
• 

  sap生产订单成本明细清单分析
• 

  惠普340G5笔记本键盘失灵的原因是什么
• 

  花粉怎样喂蜜蜂，花粉怎样消毒 花粉如何饲喂蜜蜂
• 

  龟苓膏怎么保存
• 

  ai怎么画一颗心形草莓
• 

  mysql清空所有数据 mysql快速完全清空
• 

  vb.net自动生成属性 vbnet createobject
• 

  18x|魅蓝推出回归首款新品，魅族秋季新品发布会预热
• 

  数据结构|基本排序算法总结（Java实现）

• EffectiveObjective-C2.0|EffectiveObjective-C2.0 笔记 - 第二部分
• Android中的AES加密-下
• 【读书笔记】贝叶斯原理
• 【韩语学习】（韩语随堂笔记整理）
• 【Hadoop踩雷】Mac下安装Hadoop3以及Java版本问题
• 人性的弱点-笔记
• 读书笔记:博登海默法理学|读书笔记:博登海默法理学 —— 正义的探索（1）
• D034+3组苏曼＋《写作这回事》读书笔记
• 《自我的追寻》读书笔记3
• 最有效的时间管理工具（赢效率手册和总结笔记）