几种 hive join 类型简介

做为数据分析中常常进行的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 没有限制内存的大小. 而且也能够执行全外链接.

 例子参考：http://superlxw1234.iteye.com/blog/1545150

五、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时间,

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

七、参考资料:

[1] Join Strategy in Hive

https://cwiki.apache.org/confluence/display/Hive/Presentations 

[2] Join Optimization

https://cwiki.apache.org/Hive/joinoptimization.html

[3] 官方文档：

http://hive.apache.org/docs/r0.9.0/language_manual/joins.html

[4] [一块儿学Hive]之十一-Hive中Join的类型和用法

http://superlxw1234.iteye.com/blog/2222049 

[5] [一块儿学Hive]之十-Hive中Join的原理和机制

http://superlxw1234.iteye.com/blog/2221930

[6] 你真的了解Join吗？

http://www.jianshu.com/p/47db8ac001ea

[7] SparkSQL – 有必要坐下来聊聊Join

http://hbasefly.com/2017/03/19/sparksql-basic-join/