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

结果文件 没问题