文档章节

Hadoop 中的两表join

超人学院
 超人学院
发布于 2015/02/06 15:03
字数 2077
阅读 109
收藏 0

作为数据分析中经常进行的join 操作,传统DBMS 数据库已经将各种算法优化到了极致,而对于hadoop 使用的mapreduce 所进行的join 操作,去年开始也是有各种不同的算法论文出现,讨论各种算法的适用场景和取舍条件,本文讨论hive 中出现的几种join 优化,然后讨论其他算法实现,希望能给使用hadoop 做数据分析的开发人员提供一点帮助.

 

Facebook 今年在yahoo 的hadoop summit 大会上做了一个关于最近两个版本的hive 上所做的一些join 的优化,其中主要涉及到hive 的几个关键特性: 值分区 , hash 分区 , map join , index ,

 

Common Join

最为普通的join策略,不受数据量的大小影响,也可以叫做reduce side join ,最没效率的一种join 方式. 它由一个mapreduce job 完成.

首先将大表和小表分别进行map 操作, 在map shuffle 的阶段每一个map output key 变成了table_name_tag_prefix + join_column_value , 但是在进行partition 的时候它仍然只使用join_column_value 进行hash.

每一个reduce 接受所有的map 传过来的split , 在reducce 的shuffle 阶段,它将map output key 前面的table_name_tag_prefix 给舍弃掉进行比较. 因为reduce 的个数可以由小表的大小进行决定,所以对于每一个节点的reduce 一定可以将小表的split 放入内存变成hashtable. 然后将大表的每一条记录进行一条一条的比较.

Map Join

Map Join 的计算步骤分两步,将小表的数据变成hashtable广播到所有的map 端,将大表的数据进行合理的切分,然后在map 阶段的时候用大表的数据一行一行的去探测(probe) 小表的hashtable. 如果join key 相等,就写入HDFS.

map join 之所以叫做map join 是因为它所有的工作都在map 端进行计算.

hive 在map join 上做了几个优化:

  • hive 0.6 的时候默认认为写在select 后面的是大表,前面的是小表, 或者使用 /*+mapjoin(map_table) */ 提示进行设定. hive 0.7 的时候这个计算是自动化的,它首先会自动判断哪个是小表,哪个是大表,这个参数由(hive.auto.convert.join=true)来控制. 然后控制小表的大小由(hive.smalltable.filesize=25000000L)参数控制(默认是25M),当小表超过这个大小,hive 会默认转化成common join. 你可以查看HIVE-1642.

    首先小表的Map 阶段它会将自己转化成MapReduce Local Task ,然后从HDFS 取小表的所有数据,将自己转化成Hashtable file 并压缩打包放入DistributedCache 里面.

    目前hive 的map join 有几个限制,一个是它打算用BloomFilter 来实现hashtable , BloomFilter 大概比hashtable 省8-10倍的内存, 但是BloomFilter 的大小比较难控制.

    现在DistributedCache 里面hashtable默认的复制是3份,对于一个有1000个map 的大表来说,这个数字太小,大多数map 操作都等着DistributedCache 复制.

     

    Bucket Map Join

    hive 建表的时候支持hash 分区通过指定clustered by (col_name,xxx ) into number_buckets buckets 关键字.

    当连接的两个表的join key 就是bucket column 的时候,就可以通过

    hive.optimize.bucketmapjoin= true

    来控制hive 执行bucket map join 了, 需要注意的是你的小表的number_buckets 必须是大表的倍数. 无论多少个表进行连接这个条件都必须满足.(其实如果都按照2的指数倍来分bucket, 大表也可以是小表的倍数,不过这中间需要多计算一次,对int 有效,long 和string 不清楚)

    Bucket Map Join 执行计划分两步,第一步先将小表做map 操作变成hashtable 然后广播到所有大表的map端,大表的map端接受了number_buckets 个小表的hashtable并不需要合成一个大的hashtable,直接可以进行map 操作,map 操作会产生number_buckets 个split,每个split 的标记跟小表的hashtable 标记是一样的, 在执行projection 操作的时候,只需要将小表的一个hashtable 放入内存即可,然后将大表的对应的split 拿出来进行判断,所以其内存限制为小表中最大的那个hashtable 的大小.

    Bucket Map Join 同时也是Map Side Join 的一种实现,所有计算都在Map 端完成,没有Reduce 的都被叫做Map Side Join ,Bucket 只是hive 的一种hash partition 的实现,另外一种当然是值分区.

    create table a  (xxx) partition by (col_name)

    不过一般hive 中两个表不一定会有同一个partition key, 即使有也不一定会是join key. 所以hive 没有这种基于值的map side join, hive 中的list partition 主要是用来过滤数据的而不是分区. 两个主要参数为(hive.optimize.cp = true 和 hive.optimize.pruner=true)

    hadoop 源代码中默认提供map side join 的实现, 你可以在hadoop 源码的src/contrib/data_join/src 目录下找到相关的几个类.  其中TaggedMapOutput 即可以用来实现hash 也可以实现list , 看你自己决定怎么分区. Hadoop Definitive Guide 第8章关于map side join 和side data distribution 章节也有一个例子示例怎样实现值分区的map side join.

     

    Sort Merge Bucket Map Join

    Bucket Map Join 并没有解决map join 在小表必须完全装载进内存的限制, 如果想要在一个reduce 节点的大表和小表都不用装载进内存,必须使两个表都在join key 上有序才行,你可以在建表的时候就指定sorted by join key 或者使用index 的方式.

    set hive.optimize.bucketmapjoin = true;

    set hive.optimize.bucketmapjoin.sortedmerge = true;

    set hive.input.format=org.apache.hadoop.hive.ql.io.BucketizedHiveInputFormat;

    Bucket columns == Join columns == sort columns

    这样小表的数据可以每次只读取一部分,然后还是用大表一行一行的去匹配,这样的join 没有限制内存的大小. 并且也可以执行全外连接.

     

    Skew Join

    真实数据中数据倾斜是一定的, hadoop 中默认是使用

    hive.exec.reducers.bytes.per.reducer = 1000000000

    也就是每个节点的reduce 默认是处理1G大小的数据,如果你的join 操作也产生了数据倾斜,那么你可以在hive 中设定

    set hive.optimize.skewjoin = true; 
    set hive.skewjoin.key = skew_key_threshold (default = 100000)

    hive 在运行的时候没有办法判断哪个key 会产生多大的倾斜,所以使用这个参数控制倾斜的阈值,如果超过这个值,新的值会发送给那些还没有达到的reduce, 一般可以设置成你

    (处理的总记录数/reduce个数)的2-4倍都可以接受.

    倾斜是经常会存在的,一般select 的层数超过2层,翻译成执行计划多于3个以上的mapreduce job 都很容易产生倾斜,建议每次运行比较复杂的sql 之前都可以设一下这个参数. 如果你不知道设置多少,可以就按官方默认的1个reduce 只处理1G 的算法,那么  skew_key_threshold  = 1G/平均行长. 或者默认直接设成250000000 (差不多算平均行长4个字节)

     

    Left Semi Join

    hive 中没有in/exist 这样的子句,所以需要将这种类型的子句转成left semi join. left semi join 是只传递表的join key给map 阶段 , 如果key 足够小还是执行map join, 如果不是则还是common join.

     

     

    join 策略中的难点

    大多数只适合等值连接(equal join) ,

    范围比较和全外连接没有合适的支持

    提前分区,零时分区,排序,多种不同执行计划很难评价最优方案.

    没有考虑IO 比如临时表,网络消耗和网络延迟时间,CPU时间,

    最优的方案不代表系统资源消耗最少.

