Hadoop 多表 join：map side join 范例

2024-03-19 13:58
文章标签 多表 map join hadoop 范例 side

本文主要是介绍Hadoop 多表 join：map side join 范例，希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！

在没有 pig 或者 hive 的环境下，直接在 mapreduce 中自己实现 join 是一件极其蛋疼的事情，MR中的join分为好几种，比如有最常见的 reduce side join，map side join，semi join 等。今天我们要讨论的是第 2 种：map side join，这种 join 在处理多个小表关联大表时非常有用，而 reduce join 在处理多表关联时是比较麻烦的，一次只能处理一张表。

1、原理：

      之所以存在reduce side join，是因为在map阶段不能获取所有需要的join字段，即：同一个key对应的字段可能位于不同map中。但 Reduce side join是非常低效的，因为shuffle阶段要进行大量的数据传输。Map side join是针对以下场景进行的优化：两个待连接表中，有一个表非常大，而另一个表非常小，以至于小表可以直接存放到内存中。这样，我们可以将小表复制多份，让每个map task内存中存在一份（比如存放到hash table中），然后只扫描大表：对于大表中的每一条记录key/value，在hash table中查找是否有相同的key的记录，如果有，则连接后输出即可。为了支持文件的复制，Hadoop提供了一个类DistributedCache，使用该类的方法如下：

（1）用户使用静态方法DistributedCache.addCacheFile()指定要复制的文件，它的参数是文件的URI（如果是HDFS上的文件，可以这样：hdfs://jobtracker:50030/home/XXX/file）。JobTracker在作业启动之前会获取这个URI列表，并将相应的文件拷贝到各个TaskTracker的本地磁盘上。

（2）用户使用DistributedCache.getLocalCacheFiles()方法获取文件目录，并使用标准的文件读写API读取相应的文件。

2、环境：

本实例需要的测试文件及 hdfs 文件存放目录如下：

hadoop fs -ls /test/decli
Found 4 items
-rw-r--r--   2 root supergroup        152 2013-03-06 02:05 /test/decli/login
drwxr-xr-x   - root supergroup          0 2013-03-06 02:45 /test/decli/output
-rw-r--r--   2 root supergroup         12 2013-03-06 02:12 /test/decli/sex
-rw-r--r--   2 root supergroup         72 2013-03-06 02:44 /test/decli/user

测试文件内容分别为：

root@master 192.168.120.236 02:58:03 ~/test/table >
cat login  # 登录表，需要判断 uid 列是否有效，并得到对应用户名、性别、访问次数
1       0       20121213 
2       0       20121213 
3       1       20121213 
4       1       20121213 
1       0       20121114 
2       0       20121114 
3       1       20121114 
4       1       20121114 
1       0       20121213 
1       0       20121114
9       0       20121114
root@master 192.168.120.236 02:58:08 ~/test/table >
cat sex # 性别表
0       男
1       女
root@master 192.168.120.236 02:58:13 ~/test/table >
cat user # 用户属性表
1       张三    hubei 
3       王五    tianjin 
4       赵六    guangzhou 
2       李四    beijing 
root@master 192.168.120.236 02:58:16 ~/test/table >

测试环境 hadoop 版本：

view source
print ?
1echo $HADOOP_HOME
2/work/hadoop-0.20.203.0

好了，废话少说，上代码：

3、代码：

