數據倉庫也是基于Hive搭建，每天執行近萬次的HiveETL計算流程，負責每天數百GB的數據和分析。Hive的穩定性和性能對我們的數據分析非常關1、MapReduceSQL在幾次升級Hive的過程中，我們遇到了一些大大小小的問題。通過向社區的咨詢和自己的努力，在解決這些問題的同時我們對Hive將SQL編譯為MapReduce1、MapReduceSQLSQL1.11.1J,o.orderidfromorderojoinuseruono.uid= mapvalue中為不同表的數據打上tagreduce階段根據tag1.21.2GroupByselectrank,isonline,count(*)fromcitygroupbyrank, GroupBymapkeyMapReducereduceLastKeykey。MapReduce（當然這里只是說明Reduce端的非Hash聚合過程）1.31.3Distinctselectdealid,count(distinctuid)numfromordergroupby 當只有一個distinctMapHashGroupBy，只需GroupByDistinctmapkey，mapreduceGroupByreducekeyreducedistinctselectselectdealid,count(distinctuid),count(distinctdate)fromordergroupdistinctuiddateLastKeyreduce階段在內存中通過Hash去重distinctnMapReducereduce存消耗，但是缺點是增加了shuffle的數據量。reducevaluedistinct留value值，其余distinct數據行value字段均可為空。2、SQL2、SQLMapReduceMapReduceSQLHiveMapReduce AntlrSQLAST ASTTree， QueryBlock ，減少 Phase1Phase1SQLHive使用AntlrSQLAntlrAntlrAntlr語法文件，定義詞法和語法替換規則即可，Antlr2.1.2抽象語法樹ASTHive中語則的定義文件在0.10版本以前是Hive.g一個文件，隨著語則越來越復雜，由語則生成的Java解析類可能超過Java類文件的最大上限，0.11版本將Hive.g拆成了5個文件，詞則HiveLexer.g和語則的4個文件SelectClauseParser.g，FromClauseParser.g，IdentifiersParser.g，2.1.2抽象語法樹AST經過詞法和語法解析后，如果需要對表達式做進一步的處理，使用Antlr的抽 下面的一段語法是HiveSQL中SelectStatement的語則，從中可以看出SelectStatementselect,from,where,groupbyhaving,orderby標示特定語法，比如TOK_QUERY標示一個查詢塊）::limitClause?->^(TOK_QUERYfromClause^(TOK_INSERT^(TOK_DIRselectClausewhereClause?groupByClause?havingClause?orderByClause?clusterByClause?distributeByClause?sortByClause?;2.1.3樣例SQLMapReduceSQLSQLp.datekeydatekey,p.useriduserid,cJOINfact.orderpaymentpONp.orderid=c.orderidJOINdefault.userduONdu.userid=p.useridWHEREp.datekey)INSERTOVERWRITETABLE`test`.`customer_kpi`count(distinctbase.userid)GROUPBYbase.datekey,.3SQLASTHive.gHiveLexerXHiveLexerXlexer=newHiveLexerX(new TokenRewriteStreamtokens=newTokenRewriteStream(lexer);if(ctx!=null){}HiveParserparser=new析//HiveParser.statement_returnr=null;try{}catch(RecognitionExceptione){thrownew}//最終生成的ASTTree如下圖右側（使用AntlrWorks生成，AntlrWorks是Antlr這里注意一下內層子查詢也會生成一個TOK_DESTINATION節點。請看上面SelectStatement的語則，這個節點是在語法改寫中特意增加了的一個節Insert詳細來看，將內存子查詢的from子句展開后，得到如下ASTTree，每個表生2.22.2Phase2SQLASTTreeQueryBlockSQL.1QueryBlockSQL程，輸出。簡單來講一個QueryBlock就是一個子查詢。HiveQueryBlockQB#qbp即QBParseInfoSQL支持value節點。類QBParseInfo其余HashMap屬性分別保存輸出和各個操作的ASTNode節點的對應關系。JoinQBExprUnion.2ASTTreeTOK_QUERYTOK_QUERY=>創建QBTOK_FROMQBaliasToTabsTOK_INSERTTOK_DESTINATIONQBParseInfo象的nameToDest屬性中TOK_SELECTdestToSelExpr、destToAggregationExprs、destToDistinctFuncExprs三個屬性中TOK_WHERE=>將WhereQBParseInfo\\Phase3邏輯操作符Hive最終生成的MapReduce任務，Map階段和Reduce階段均由從名字就能猜出各個操作符完成的功能，TableScanOperatorMapReduce取數據。JoinOperator完成Join操作。FilterOperator完成過濾操作ReduceSinkOperatorMapReduceKey/value,PartitionKey，只可能出現在Map階段，同時也標志著Hive生成的MapReduce程序中Map階段的結束。OperatorMapReduceOperatorchildOperatorOperatorRowSchemaRowSchemaOperatorInputObjInspectoroutputObjInspectorprocessOpOperator，forward遞給子Operator處理Hive每一行數據經過一個Operator處理之后，會對字段重新編號，colExprMapOperatorHive的MapReduceOperator將所有運行時需要的參數保存在OperatorDesc中，并反序列化。MapOperatorTreeHDFS在 n”)+.2QueryBlockOperatorQueryBlockOperatorTreeQBQB#aliasToSubqQB#aliasToSubqQBParseInfo#joinExpr=>QBJoinTree=>ReduceSinkOperator+TableScanOperator(“dim.userTableScanOperator(“dim.user”)TS[0] ”)TS[1]Join/GroupBy/OrderByReduce化為ReduceKey/value,PartitionKeyp=fact.orderpaymentQBParseInfo#joinExprQBParseInfo#joinExprQBJoinTreeQBJoinTree/ /\cp 和fact.orderpaymentJoin圖中TS=TableScanOperatorRS=ReduceSinkOperator dim.userJoin QB2QBParseInfo#destToWhereExprFilterOperator下圖中圖中FILFilterOperatorSEL根據QB1的QBParseInfo#destToGroupby生成+圖中GBYGBY[12HASHHashFileSinkOperator，2.42.4Phase4邏輯層優化器MapReduceJob，減少shuffle數據量的目的。②GroupBy②MapJoin，需要SQL中提供hint，0.11版本已②②Map①合并線性的OperatorTree中partition/sortkey相同的reduce①①利用查詢中的相關性，合并有相關性的Job，Job關性的Job，參考HiveCorrelationOptimizer2.4.1PredicatePushDown2.4.1PredicatePushDown.2NonBlockingOpDeDupProcNonBlockingOpDeDupProcSEL-SEL或者FIL-FIL為一個.3ReduceSinkDeDuplicationReduceSinkDeDuplication可以合并線性相連的兩個RS。實際上CorrelationOptimizerReduceSinkDeDuplication非線性的操作RS，但是Hive先實現的ReduceSinkDeDuplicationSQLfromfrom(selectkey,valuefromsrcgroupbykey,value)sselects.keygroupOperatorTreeRSKey排序順序一致；2pRSPartitionKeycRSPartitionKey。符合優化條ReduceSinkDeDuplication將childRS和parentheRS與childRS之間的OperatorRSKeykey,valuePartitionKeykey段。合并后的OperatorTree如下：2.52.5Phase5OperatorTreeMapReduceJobReduceSinkOperatorMap/ReduceJobJoinOperatorStatTaskMapReduceOperator.1對輸出表生成MoveTask，完成將最終生成的HDFS臨時文件移動到目標表 Move2.5.2開始遍歷中的元素（省略QB1，未畫出）TS[p]放入棧opStack{TS[p].3Rule1TS%生成MapReduceTaskR1（python"".join([t+"%"fortinopStack])== MapWork屬性保存Operator根節點的由于OperatorTree之間之間的所有Operator.4Rule2TS%.*RS%確定TS[pOperatorOperatoropStack的規則R2"".join([t+"%"fortinopStack])== 這時候在]對象的屬性保存2.5.5Rule2.5.5Rule3RS%.*RS%生成新MapReduceTaskJOIN[5Operator，OperatoropStackRSopStack{TS[p],FIL[18RS[4],JOIN[5],RS[6]}時，就會滿足下面的規則R3"".join([t"".join([tfortinopStack]“RS%.*RS%”//opStack這時候創建一個新的MapReduceTask[Stage-2]對象，將OperatorTree從JOIN[5RS[6]之間剪開，并為JOIN[5OperatorFS[19]，RS[6]生成一個TS[20]，MapReduceTask[Stage-2]對象的MapWork屬性保存TS[20]的新生成的FS[19]將中間數據落地，在HDFS臨時文件中RS[6OperatorOperatoropStackopStack{TS[p],FIL[18]RS[4],JOIN[5],RS[6],JOIN[8],GBY[12],RS[13R3MapReduceTask[Stage-3]對象，并切開Stage-2和Stage-3的.6R4FS%連接MapReduceTask最終將所有子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]}"".join([t+"%"fortinopStack])== .7合并將opStacktoWalkopStackTS[duopStack={TS[du],RS[7]R2JOIN[8MapReduceTask[Stage-5ReduceWorkMapOperator與MapReduceWorkMap象中發現，JOIN[8]已經存在。此時將MapReduceTask[Stage-2]和MapReduceTask[Stage-5]合并為一個MapReduceTask2.5.8切分2.5.8切分MapReduceMapWork和ReduceWorkOperatorTreeRS2.62.6Phase6物理層優化器MapJoinbucketorderby0.12中發 .1MapJoinMapJoinMapJoinDistributedDistributedCacheHashTableFilesMapReduceJob在Map階段，每個Mapper從DistributedCache傳遞給下一個MapReduce任務。2.6.2Com