本文主要是介绍Map端的Join map-side join，希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！
- map-side join:(最为高效)
- 核心思想：将小表进行分布式缓存，在map-task阶段读取缓存文件数据存储到内存数据结构中，以供reduce阶段连接查找。
- 适用场景：有一个或者多个小表(文件)
- 优点：将小表缓存，可以高效查询；由于在map阶段进行连接，所以将会大大减小map到reduce端的数据传输，从而减少不必要的shuffle耗时，提高整个mr的执行效率
- 缺点：如果业务全是大表不适合
- semi-join(半连接)：
- 核心思想：将大表过滤或者清洗后进行缓存，从而转换为map-side端的join。
导入的包 注意导入长包
以及数据 对应三个文件
/*** 作者:Shishuai* 文件名：MapSideJoinDemo* 时间：2019/9/4 19:11*/package com.mapjoin_reducejoin;import ali.mr.day02.MapSideJoin;
import jdk.nashorn.internal.ir.BaseNode;
import org.apache.hadoop.conf.Configuration;
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.filecache.DistributedCache;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.jboss.netty.util.internal.ConcurrentHashMap;import java.io.BufferedReader;
import java.io.File;
import java.io.FileReader;
import java.io.IOException;
import java.net.URI;
import java.net.URISyntaxException;
import java.util.Map;/*** map-side join:(最为高效)* 核心思想：将小表进行分布式缓存，在map-task阶段读取缓存文件数据存储到内存数据结构中，以供reduce阶段连接查找。* 适用场景：有一个或者多个小表(文件)* 优点：将小表缓存，可以高效查询；由于在map阶段进行连接，所以将会大大减小map到reduce端的数据传输，从而减少不必要的shuffle耗时，提高整个mr的执行效率* 缺点：如果业务全是大表不适合** semi-join(半连接)：* 核心思想：将大表过滤或者清洗后进行缓存，从而转换为map-side端的join。** login：uid sexid logindate1 1 2017-04-17 08:16:202 2 2017-04-15 06:18:203 1 2017-04-16 05:16:244 2 2017-04-14 03:18:205 1 2017-04-13 02:16:256 2 2017-04-13 01:15:207 1 2017-04-12 08:16:348 2 2017-04-11 09:16:209 0 2017-04-10 05:16:50sex：sexMap0 不知道1 男2 女user uname1 小红2 小行3 小通4 小闪5 小镇6 小振7 小秀8 小微9 小懂10 小明11 小刚12 小举13 小黑14 小白15 小鹏16 小习输出：
loginuid sex uname logindate1 男 小红 2017-04-17 08:16:202 女 小行 2017-04-15 06:18:203 男 小通 2017-04-16 05:16:244 女 小闪 2017-04-14 03:18:205 男 小镇 2017-04-13 02:16:256 女 小振 2017-04-13 01:15:207 男 小秀 2017-04-12 08:16:348 女 小微 2017-04-11 09:16:209 不知道 小懂 2017-04-10 05:16:50*** @Author Shishuai* @Email 1198319583@qq.com* @Description //TODO* @Date 19:39 2019/9/4* @Param* @Retrun 这个打包到集群上运行 两个表ur sex 以及login已经上传到hdfs 而且使用的是ha模式 我的端口是默认8020没改 改过的一般是9000**/主要的一个setup和一个map函数
在setup 读取缓存文件 就是两个小表 ur 和 sex 因为就两列 读出来数据存到map中
在map 一行一行的读取login数据，切割后得到id
然后根据id取出对应map的值
public class MapSideJoinDemo {//自定义的mapper类public static class MyMapper extends Mapper<LongWritable, Text, Text, NullWritable>{public Text k = new Text();//读取缓存文件,并按照文件名称读取到map或者别的数据结构中//定义一个存储sex缓存的数据结构Map<String, String> sexMap = new ConcurrentHashMap<String, String>();Map<String, String> userMap = new ConcurrentHashMap<String, String>();//读取缓存在hdfs上的两个表文件//找到这两个缓存文件 将他们放入map中 因为就两列 所以@Overrideprotected void setup(Context context) throws IOException, InterruptedException {Path[] paths = DistributedCache.getLocalCacheFiles(context.getConfiguration());for(Path p : paths){String fileName = p.getName();BufferedReader bufferedReader = null;if(fileName.endsWith("sex")){bufferedReader = new BufferedReader(new FileReader(new File(p.toString())));while(bufferedReader.ready()){String line = bufferedReader.readLine();String sexs[] = line.split("\t");sexMap.put(sexs[0], sexs[1]);}}else if(fileName.equals("ur")){bufferedReader = new BufferedReader(new FileReader(new File(p.toString())));while(bufferedReader.ready()){String line = bufferedReader.readLine();String users[] = line.split("\t");userMap.put(users[0], users[1]);}}if(bufferedReader != null){bufferedReader.close();}}}//抽象map函数 (map阶段的核心业务逻辑)//然后进行map过程 一行一行读入login表中的信息//uid sexid time//1 1 2017-04-17 08:16:20 比如读入这个 切割后根据前边两个id去拿两个map对应的值@Overrideprotected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {String[] words = value.toString().split("\t");String uid = words[0];String sex_id = words[1];String uname = "";String sexlab = "";sexlab = sexMap.getOrDefault(sex_id, "");uname = userMap.getOrDefault(uid, "");this.k.set(uid + "\t" + sexlab + "\t" + uname + "\t" + words[2]);context.write(k, NullWritable.get());}}//驱动方法public static void main(String[] args) {try {//1、获取配置对象并进行属性设置Configuration conf = new Configuration();//对conf设置conf.set("fs.defaultFS", "hdfs://qf");conf.set("dfs.nameservices", "qf");conf.set("dfs.ha.namenodes.qf", "nn1, nn2");conf.set("dfs.namenode.rpc-address.qf.nn1", "hadoop01:8020");conf.set("dfs.namenode.rpc-address.qf.nn2", "hadoop02:8020");conf.set("dfs.client.failover.proxy.provider.qf", "org.apache.hadoop.hdfs.server.namenode.ha.ConfiguredFailoverProxyProvider");//2、获取jobJob job = Job.getInstance(conf, "mapSideJoin");//3、对job设置运行主类job.setJarByClass(MapSideJoinDemo.class);//4、对job的map端属性设置job.setMapperClass(MyMapper.class);job.setMapOutputKeyClass(Text.class);job.setMapOutputValueClass(NullWritable.class);//设置缓存 (缓存文件读取不了)//job.setCacheFiles();job.addCacheFile(new URI("hdfs://qf:8020/sex"));job.addCacheFile(new URI("hdfs://qf:8020/ur"));//6、设置job的输入路径和输出路径FileInputFormat.addInputPath(job, new Path(args[0]));FileOutputFormat.setOutputPath(job, new Path(args[1]));//7、提交作业int success = job.waitForCompletion(true) ? 0 : 1;//8、退出System.exit(success);} catch (IOException e) {e.printStackTrace();} catch (URISyntaxException e) {e.printStackTrace();} catch (InterruptedException e) {e.printStackTrace();} catch (ClassNotFoundException e) {e.printStackTrace();}}
}
[root@hadoop01 join]# yarn jar /home/hadoopDemo-1.0-SNAPSHOT.jar com.mapjoin_reducejoin.MapSideJoinDemo /login /out/03结果文件 没问题
这篇关于Map端的Join map-side join的文章就介绍到这儿，希望我们推荐的文章对编程师们有所帮助！