view source
print ?
001import java.io.BufferedReader;
002import java.io.FileReader;
003import java.io.IOException;
004import java.util.HashMap;
005import java.util.Map;
006 
007import org.apache.hadoop.conf.Configuration;
008import org.apache.hadoop.conf.Configured;
009import org.apache.hadoop.filecache.DistributedCache;
010import org.apache.hadoop.fs.Path;
011import org.apache.hadoop.io.LongWritable;
012import org.apache.hadoop.io.Text;
013import org.apache.hadoop.mapreduce.Job;
014import org.apache.hadoop.mapreduce.Mapper;
015import org.apache.hadoop.mapreduce.Reducer;
016import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
017import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
018import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
019import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
020import org.apache.hadoop.util.GenericOptionsParser;
021import org.apache.hadoop.util.Tool;
022import org.apache.hadoop.util.ToolRunner;
023 
024public class MultiTableJoin extends Configured implements Tool {
025    public static class MapClass extends Mapper<LongWritable, Text, Text, Text> {
026 
027        // 用于缓存 sex、user 文件中的数据
028        private Map<String, String> userMap = new HashMap<String, String>();
029        private Map<String, String> sexMap = new HashMap<String, String>();
030 
031        private Text oKey = new Text();
032        private Text oValue = new Text();
033        private String[] kv;
034 
035        // 此方法会在map方法执行之前执行
036        @Override
037        protected void setup(Context context) throws IOException,
038                InterruptedException {
039            BufferedReader in = null;
040 
041            try {
042                // 从当前作业中获取要缓存的文件
043                Path[] paths = DistributedCache.getLocalCacheFiles(context
044                        .getConfiguration());
045                String uidNameAddr = null;
046                String sidSex = null;
047                for (Path path : paths) {
048                    if (path.toString().contains("user")) {
049                        in = new BufferedReader(new FileReader(path.toString()));
050                        while (null != (uidNameAddr = in.readLine())) {
051                            userMap.put(uidNameAddr.split("\t", -1)[0],
052                                    uidNameAddr.split("\t", -1)[1]);
053                        }
054                    } else if (path.toString().contains("sex")) {
055                        in = new BufferedReader(new FileReader(path.toString()));
056                        while (null != (sidSex = in.readLine())) {
057                            sexMap.put(sidSex.split("\t", -1)[0], sidSex.split(
058                                    "\t", -1)[1]);
059                        }
060                    }
061                }
062            } catch (IOException e) {
063                e.printStackTrace();
064            } finally {
065                try {
066                    if (in != null) {
067                        in.close();
068                    }
069                } catch (IOException e) {
070                    e.printStackTrace();
071                }
072            }
073        }
074 
075        public void map(LongWritable key, Text value, Context context)
076                throws IOException, InterruptedException {
077 
078            kv = value.toString().split("\t");
079            // map join: 在map阶段过滤掉不需要的数据
080            if (userMap.containsKey(kv[0]) && sexMap.containsKey(kv[1])) {
081                oKey.set(userMap.get(kv[0]) + "\t" + sexMap.get(kv[1]));
082                oValue.set("1");
083                context.write(oKey, oValue);
084            }
085        }
086 
087    }
088 
089    public static class Reduce extends Reducer<Text, Text, Text, Text> {
090 
091        // private Text oValue = new Text();
092        // private StringBuilder sb;
093 
094        public void reduce(Text key, Iterable<Text> values, Context context)
095                throws IOException, InterruptedException {
096            int sumCount = 0;
097 
098            for (Text val : values) {
099                sumCount += Integer.parseInt(val.toString());
100            }
101 
102            context.write(key, new Text(String.valueOf(sumCount)));
103        }
104 
105    }
106 
107    public int run(String[] args) throws Exception {
108        Job job = new Job(getConf(), "MultiTableJoin");
109 
110        job.setJobName("MultiTableJoin");
111        job.setJarByClass(MultiTableJoin.class);
112        job.setMapperClass(MapClass.class);
113        job.setReducerClass(Reduce.class);
114 
115        job.setInputFormatClass(TextInputFormat.class);
116        job.setOutputFormatClass(TextOutputFormat.class);
117 
118        job.setOutputKeyClass(Text.class);
119        job.setOutputValueClass(Text.class);
120 
121        String[] otherArgs = new GenericOptionsParser(job.getConfiguration(),
122                args).getRemainingArgs();
123 
124        // 我们把第1、2个参数的地址作为要缓存的文件路径
125        DistributedCache.addCacheFile(new Path(otherArgs[1]).toUri(), job
126                .getConfiguration());
127        DistributedCache.addCacheFile(new Path(otherArgs[2]).toUri(), job
128                .getConfiguration());
129 
130        FileInputFormat.addInputPath(job, new Path(otherArgs[3]));
131        FileOutputFormat.setOutputPath(job, new Path(otherArgs[4]));
132 
133        return job.waitForCompletion(true) ? 0 : 1;
134    }
135 
136    public static void main(String[] args) throws Exception {
137        int res = ToolRunner.run(new Configuration(), new MultiTableJoin(),
138                args);
139        System.exit(res);
140    }
141 
142}

运行命令：

view source
print ?
1hadoop jar MultiTableJoin.jar MultiTableJoin /test/decli/sex /test/decli/user /test/decli/login /test/decli/output

4、结果：

运行结果：