© 著作权归作者所有

共有 人打赏支持
超人学院
粉丝 110
博文 335
码字总数 388917
作品 0
昌平
CTO(技术副总裁)
私信 提问
Apache Spark探秘:实现Map-side Join和Reduce-side Join

Apache Spark探秘:实现Map-side Join和Reduce-side JoinCategory: Frameworks (Spark) On YARN View: 13,756 Author: Dong 作者:Dong | 新浪微博:西成懂 | 可以转载, 但必须以超链接形式标...

片刻
2015/12/23
92
0
大数据教程(9.5)用MR实现sql中的jion逻辑

上一篇博客讲解了使用jar -jar的方式来运行提交MR程序,以及通过修改YarnRunner的源码来实现MR的windows开发环境提交到集群的方式。本篇博主将分享sql中常见的join操作。 一、需求 订单数据表...

em_aaron
前天
0
0
Hive 中的 LEFT SEMI JOIN 与 JOIN ON 的前世今生

hive 的 join 类型有好几种,其实都是把 MR 中的几种方式都封装实现了,其中 join on、left semi join 算是里边具有代表性,且使用频率较高的 join 方式。 1、联系 他们都是 hive join 方式的...

大数据之路
2013/12/28
0
3
MapReduce 中的两表 join 几种方案简介

1. 概述 在传统数据库(如:MYSQL)中,JOIN操作是非常常见且非常耗时的。而在HADOOP中进行JOIN操作,同样常见且耗时,由于Hadoop的独特设计思想,当进行JOIN操作时,有一些特殊的技巧。 本文...

恶魔苏醒ing
2016/10/19
0
0
MapReduce 中的两表 join 几种方案简介

概述 在传统数据库(如:MYSQL)中,JOIN操作是非常常见且非常耗时的。而在HADOOP中进行JOIN操作,同样常见且耗时,由于Hadoop的独特设计思想,当进行JOIN操作时,有一些特殊的技巧。 本文首...

大数据之路
2012/12/11
0
8

没有更多内容

加载失败,请刷新页面

加载更多

一、什么是ActiveMQ

首先我们应该先了解J2EE中的一个重要规范:JMS(The Java Message Service)Java消息服务。而JMS的客户端之间可以通过JMS服务进行异步的消息传输。它主要有两种模型:点对点和发布订阅模型。 ...

watermelon11
17分钟前
0
0
课时17 第三课Spark内部原理剖析与源码阅读(五)

为何spark shuffle比mapreduce shuffle慢? 主要是spark shuffle的shuffle read阶段还不够优秀,它是基于hashmap实现的,shuffle read会把shuffel write阶段已经排序数据给重新转成乱序的,转...

刀锋
33分钟前
1
0
Function函数式接口

Function函数式接口传入一个参数,返回一个值。 然后我们使用这个写个demo看看: 输出: 接口内部还有两个default方法和一个static方法,然后我们先看一下static方法 返回一个始终返回其输入...

woshixin
48分钟前
1
0
开发者和架构师之间最大的区别是什么?

1、开发者和架构师之间最大的区别是什么? 架构师和开发者一样,也经常写代码,简单的说,开发者和架构师之间最大的区别就是技术领导力。 软件架构师的角色需要理解最重要的架构驱动力是什么...

James-
今天
2
0
java框架学习日志-4

补充一些spring配置文件的方法。 设置别名: <!--通过name直接设置别名--> <bean name="user2" class="cn.sxt.factory.UserDynamicFactory"> </bean> <!--有id的情况下也可以设置......

白话
今天
2
0

没有更多内容

加载失败,请刷新页面

加载更多

返回顶部
顶部