Hive SQL的编译过程
Hive是基于Hadoop的一個數據倉庫系統，在各大公司都有廣泛的應用。美團數據倉庫也是基于Hive搭建，每天執行近萬次的Hive ETL計算流程，負責每天數百GB的數據存儲和分析。Hive的穩定性和性能對我們的數據分析非常關鍵。
在幾次升級Hive的過程中，我們遇到了一些大大小小的問題。通過向社區的咨詢和自己的努力，在解決這些問題的同時我們對Hive將SQL編譯為MapReduce的過程有了比較深入的理解。對這一過程的理解不僅幫助我們解決了一些Hive的bug，也有利于我們優化Hive SQL，提升我們對Hive的掌控力，同時有能力去定制一些需要的功能。
MapReduce實現基本SQL操作的原理
詳細講解SQL編譯為MapReduce之前，我們先來看看MapReduce框架實現SQL基本操作的原理
Join的實現原理
select u.name, o.orderid from order o join user u on o.uid = u.uid;在map的輸出value中為不同表的數據打上tag標記，在reduce階段根據tag判斷數據來源。MapReduce的過程如下（這里只是說明最基本的Join的實現，還有其他的實現方式）
Group By的實現原理
select rank, isonline, count(*) from city group by rank, isonline;將GroupBy的字段組合為map的輸出key值，利用MapReduce的排序，在reduce階段保存LastKey區分不同的key。MapReduce的過程如下（當然這里只是說明Reduce端的非Hash聚合過程）
Distinct的實現原理
select dealid, count(distinct uid) num from order group by dealid;當只有一個distinct字段時，如果不考慮Map階段的Hash GroupBy，只需要將GroupBy字段和Distinct字段組合為map輸出key，利用mapreduce的排序，同時將GroupBy字段作為reduce的key，在reduce階段保存LastKey即可完成去重
如果有多個distinct字段呢，如下面的SQL
select dealid, count(distinct uid), count(distinct date) from order group by dealid;實現方式有兩種：
（1）如果仍然按照上面一個distinct字段的方法，即下圖這種實現方式，無法跟據uid和date分別排序，也就無法通過LastKey去重，仍然需要在reduce階段在內存中通過Hash去重
（2）第二種實現方式，可以對所有的distinct字段編號，每行數據生成n行數據，那么相同字段就會分別排序，這時只需要在reduce階段記錄LastKey即可去重。
這種實現方式很好的利用了MapReduce的排序，節省了reduce階段去重的內存消耗，但是缺點是增加了shuffle的數據量。
需要注意的是，在生成reduce value時，除第一個distinct字段所在行需要保留value值，其余distinct數據行value字段均可為空。
SQL轉化為MapReduce的過程
了解了MapReduce實現SQL基本操作之后，我們來看看Hive是如何將SQL轉化為MapReduce任務的，整個編譯過程分為六個階段：
下面分別對這六個階段進行介紹
Phase1 SQL詞法，語法解析
Antlr
Hive使用Antlr實現SQL的詞法和語法解析。Antlr是一種語言識別的工具，可以用來構造領域語言。 這里不詳細介紹Antlr，只需要了解使用Antlr構造特定的語言只需要編寫一個語法文件，定義詞法和語法替換規則即可，Antlr完成了詞法分析、語法分析、語義分析、中間代碼生成的過程。
Hive中語法規則的定義文件在0.10版本以前是Hive.g一個文件，隨著語法規則越來越復雜，由語法規則生成的Java解析類可能超過Java類文件的最大上限，0.11版本將Hive.g拆成了5個文件，詞法規則HiveLexer.g和語法規則的4個文件SelectClauseParser.g，FromClauseParser.g，IdentifiersParser.g，HiveParser.g。
抽象語法樹AST Tree
經過詞法和語法解析后，如果需要對表達式做進一步的處理，使用 Antlr 的抽象語法樹語法Abstract Syntax Tree，在語法分析的同時將輸入語句轉換成抽象語法樹，后續在遍歷語法樹時完成進一步的處理。
下面的一段語法是Hive SQL中SelectStatement的語法規則，從中可以看出，SelectStatement包含select, from, where, groupby, having, orderby等子句。 （在下面的語法規則中，箭頭表示對于原語句的改寫，改寫后會加入一些特殊詞標示特定語法，比如TOK_QUERY標示一個查詢塊）
selectStatement:selectClausefromClausewhereClause?groupByClause?havingClause?orderByClause?clusterByClause?distributeByClause?sortByClause?limitClause? -> ^(TOK_QUERY fromClause ^(TOK_INSERT ^(TOK_DESTINATION ^(TOK_DIR TOK_TMP_FILE))selectClause whereClause? groupByClause? havingClause? orderByClause? clusterByClause?distributeByClause? sortByClause? limitClause?));樣例SQL
為了詳細說明SQL翻譯為MapReduce的過程，這里以一條簡單的SQL為例，SQL中包含一個子查詢，最終將數據寫入到一張表中
FROM ( SELECTp.datekey datekey,p.userid userid,c.clienttypeFROMdetail.usersequence_client cJOIN fact.orderpayment p ON p.orderid = c.orderidJOIN default.user du ON du.userid = p.useridWHERE p.datekey = 20131118 ) base INSERT OVERWRITE TABLE `test`.`customer_kpi` SELECTbase.datekey,base.clienttype,count(distinct base.userid) buyer_count GROUP BY base.datekey, base.clienttypeSQL生成AST Tree
Antlr對Hive SQL解析的代碼如下，HiveLexerX，HiveParser分別是Antlr對語法文件Hive.g編譯后自動生成的詞法解析和語法解析類，在這兩個類中進行復雜的解析。
HiveLexerX lexer = new HiveLexerX(new ANTLRNoCaseStringStream(command)); //詞法解析，忽略關鍵詞的大小寫 TokenRewriteStream tokens = new TokenRewriteStream(lexer); if (ctx != null) {ctx.setTokenRewriteStream(tokens); } HiveParser parser = new HiveParser(tokens); //語法解析 parser.setTreeAdaptor(adaptor); HiveParser.statement_return r = null; try {r = parser.statement(); //轉化為AST Tree } catch (RecognitionException e) {e.printStackTrace();throw new ParseException(parser.errors); }最終生成的AST Tree如下圖右側（使用Antlr Works生成，Antlr Works是Antlr提供的編寫語法文件的編輯器），圖中只是展開了骨架的幾個節點，沒有完全展開。 子查詢1/2，分別對應右側第1/2兩個部分。
這里注意一下內層子查詢也會生成一個TOK_DESTINATION節點。請看上面SelectStatement的語法規則，這個節點是在語法改寫中特意增加了的一個節點。原因是Hive中所有查詢的數據均會保存在HDFS臨時的文件中，無論是中間的子查詢還是查詢最終的結果，Insert語句最終會將數據寫入表所在的HDFS目錄下。
詳細來看，將內存子查詢的from子句展開后，得到如下AST Tree，每個表生成一個TOK_TABREF節點，Join條件生成一個“=”節點。其他SQL部分類似，不一一詳述。
Phase2 SQL基本組成單元QueryBlock
AST Tree仍然非常復雜，不夠結構化，不方便直接翻譯為MapReduce程序，AST Tree轉化為QueryBlock就是將SQL進一部抽象和結構化。
QueryBlock
QueryBlock是一條SQL最基本的組成單元，包括三個部分：輸入源，計算過程，輸出。簡單來講一個QueryBlock就是一個子查詢。
下圖為Hive中QueryBlock相關對象的類圖，解釋圖中幾個重要的屬性
- QB#aliasToSubq（表示QB類的aliasToSubq屬性）保存子查詢的QB對象，aliasToSubq key值是子查詢的別名
- QB#qbp即QBParseInfo保存一個基本SQL單元中的給個操作部分的AST Tree結構，QBParseInfo#nameToDest這個HashMap保存查詢單元的輸出，key的形式是inclause-i（由于Hive支持Multi Insert語句，所以可能有多個輸出），value是對應的ASTNode節點，即TOK_DESTINATION節點。類QBParseInfo其余HashMap屬性分別保存輸出和各個操作的ASTNode節點的對應關系。
- QBParseInfo#JoinExpr保存TOK_JOIN節點。QB#QBJoinTree是對Join語法樹的結構化。
- QB#qbm保存每個輸入表的元信息，比如表在HDFS上的路徑，保存表數據的文件格式等。
- QBExpr這個對象是為了表示Union操作。
AST Tree生成QueryBlock
AST Tree生成QueryBlock的過程是一個遞歸的過程，先序遍歷AST Tree，遇到不同的Token節點，保存到相應的屬性中，主要包含以下幾個過程
- TOK_QUERY => 創建QB對象，循環遞歸子節點
- TOK_FROM => 將表名語法部分保存到QB對象的aliasToTabs等屬性中
- TOK_INSERT => 循環遞歸子節點
- TOK_DESTINATION => 將輸出目標的語法部分保存在QBParseInfo對象的nameToDest屬性中
- TOK_SELECT => 分別將查詢表達式的語法部分保存在destToSelExpr、destToAggregationExprs、destToDistinctFuncExprs三個屬性中
- TOK_WHERE => 將Where部分的語法保存在QBParseInfo對象的destToWhereExpr屬性中
最終樣例SQL生成兩個QB對象，QB對象的關系如下，QB1是外層查詢，QB2是子查詢
QB1\QB2Phase3 邏輯操作符Operator
Operator
Hive最終生成的MapReduce任務，Map階段和Reduce階段均由OperatorTree組成。邏輯操作符，就是在Map階段或者Reduce階段完成單一特定的操作。
基本的操作符包括TableScanOperator，SelectOperator，FilterOperator，JoinOperator，GroupByOperator，ReduceSinkOperator
從名字就能猜出各個操作符完成的功能，TableScanOperator從MapReduce框架的Map接口原始輸入表的數據，控制掃描表的數據行數，標記是從原表中取數據。JoinOperator完成Join操作。FilterOperator完成過濾操作
ReduceSinkOperator將Map端的字段組合序列化為Reduce Key/value, Partition Key，只可能出現在Map階段，同時也標志著Hive生成的MapReduce程序中Map階段的結束。
Operator在Map Reduce階段之間的數據傳遞都是一個流式的過程。每一個Operator對一行數據完成操作后之后將數據傳遞給childOperator計算。
Operator類的主要屬性和方法如下
- RowSchema表示Operator的輸出字段
- InputObjInspector outputObjInspector解析輸入和輸出字段
- processOp接收父Operator傳遞的數據，forward將處理好的數據傳遞給子Operator處理
- Hive每一行數據經過一個Operator處理之后，會對字段重新編號，colExprMap記錄每個表達式經過當前Operator處理前后的名稱對應關系，在下一個階段邏輯優化階段用來回溯字段名
- 由于Hive的MapReduce程序是一個動態的程序，即不確定一個MapReduce Job會進行什么運算，可能是Join，也可能是GroupBy，所以Operator將所有運行時需要的參數保存在OperatorDesc中，OperatorDesc在提交任務前序列化到HDFS上，在MapReduce任務執行前從HDFS讀取并反序列化。Map階段OperatorTree在HDFS上的位置在Job.getConf(“hive.exec.plan”) + “/map.xml”
QueryBlock生成Operator Tree
QueryBlock生成Operator Tree就是遍歷上一個過程中生成的QB和QBParseInfo對象的保存語法的屬性，包含如下幾個步驟：
- QB#aliasToSubq => 有子查詢，遞歸調用
- QB#aliasToTabs => TableScanOperator
- QBParseInfo#joinExpr => QBJoinTree => ReduceSinkOperator + JoinOperator
- QBParseInfo#destToWhereExpr => FilterOperator
- QBParseInfo#destToGroupby => ReduceSinkOperator + GroupByOperator
- QBParseInfo#destToOrderby => ReduceSinkOperator + ExtractOperator
由于Join/GroupBy/OrderBy均需要在Reduce階段完成，所以在生成相應操作的Operator之前都會先生成一個ReduceSinkOperator，將字段組合并序列化為Reduce Key/value, Partition Key
接下來詳細分析樣例SQL生成OperatorTree的過程
先序遍歷上一個階段生成的QB對象
首先根據子QueryBlock QB2#aliasToTabs {du=dim.user, c=detail.usersequence_client, p=fact.orderpayment}生成TableScanOperator
TableScanOperator(“dim.user”) TS[0] TableScanOperator(“detail.usersequence_client”) TS[1] TableScanOperator(“fact.orderpayment”) TS[2]先序遍歷QBParseInfo#joinExpr生成QBJoinTree，類QBJoinTree也是一個樹狀結構，QBJoinTree保存左右表的ASTNode和這個查詢的別名，最終生成的查詢樹如下
base/ \p du/ \ c p前序遍歷QBJoinTree，先生成detail.usersequence_client和fact.orderpayment的Join操作樹
圖中 TS=TableScanOperator RS=ReduceSinkOperator JOIN=JoinOperator
下圖中SelectOperator在某些場景下會根據一些條件判斷是否需要解析字段。
圖中 FIL= FilterOperator SEL= SelectOperator
圖中 GBY= GroupByOperator GBY[12]是HASH聚合，即在內存中通過Hash進行聚合運算
圖中FS=FileSinkOperator
Phase4 邏輯層優化器
大部分邏輯層優化器通過變換OperatorTree，合并操作符，達到減少MapReduce Job，減少shuffle數據量的目的。
②?SimpleFetchOptimizer | 優化沒有GroupBy表達式的聚合查詢 |
②?MapJoinProcessor | MapJoin，需要SQL中提供hint，0.11版本已不用 |
②?BucketMapJoinOptimizer | BucketMapJoin |
② GroupByOptimizer | Map端聚合 |
① ReduceSinkDeDuplication | 合并線性的OperatorTree中partition/sort key相同的reduce |
① PredicatePushDown | 謂詞前置 |
① CorrelationOptimizer | 利用查詢中的相關性，合并有相關性的Job，HIVE-2206 |
ColumnPruner | 字段剪枝 |
表格中①的優化器均是一個Job干盡可能多的事情/合并。②的都是減少shuffle數據量，甚至不做Reduce。
CorrelationOptimizer優化器非常復雜，都能利用查詢中的相關性，合并有相關性的Job，參考 Hive Correlation Optimizer
對于樣例SQL，有兩個優化器對其進行優化。下面分別介紹這兩個優化器的作用，并補充一個優化器ReduceSinkDeDuplication的作用
PredicatePushDown優化器
斷言判斷提前優化器將OperatorTree中的FilterOperator提前到TableScanOperator之后
NonBlockingOpDeDupProc優化器
NonBlockingOpDeDupProc優化器合并SEL-SEL 或者 FIL-FIL 為一個Operator
ReduceSinkDeDuplication優化器
ReduceSinkDeDuplication可以合并線性相連的兩個RS。實際上CorrelationOptimizer是ReduceSinkDeDuplication的超集，能合并線性和非線性的操作RS，但是Hive先實現的ReduceSinkDeDuplication
譬如下面這條SQL語句
from (select key, value from src group by key, value) s select s.key group by s.key;經過前面幾個階段之后，會生成如下的OperatorTree，兩個Tree是相連的，這里沒有畫到一起
這時候遍歷OperatorTree后能發現前前后兩個RS輸出的Key值和PartitionKey如下
key | key |
key,value | key,value |
ReduceSinkDeDuplication優化器檢測到：1. pRS Key完全包含cRS Key，且排序順序一致；2. pRS PartitionKey完全包含cRS PartitionKey。符合優化條件，會對執行計劃進行優化。
ReduceSinkDeDuplication將childRS和parentheRS與childRS之間的Operator刪掉，保留的RS的Key為key,value字段，PartitionKey為key字段。合并后的OperatorTree如下：
Phase5 OperatorTree生成MapReduce Job的過程
OperatorTree轉化為MapReduce Job的過程分為下面幾個階段
對輸出表生成MoveTask
由上一步OperatorTree只生成了一個FileSinkOperator，直接生成一個MoveTask，完成將最終生成的HDFS臨時文件移動到目標表目錄下
MoveTask[Stage-0] Move Operator開始遍歷
將OperatorTree中的所有根節點保存在一個toWalk的數組中，循環取出數組中的元素（省略QB1，未畫出）
取出最后一個元素TS[p]放入棧 opStack{TS[p]}中
Rule #1 TS% 生成MapReduceTask對象，確定MapWork
發現棧中的元素符合下面規則R1（這里用python代碼簡單表示）
"".join([t + "%" for t in opStack]) == "TS%"生成一個MapReduceTask[Stage-1]對象，MapReduceTask[Stage-1]對象的MapWork屬性保存Operator根節點的引用。由于OperatorTree之間之間的Parent Child關系，這個時候MapReduceTask[Stage-1]包含了以TS[p]為根的所有Operator
Rule #2 TS%.*RS% 確定ReduceWork
繼續遍歷TS[p]的子Operator，將子Operator存入棧opStack中 當第一個RS進棧后，即棧opStack = {TS[p], FIL[18], RS[4]}時，就會滿足下面的規則R2
"".join([t + "%" for t in opStack]) == "TS%.*RS%"這時候在MapReduceTask[Stage-1]對象的ReduceWork屬性保存JOIN[5]的引用
Rule #3 RS%.*RS% 生成新MapReduceTask對象，切分MapReduceTask
繼續遍歷JOIN[5]的子Operator，將子Operator存入棧opStack中
當第二個RS放入棧時，即當棧opStack = {TS[p], FIL[18], RS[4], JOIN[5], RS[6]}時，就會滿足下面的規則R3
"".join([t + "%" for t in opStack]) == “RS%.*RS%” //循環遍歷opStack的每一個后綴數組這時候創建一個新的MapReduceTask[Stage-2]對象，將OperatorTree從JOIN[5]和RS[6]之間剪開，并為JOIN[5]生成一個子Operator FS[19]，RS[6]生成一個TS[20]，MapReduceTask[Stage-2]對象的MapWork屬性保存TS[20]的引用。
新生成的FS[19]將中間數據落地，存儲在HDFS臨時文件中。
繼續遍歷RS[6]的子Operator，將子Operator存入棧opStack中
當opStack = {TS[p], FIL[18], RS[4], JOIN[5], RS[6], JOIN[8], SEL[10], GBY[12], RS[13]}時，又會滿足R3規則
同理生成MapReduceTask[Stage-3]對象，并切開 Stage-2 和 Stage-3 的OperatorTree
R4 FS% 連接MapReduceTask與MoveTask
最終將所有子Operator存入棧中之后，opStack = {TS[p], FIL[18], RS[4], JOIN[5], RS[6], JOIN[8], SEL[10], GBY[12], RS[13], GBY[14], SEL[15], FS[17]} 滿足規則R4
"".join([t + "%" for t in opStack]) == “FS%”這時候將MoveTask與MapReduceTask[Stage-3]連接起來，并生成一個StatsTask，修改表的元信息
合并Stage
此時并沒有結束，還有兩個根節點沒有遍歷。
將opStack棧清空，將toWalk的第二個元素加入棧。會發現opStack = {TS[du]}繼續滿足R1 TS%，生成MapReduceTask[Stage-5]
繼續從TS[du]向下遍歷，當opStack={TS[du], RS[7]}時，滿足規則R2 TS%.*RS%
此時將JOIN[8]保存為MapReduceTask[Stage-5]的ReduceWork時，發現在一個Map對象保存的Operator與MapReduceWork對象關系的Map<Operator, MapReduceWork>對象中發現，JOIN[8]已經存在。此時將MapReduceTask[Stage-2]和MapReduceTask[Stage-5]合并為一個MapReduceTask
同理從最后一個根節點TS[c]開始遍歷，也會對MapReduceTask進行合并
切分Map Reduce階段
最后一個階段，將MapWork和ReduceWork中的OperatorTree以RS為界限剪開
OperatorTree生成MapReduceTask全貌
最終共生成3個MapReduceTask，如下圖
Phase6 物理層優化器
這里不詳細介紹每個優化器的原理，單獨介紹一下MapJoin的優化器
Vectorizer | HIVE-4160，將在0.13中發布 |
SortMergeJoinResolver | 與bucket配合，類似于歸并排序 |
SamplingOptimizer | 并行order by優化器，在0.12中發布 |
CommonJoinResolver + MapJoinResolver | MapJoin優化器 |
MapJoin原理
MapJoin簡單說就是在Map階段將小表讀入內存，順序掃描大表完成Join。
上圖是Hive MapJoin的原理圖，出自Facebook工程師Liyin Tang的一篇介紹Join優化的slice，從圖中可以看出MapJoin分為兩個階段：
通過MapReduce Local Task，將小表讀入內存，生成HashTableFiles上傳至Distributed Cache中，這里會對HashTableFiles進行壓縮。
MapReduce Job在Map階段，每個Mapper從Distributed Cache讀取HashTableFiles到內存中，順序掃描大表，在Map階段直接進行Join，將數據傳遞給下一個MapReduce任務。
如果Join的兩張表一張表是臨時表，就會生成一個ConditionalTask，在運行期間判斷是否使用MapJoin
CommonJoinResolver優化器
CommonJoinResolver優化器就是將CommonJoin轉化為MapJoin，轉化過程如下
遍歷上一個階段生成的MapReduce任務，發現MapReduceTask[Stage-2] JOIN[8]中有一張表為臨時表，先對Stage-2進行深度拷貝（由于需要保留原始執行計劃為Backup Plan，所以這里將執行計劃拷貝了一份），生成一個MapJoinOperator替代JoinOperator，然后生成一個MapReduceLocalWork讀取小表生成HashTableFiles上傳至DistributedCache中。
MapReduceTask經過變換后的執行計劃如下圖所示
MapJoinResolver優化器
MapJoinResolver優化器遍歷Task Tree，將所有有local work的MapReduceTask拆成兩個Task
最終MapJoinResolver處理完之后，執行計劃如下圖所示
Hive SQL編譯過程的設計
從上述整個SQL編譯的過程，可以看出編譯過程的設計有幾個優點值得學習和借鑒
- 使用Antlr開源軟件定義語法規則，大大簡化了詞法和語法的編譯解析過程，僅僅需要維護一份語法文件即可。
- 整體思路很清晰，分階段的設計使整個編譯過程代碼容易維護，使得后續各種優化器方便的以可插拔的方式開關，譬如Hive 0.13最新的特性Vectorization和對Tez引擎的支持都是可插拔的。
- 每個Operator只完成單一的功能，簡化了整個MapReduce程序。
社區發展方向
Hive依然在迅速的發展中，為了提升Hive的性能，hortonworks公司主導的Stinger計劃提出了一系列對Hive的改進，比較重要的改進有：
- Vectorization - 使Hive從單行單行處理數據改為批量處理方式，大大提升了指令流水線和緩存的利用率
- Hive on Tez - 將Hive底層的MapReduce計算框架替換為Tez計算框架。Tez不僅可以支持多Reduce階段的任務MRR，還可以一次性提交執行計劃，因而能更好的分配資源。
- Cost Based Optimizer - 使Hive能夠自動選擇最優的Join順序，提高查詢速度
- Implement insert, update, and delete in Hive with full ACID support - 支持表按主鍵的增量更新
我們也將跟進社區的發展，結合自身的業務需要，提升Hive型ETL流程的性能
參考
Antlr: http://www.antlr.org/ Wiki Antlr介紹: http://en.wikipedia.org/wiki/ANTLR Hive Wiki: https://cwiki.apache.org/confluence/display/Hive/Home HiveSQL編譯過程: http://www.slideshare.net/recruitcojp/internal-hive Join Optimization in Hive: Join Strategies in Hive from the 2011 Hadoop Summit (Liyin Tang, Namit Jain) Hive Design Docs: https://cwiki.apache.org/confluence/display/Hive/DesignDocs
總結
以上是生活随笔為你收集整理的Hive SQL的编译过程的全部內容，希望文章能夠幫你解決所遇到的問題。
- 上一篇： 史上最强大型分布式架构详解：高并发+数据
- 下一篇： Intel PAUSE指令变化影响到My