root@master 192.168.120.236 02:47:18 ~/test/table >
hadoop fs -cat /test/decli/output/*|column -t
cat: File does not exist: /test/decli/output/_logs
张三  男  4
李四  男  2
王五  女  2
赵六  女  2
root@master 192.168.120.236 02:47:26 ~/test/table >

TIPS：

更多关于 hadoop mapreduce 相关 join 介绍，请参考之前的博文：

MapReduce 中的两表 join 几种方案简介

http://my.oschina.net/leejun2005/blog/95186

这篇关于Hadoop 多表 join：map side join 范例的文章就介绍到这儿，希望我们推荐的文章对编程师们有所帮助！


http://www.chinasem.cn/article/826213。

相关文章

SpringBoot如何通过Map实现策略模式

SpringBoot如何通过Map实现策略模式

《SpringBoot如何通过Map实现策略模式》策略模式是一种行为设计模式,它允许在运行时选择算法的行为,在Spring框架中,我们可以利用@Resource注解和Map集合来优雅地实现策略模式,这... 目录前言底层机制解析Spring的集合类型自动装配@Resource注解的行为实现原理使用直接使用M
阅读更多...
java String.join()的使用小结

java String.join()的使用小结

《javaString.join()的使用小结》String.join()是Java8引入的一个实用方法,用于将多个字符串按照指定分隔符连接成一个字符串,本文主要介绍了javaString.join... 目录1. 方法定义2. 基本用法2.1 拼接多个字符串2.2 拼接集合中的字符串3. 使用场景和示例3
阅读更多...
C++ 各种map特点对比分析

C++ 各种map特点对比分析

《C++各种map特点对比分析》文章比较了C++中不同类型的map（如std::map,std::unordered_map,std::multimap,std::unordered_multima... 目录特点比较C++ 示例代码 ​​​​​​代码解释特点比较1. std::map底层实现：基于红黑
阅读更多...
MySQL中实现多表查询的操作方法(配sql+实操图+案例巩固 通俗易懂版)

MySQL中实现多表查询的操作方法(配sql+实操图+案例巩固 通俗易懂版)

《MySQL中实现多表查询的操作方法(配sql+实操图+案例巩固通俗易懂版)》本文主要讲解了MySQL中的多表查询,包括子查询、笛卡尔积、自连接、多表查询的实现方法以及多列子查询等,通过实际例子和操... 目录复合查询1. 回顾查询基本操作group by 分组having1. 显示部门号为10的部门名，员
阅读更多...
JavaScript中的Map用法完全指南

JavaScript中的Map用法完全指南

《JavaScript中的Map用法完全指南》：本文主要介绍JavaScript中Map用法的相关资料,通过实例讲解了Map的创建、常用方法和迭代方式,还探讨了Map与对象的区别,并通过一个例子展... 目录引言1. 创建 Map2. Map 和对象的对比3. Map 的常用方法3.1 set(key, v
阅读更多...
Golang中map缩容的实现

Golang中map缩容的实现

《Golang中map缩容的实现》本文主要介绍了Go语言中map的扩缩容机制,包括grow和hashGrow方法的处理,具有一定的参考价值,感兴趣的可以了解一下... 目录基本分析带来的隐患为什么不支持缩容基本分析在 Go 底层源码 src/runtime/map.go 中，扩缩容的处理方法是 grow
阅读更多...
Go语言利用泛型封装常见的Map操作

Go语言利用泛型封装常见的Map操作

《Go语言利用泛型封装常见的Map操作》Go语言在1.18版本中引入了泛型,这是Go语言发展的一个重要里程碑,它极大地增强了语言的表达能力和灵活性,本文将通过泛型实现封装常见的Map操作,感... 目录什么是泛型泛型解决了什么问题Go泛型基于泛型的常见Map操作代码合集总结什么是泛型泛型是一种编程范式，允
阅读更多...
数据库使用之union、union all、各种join的用法区别解析

数据库使用之union、union all、各种join的用法区别解析

《数据库使用之union、unionall、各种join的用法区别解析》：本文主要介绍SQL中的Union和UnionAll的区别,包括去重与否以及使用时的注意事项,还详细解释了Join关键字,... 目录一、Union 和Union All1、区别：2、注意点：3、具体举例二、Join关键字的区别&php
阅读更多...
JSON字符串转成java的Map对象详细步骤

JSON字符串转成java的Map对象详细步骤

《JSON字符串转成java的Map对象详细步骤》：本文主要介绍如何将JSON字符串转换为Java对象的步骤,包括定义Element类、使用Jackson库解析JSON和添加依赖,文中通过代码介绍... 目录步骤 1: 定义 Element 类步骤 2: 使用 Jackson 库解析 jsON步骤 3: 添
阅读更多...
Java中List转Map的几种具体实现方式和特点

Java中List转Map的几种具体实现方式和特点

《Java中List转Map的几种具体实现方式和特点》：本文主要介绍几种常用的List转Map的方式,包括使用for循环遍历、Java8StreamAPI、ApacheCommonsCollect... 目录前言1、使用for循环遍历：2、Java8 Stream API：3、Apache Commons
阅读更多...

Copyright China编程 23002807@qq.com

沪ICP备2023033072号-2