幾種 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/