Map Join

使用场景

Map Join 适用于一张表十分小、一张表很大的场景。

优点

思考:在 Reduce 端处理过多的表，非常容易产生数据倾斜。怎么办?
在 Map 端缓存多张表，提前处理业务逻辑，这样增加 Map 端业务，减少 Reduce 端数
据的压力，尽可能的减少数据倾斜。

具体办法:采用 DistributedCache

在 Mapper 的 setup 阶段，将文件读取到缓存集合中；
在 Driver 驱动类中加载缓存。

缓存普通文件到 Task 运行节点：

job.addCacheFile(new URI("file:///e:/cache/pd.txt"));

如果是集群运行,需要设置 HDFS 路径：

job.addCacheFile(new URI("hdfs://hadoop102:8020/cache/pd.txt"));

Map Join 案例

需求

：同Reduce Join案例；

需求分析

MapJoin 适用于关联表中有小表的情形。

Map端表合并案例分析(Distributedcache)

源码

MapJoinMapper类

package com.xiaobai.mapreduce.mapjoin;import org.apache.commons.lang3.StringUtils;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IOUtils;
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.IOException;
import java.io.InputStreamReader;
import java.net.URI;
import java.util.HashMap;public class MapJoinMapper extends Mapper<LongWritable, Text,Text, NullWritable> {private HashMap<String, String> pdMap = new HashMap<>();private Text outK = new Text();@Overrideprotected void setup(Context context) throws IOException, InterruptedException {//获取缓存的文件，并把文件内容封装到集合 pd.txtURI[] cacheFiles = context.getCacheFiles();FileSystem fs = FileSystem.get(context.getConfiguration());FSDataInputStream fis = fs.open(new Path(cacheFiles[0]));//从流中读取数据BufferedReader reader = new BufferedReader(new InputStreamReader(fis, "UTF-8"));String line;while(StringUtils.isNotEmpty(line = reader.readLine())){//切割String[] fields = line.split("\t");//赋值pdMap.put(fields[0],fields[1]);}//关流IOUtils.closeStream(reader);}@Overrideprotected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {//处理order.txtString line = value.toString();String[] fields = line.split("\t");//获取pidString pname = pdMap.get(fields[1]);//获取订单id和订单数量//封装outK.set(fields[0] + "\t" + pname + "\t" + fields[2]);context.write(outK,NullWritable.get());}
}

MapJoinDriver类

package com.xiaobai.mapreduce.mapjoin;import org.apache.hadoop.conf.Configuration;
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.io.IOException;
import java.net.URI;
import java.net.URISyntaxException;public class MapJoinDriver {public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException, URISyntaxException {//1. 获取job信息Configuration conf = new Configuration();Job job = Job.getInstance(conf);//2. 设置加载jar包路径job.setJarByClass(MapJoinDriver.class);//3. 关联mapperjob.setMapperClass(MapJoinMapper.class);//4.设置Map输出kv类型job.setMapOutputKeyClass(Text.class);job.setMapOutputValueClass(NullWritable.class);//5.设置最终输出kv类型job.setOutputKeyClass(Text.class);job.setOutputValueClass(NullWritable.class);//加载缓存数据job.addCacheFile(new URI("/Users/jane/Desktop/test/JoinTest/pd.txt"));//Map端join的逻辑不需要reduce阶段，设置reduceTask数量为0job.setNumReduceTasks(0);//6. 设置输入输出路径FileInputFormat.setInputPaths(job,new Path("/Users/jane/Desktop/test/JoinTest"));FileOutputFormat.setOutputPath(job,new Path("/Users/jane/Desktop/hadoop/MapJoinTestOutput"));//7. 提交boolean b = job.waitForCompletion(true);System.exit(b ? 0 : 1);}
}

hadoop--Map Join相关推荐

Map Join介绍及案例
Map Join介绍及案例 Map Join介绍 1. 使用场景 2. 优点 3. 实现方法 Map Join案例 1. 需求 (1)需求说明 (2)文件 2.案例分析 (1)需求分析 (2)输入数据 ...
Hadoop之Join、计数器、数据清洗概述
Hadoop之Join.计数器.数据清洗概述目录 Reduce join Map join 计数器应用数据清洗(ETL) 1. Reduce join 原理 Map端的主要工作:为来自不同表(文件 ...
MapReduce之Map join操作
MapReduce之Map join操作(分布式缓存) 文章目录 MapReduce之Map join操作(分布式缓存) 案例结合利用MapReduce中的setup方法与DistributedCa ...
MR实现reduce join和map join及hive的执行计划
一.涵盖 MapReduce InputFormat RecordReader 切片:block=input split 1.1 File- Text- NLine- DB- Mapper setup ...
关于hive中Map join 时大表left join小表的问题
在hive中,(启用Map join时) 大表left join小表,加载从右向左,所以小表会加载进内存,存储成map键值对,通过大表驱动小表,来进行join,即大表中的join字段作为key 来获取 ...
Hadoop Map/Reduce教程
Hadoop Map/Reduce教程目的先决条件概述输入与输出例子:WordCount v1.0 源代码用法 ...
一步一步跟我学习hadoop(5)----hadoop Map/Reduce教程（2）
Map/Reduce用户界面本节为用户採用框架要面对的各个环节提供了具体的描写叙述,旨在与帮助用户对实现.配置和调优进行具体的设置.然而,开发时候还是要相应着API进行相关操作. 首先我们须要了解M ...
Hive的Map Join与Common Join
笼统的说,Hive中的Join可分为Common Join(Reduce阶段完成join)和Map Join(Map阶段完成join). 一.Hive Common Join 如果不指定MapJoin ...
eclipse的plugins导入hadoop-eclipse-plugin-2.6.0.jar后Preference下没有hadoop Map/Reduce的解决方法
参考文章:eclipse下的plugins导入hadoop-eclipse-plugin-2.7.1.jar,Preference下没有hadoop Map/Reduce的解决方法这种现象一般是由于 ...
Hive中的map join、left semi join和sort merge bucket join
map join map join是将join双方比较小的表直接分发到各个 map进程的内存中,在map进程中进行join操作,这样就不用进行reduce步骤,从而提高了速度. 如果不指定mapjoi ...

hadoop--Map Join

目录