帶你探尋 Hive 背后的執(zhí)行機制和原理
前言
我們都知道，Hive SQL 實際上是翻譯為 MapReduce 執(zhí)行的， 那么它具體過程如何呢？今天我們就來探尋一下 Hive SQL 背后的執(zhí)行機制和原理。
要知道，進(jìn)一步理解和掌握 Hive SQL 的執(zhí)行原理對于平時離線任務(wù)的開發(fā)和優(yōu)化非常重要，直接關(guān)系到 Hive SQL 的執(zhí)行效率和時間。
一、Hive 基本架構(gòu)
作為基于 Hadoop 主要數(shù)據(jù)倉庫解決方案， Hive SQL 是主要的交互接口，實際的數(shù)據(jù)保存在 HDFS 文件中，真正的計算和執(zhí)行則由 MapReduce 完成，它們之間的橋梁是 Hive 引擎。
Hive 主要組件包括 UI 組件、 Driver 組件（ Complier Optimizer Executor ）、 Metastore組件、 CLI ( Command Line Interface ，命令行接口）、 JDBC/ODBC 、Thrift Server 和 Hive Web Interface (HWI ）等。
Hive 就是通過 CLI 、JDBC / ODBC 或者 HWI 接收相關(guān)的 Hive SQL 查詢，并通過 Driver 組件進(jìn)行編譯，分析優(yōu)化，最后變成可執(zhí)行的 MapReduce。
二、Hive SQL
Hive SQL 是類似于 ANSI SQL 標(biāo)準(zhǔn)的SQL 語言，但兩者又不完全相同。Hive SQL 和 MySQL 的 SQL 語言最為接近，但兩者之間也存在顯著差異，比如 Hive 不支持行級數(shù)據(jù)插人、更新和刪除，也不支持事務(wù)等。
Hive 關(guān)鍵概念
1. Hive 數(shù)據(jù)庫
Hive 中的數(shù)據(jù)庫從本質(zhì)上來說僅僅是一個目錄或者命名空間，但是對于具有很多用戶和組的集群來說，這個概念非常有用 。
首先，這樣可以避免表命名沖突；其次，它等同于關(guān)系型數(shù)據(jù)庫中的數(shù)據(jù)庫概念，是一組表或者表的邏輯組，非常容易理解。
2. Hive 表
Hive 中的表（ Table ）和關(guān)系數(shù)據(jù)庫中的 table 在概念上是類似的，每個 table 在 Hive 中都有一個相應(yīng)的目錄存儲數(shù)據(jù)，如果沒有指定表的數(shù)據(jù)庫，那么 Hive 會通過｛HIVE_HOME} /conf/hive-site.xml 配置文件中的 hive.metastore.warehouse.dir 屬性來使用默認(rèn)值（一般是 /user/hive/warehouse ，也可以根據(jù)實際的情況來修改這個配置），所有的 table 數(shù)據(jù)（不包括外部表) 都保存在這個目錄中。
Hive 表分為兩類，即內(nèi)部表和外部表。所謂內(nèi)部表（managed table) 即 Hive 管理的表，Hive 內(nèi)部表的管理既包含邏輯以及語法上的，也包含實際物理意義上的，即創(chuàng)建 Hive 內(nèi)部表時，數(shù)據(jù)將真實存在于表所在的目錄內(nèi)，刪除內(nèi)部表時，物理數(shù)據(jù)和文件也一并刪除。
「那么到底是選擇內(nèi)部表還是外部表呢？」
大多數(shù)情況下，這兩者的區(qū)別不是很明顯。如果數(shù)據(jù)的所有處理都在 Hive 中進(jìn)行，那么更傾向于選擇內(nèi)部表。但是如果 Hive 和其他工具針對相同的數(shù)據(jù)集做處理，那么外部表更合適。
一種常見的模式是使用外部表訪問存儲的 HDFS （通常由其他工具創(chuàng)建）中的初始數(shù)據(jù)，然后使用 Hive 轉(zhuǎn)換數(shù)據(jù)并將其結(jié)果放在內(nèi)部表中。相反，外部表也可以用于將 Hive 的處理結(jié)果導(dǎo)出供其他應(yīng)用使用。
使用外部表的另一種場景是針對一個數(shù)據(jù)集，關(guān)聯(lián)多個 Schema。
3. 分區(qū)和桶
Hive 將表劃分為分區(qū)（partition)，partition 根據(jù)分區(qū)字段進(jìn)行。分區(qū)可以讓數(shù)據(jù)的部分查詢變得更快 。表或者分區(qū)可以進(jìn)一步被劃分為桶（ bucket）。桶通常在原始數(shù)據(jù)中加入一些額外的結(jié)構(gòu)，這些結(jié)構(gòu)可以用于高效查詢。
例如 ，基于用戶 ID 的分桶可以使基于用戶的查詢非常快。
( 1 ）分區(qū)
假設(shè)日志數(shù)據(jù)中，每條記錄都帶有時間戳 。如果根據(jù)時間來分區(qū)，那么同一天的數(shù)據(jù)將被劃分到同一個分區(qū)中。
分區(qū)可以通過多個維度來進(jìn)行。例如，通過日期劃分之后，還可以根據(jù)國家進(jìn)一步劃分。
分區(qū)在創(chuàng)建表的時候使用 PARTITIONED BY 從句定義，該從句接收一個字段列表：
CREATE?TABLE?logs?(ts?BIGINT?,?line?STRING)
PARTITIONED?BY?(dt?STRING,country?STRING);
當(dāng)導(dǎo)入數(shù)據(jù)到分區(qū)表時，分區(qū)的值被顯式指定：
LOAD?DATA?INPATH?’/user/root/path’?
INTO?TABLE?logs?
PARTITION?(dt='2001-01-01',country='GB’);
實際 SQL 中，靈活指定分區(qū)將大大提高其效率，如下代碼將僅會掃描 2001-01-01下的 GB 目錄。
SELECT?ts?,?dt?,?line?FROM?logs?WHERE?dt=‘2001-01-01'?and?country='GB'?
( 2 ）分桶
在表或者分區(qū)中使用桶通常有兩個原因：
一是為了高效查詢 。桶在表中加入了特殊的結(jié)果， Hive 在查詢的時候可以利用這些結(jié)構(gòu)提高效率。例如，如果兩個表根據(jù)相同的字段進(jìn)行分桶，則在對這兩個表進(jìn)行關(guān)聯(lián)的時候，可以使用 map-side 關(guān)聯(lián)高效實現(xiàn)，前提是關(guān)聯(lián)的字段在分桶字段中出現(xiàn)。 二是可以高效地進(jìn)行抽樣。在分析大數(shù)據(jù)集時，經(jīng)常需要對部分抽樣數(shù)據(jù)進(jìn)行觀察和分析，分桶有利于高效實現(xiàn)抽樣。
為了讓 Hive 對表進(jìn)行分桶，通過 CLUSTERED BY 從句在創(chuàng)建表的時候指定：
CREATE?TABLE?bucketed?users(id?INT,?name?STRING)?
CLUSTERED?BY?(id)?INTO?4?BUCKETS;
指定表根據(jù) id 字段進(jìn)行分桶，并且分為 4 個桶 。分桶時， Hive 根據(jù)字段哈希后取余數(shù)來決定數(shù)據(jù)應(yīng)該放在哪個桶，因此每個桶都是整體數(shù)據(jù)的隨機抽樣。
在 map-side 的關(guān)聯(lián)中，兩個表根據(jù)相同的宇段進(jìn)行分桶，因此處理左邊表的 bucket 時，可以直接從外表對應(yīng)的 bucket 中提取數(shù)據(jù)進(jìn)行關(guān)聯(lián)操作。map-side 關(guān)聯(lián)的兩個表不一定需要完全相同 bucket 數(shù)量，只要成倍數(shù)即可。
需要注意的是， Hive 并不會對數(shù)據(jù)是否滿足表定義中的分桶進(jìn)行校驗，只有在查詢時出現(xiàn)異常才會報錯 。因此，一種更好的方式是將分桶的工作交給 Hive 來完成（設(shè) hive.enforce.bucketing 屬性為 true 即可）。
Hive DDL
1. 創(chuàng)建表
CREATE TABLE：用于創(chuàng)建一個指定名字的表 。如果相同名字的表已經(jīng)存在，則拋出異常 用戶可以用 IF NOT EXIST 選項來忽略這個異常。 EXTERNAL ：該關(guān)鍵字可以讓用戶創(chuàng)建一個外部表，在創(chuàng)建表的同時指定一個指向?qū)嶋H數(shù)據(jù)的路徑（LOCATION）。 COMMENT ：可以為表與字段增加描述。 ROW FORMAT ：用戶在建表的時候可以自定義 SerDe 或者使用自帶的 SerDe。 STORED AS ：如果文件數(shù)據(jù)是純文本，則使用 STORED AS TEXTFILE ；如果數(shù)據(jù)需要壓縮， 則使用 STORED AS SEQUENCE 。 LIKE： 允許用戶復(fù)制現(xiàn)有的表結(jié)構(gòu)，但是不復(fù)制數(shù)據(jù)。
hive>?CREATE?TABLE?empty?key?value?store?
LIKE?key?value?store;
還可以通過 CREATE TABLE AS SELECT 的方式來創(chuàng)建表，示例如下：
Hive>?CREATE?TABLE?new?key?value?store?
?ROW?FORMAT?
SERDE?"org.apache.Hadoop.hive.serde2.columnar.ColumnarSerDe"?
?STORED?AS?RCFile?
?AS?
SELECT?(key?%?1024)?new_key,?concat(key,?value)?key_value_pair?
FROM?key_value_store?
SORT?BY?new_key,?key_value_pair;
2. 修改表
修改表名的語法如下：
hive>?ALTER?TABLE?old_table_name?RENAME?TO?new_table_name;
修改列名的語法如下：
ALTER?TABLE?table_name?CHANGE?(COLUMN)?old_col_name?new_col_name?column_type?
[COMMENT?col_comment)?(FIRST|AFTER?column_name)
上述語法允許改變列名 數(shù)據(jù)類型 注釋 列位 它們的任意組合 建表后如果要新增一列，則使用如下語法：
hive>?ALTER?TABLE?pokes?ADD?COLUMNS?(new_col?INT?COMMENT?'new?col?comment');
3. 刪除表
DROP TABLE 語句用于刪除表的數(shù)據(jù)和元數(shù)據(jù) 。對于外部表，只刪除 Metastore 中的元數(shù)據(jù)，而外部數(shù)據(jù)保存不動，示例如下：
drop?table?my_table;
如果只想刪除表數(shù)據(jù)，保留表結(jié)構(gòu)，跟 MySQL 類似，使用 TRUNCATE 語句：
TRUNCATE?TABLE?my_table;
4. 插入表
( 1 ）向表中加載數(shù)據(jù)
相對路徑的示例如下：
hive>?LOAD?DATA?LOCAL?INPATH?’./exarnples/files/kvl.txt?’?OVERWRITE?INTO?
TABLE?pokes;
( 2 ）將查詢結(jié)果插入 Hive
將查詢結(jié)果寫入 HDFS 文件系統(tǒng)。
INSERT?OVERWRITE?TABLE?tablenamel?[PARTITION?(partcoll=val1,?partcol2=val2?...?)]?
select_statement1?FROM?from_statement
這是基礎(chǔ)模式，還有多插入模式和自動分區(qū)模式，這里就不再敘述。
Hive DML
1. 基本的 select 操作
SELECT?[ALL?|?DISTINCT]?select_expr,?select_expr,?...
FROM?table_reference?
[WHERE?where_condition]?
[GROUP?BY?col_list?[HAVING?condition]]?
[?CLUSTER?BY?col_list?
|?[DISTRIBUTE?BY?col_list]?[SORT?BY?|?ORDER?BY?col_list]?
]
[LIMIT?number]
使用 ALL、 DISTINCT 選項區(qū)分對重復(fù)記錄的處理 。默認(rèn)是 ALL ，表示查詢所有記錄， DISTINCT 表示去掉重復(fù)的記錄 WHERE 條件：類似于傳統(tǒng) SQL的 where 條件，支持 AND 、OR 、BETWEEN、 IN 、NOT IN 等 ORDER BY 與 SORT BY 的不同：ORDER BY 指全局排序，只有一個 Reduce 任務(wù)，而 SORT BY 只在本機做排序 LIMIT ：可以限制查詢的記錄數(shù)，如 SELECT * FROM tl LJMIT5 ，也可以實現(xiàn) Topk 查詢，比如下面的查詢語句可以查詢銷售記錄最多的 5個銷售代表
SET?mapred.reduce.tasks?=?1?
SELECT?*?FROM?test?SORT?BY?amount?DESC?LIMIT?5
REGEX Column Specification : select 語句可以使用正則表達(dá)式做列選擇，下面的語句查詢除了 ds 和 hr 之外的所有列
SELECT?`(ds|hr)?＋.＋`?FROM?test
2. join 表
join_table:
table_reference?(INNER]?JOIN?table_factor?(join_condition]
|?table_reference?{LEFTIRIGHTjFULL}?(OUTER]?JOIN?table_reference?join_?condition
|?table_reference?LEFT?SEM?JOIN?table_reference?join_condition
|?table_reference?CROSS?JOIN?table_reference?(join_condition]?(as?of?Hive?0.10)
table?reference:
table_factor
|?join_table
table_factor:
tbl_name?[alias]
|?table_subquery?alias
|?(table_references)
join_condition:
on?expression
Hive中只支持等值連接，外連接和左半連接（left semi join），（從2.2.0版本后支持非等值連接）; 可以連接2個以上的表，如：
select?a.val,?b.val,c.val?
from?a?
join?b?
on?(a.key=b.key1)?
join?c?
on(c.key?=?b.key2);
如果連接中多個表的join key是同一個,則連接會被轉(zhuǎn)化為單個Map/Reduce任務(wù)
select?a.val,b.val,c.val?
from?a?
join?b?
on?(a.key=b.key1)?
join?c?
on(c.key=b.key1);
join時大表放在最后: Reduce會緩存join序列中除最后一個表之外的所有表的記錄，再通過最后一個表將結(jié)果序列化到文件系統(tǒng) 如果想限制join的輸出, 應(yīng)該在where子句中寫過濾條件，或是在join子句中寫。 但是有表分區(qū)的情況，比如下面的第一個 SQL 語句所示，如果d表中找不到對應(yīng)c表的記錄， d表的所有列都會列出 NULL ，包括 ds列。也就是說， join 會過濾d表中不能找到匹配c表 join key 的所有記錄。這樣， LEFT OUTER 就使得查詢結(jié)果與 WHERE 子句無關(guān)，解決辦法是在join 時指定分區(qū)（如下面的第二個 SQL 語句所示）
--第一個?SQL?語句
SELECT?c.val,?d.val?FROM?c?LEFT?OUTER?JOIN?d?ON?(c.key=d.key)?
WHERE?a.ds='2010-07-07'?AND?b.ds='2010-07-07'
--?第二個?SQL?語句
SELECT?c.val,?d.val?FROM?c?LEFT?OUTER?JOIN?d?
ON?(c.key=d.key?AND?d.ds=’?2009-07-07?’?AND?c.ds='2009-07-07')
left semi join是in/exists子查詢的一種更高效的實現(xiàn)，join子句中右邊的表只能在on子句中設(shè)置過濾條件，在where子句、select子句中或其他方式過濾都不行
?SELECT?a.key,?a.value
?FROM?a?
?WHERE?a.key?in?
?(SELECT?b.key?FROM?B);?
?--可以被重寫為：
?SELECT?a.key,?a.val?
?FROM?a?LEFT?SEMI?JOIN?b?on?(a.key?=?b.key)
三、Hive SQL 執(zhí)行原理圖解
我們都知道，一個好的的 Hive SQL 和寫得不好的 Hive SQL ，對底層計算和資源的使用可能相差百倍甚至千倍、萬倍。
除了資源的浪費，不恰當(dāng)?shù)厥褂?Hive SQL 可能會運行幾個小時甚至十幾個小時都得不到運算結(jié)果。因此，我們深入的理解 Hive SQL 的執(zhí)行過程和原理是非常有必要的。
以 group by 語句執(zhí)行圖解為例：
我們假定一個業(yè)務(wù)背景：分析購買iPhone7客戶在各城市中的分布情況，即哪個城市購買得最多、哪個最少。
select?city,count(order_id)?as?iphone7_count?from?orders_table?where?day='201901010'?and?cat_name='iphone7'?group?by?city;
底層MapReduce執(zhí)行過程：
Hive SQL 的 group by 語句涉及數(shù)據(jù)的重新分發(fā)和分布，因此其執(zhí)行過程完整地包含了 MapReduce 任務(wù)的執(zhí)行過程。
( 1 ）輸入分片
group by 語句的輸入文件依然為 day=20170101 的分區(qū)文件，其輸入分片過程和個數(shù)同 select 語句，也是被分為大小分別為：128MB 、128MB、44MB 三個分片文件。
( 2 ) Map 階段
Hadoop 集群同樣啟動三個 Map 任務(wù)，處理對應(yīng)的三個分片文件；每個 map 任務(wù)處理其對應(yīng)分片文件中的每行，檢查其商品類目是否為 iPhone7 ，如果是，則輸出形如
( 3 ) Combiner 階段
Combiner 階段是可選的，如果指定了 Combiner 操作，那么 Hadoop 會在 Map 任務(wù)的地輸出中執(zhí)行 Combiner 操作，其好處是可以去除冗余輸出，避免不必要的后續(xù)處理和網(wǎng)絡(luò)傳輸開銷等 此列中，Map Task1 的輸出中< hz,1＞出現(xiàn)了兩次，那么 Combiner 操作就可以將其合并為 Combiner 操作是有風(fēng)險的，使用它的原則是 Combiner 的輸出不會影響到 Reduce 計算的最終輸入。例如，如果計算只是求總數(shù)、最大值和最小值，可以使用 combiner ，但是如果做平均值計算使用了 Combiner ，最終的 Reduce 計算結(jié)果就會出錯
( 4 ) Shuffle 階段
完整的shuffle包括分區(qū)(partition)，排序(sort)和分隔(spill)、復(fù)制(copy)、合并(merge)等過程。
對于理解group by語句，關(guān)鍵的過程實際就兩個，即分區(qū)和合并；所謂分區(qū)，即 Hadoop 如何決定將每個 Map 任務(wù)的每個輸出鍵值對分配到那個 Reduce Task 所謂合井，即在 一個Reduce Task 中，如何將來自于多個 Map Task 的同樣一個鍵的值進(jìn)行合并 Hadoop 中最為常用的分區(qū)方法是 Hash Partitioner ，即 Hadoop 會對每個鍵取 hash 值，然后再對此 hash 值按照 reduce 任務(wù)數(shù)目取模，從而得到對應(yīng)的 reduce ，這樣保證相同的鍵，肯定被分配到同一個 reduce 上，同時 hash 函數(shù)也能確保 Map 任務(wù)的輸出被均勻地分配到所有的 Reduce務(wù)上
( 5 ）Reduce 階段
調(diào)用reduce函數(shù)，每個reduce任務(wù)的輸出存到本地文件中
( 6 ）輸出文件
hadoop 合并 Reduce Task任務(wù)的輸出文件到輸出目錄
四、小結(jié)
我們介紹了 Hive SQL 的執(zhí)行原理。當(dāng)然了，要知其然，并要知其所以然，理解 Hive 的執(zhí)行原理是寫高效 SQL 的前提和基礎(chǔ)，也是掌握 Hive SQL 優(yōu)化技巧的根本，接下來我們就要進(jìn)入 Hive 優(yōu)化實踐的環(huán)節(jié)啦。