[一起學Hive]之十九-使用Hive API分析HQL的執行計划、Job數量和表的血緣關系

關系

Hive [e-mail] 9個月前 (09-01) 2468℃ 1評論

關鍵字：Hive HQL Job數量、Hive執行計划、Hive LineageInfo

本文介紹使用Hive的API獲取一條HQL的最終執行計划，從而獲取這條HQL的Job數量，另外，介紹使用API分析一條HQL中所包含的輸入表和輸出表。這些信息在做元數據管理和Hive表的血緣分析時候很有用。

Hive在執行一條HQL的時候，會經過以下步驟：

1. 語法解析：Antlr定義SQL的語法規則，完成SQL詞法，語法解析，將SQL轉化為抽象 語法樹AST Tree；
2. 語義解析：遍歷AST Tree，抽象出查詢的基本組成單元QueryBlock；
3. 生成邏輯執行計划：遍歷QueryBlock，翻譯為執行操作樹OperatorTree；
4. 優化邏輯執行計划：邏輯層優化器進行OperatorTree變換，合並不必要的ReduceSinkOperator，減少shuffle數據量；

5. 生成物理執行計划：遍歷OperatorTree，翻譯為MapReduce任務；
6. 優化物理執行計划：物理層優化器進行MapReduce任務的變換，生成最終的執行計划；

關於這幾個步驟，在美團的技術博客上有一篇文章介紹的非常好，可以參考：http://tech.meituan.com/hive-sql-to-mapreduce.html

一般情況下，HQL中的每一個表或者子查詢都會生成一個job，這是邏輯執行計划中生成的，但后面Hive還會優化，比如：使用MapJoin，最終一條HQL語句生成的job數量很難通過HQL觀察出來。

獲取HQL的執行計划和Job數量

直接看代碼吧：

1. package com.lxw1234.test;
2.  
3. import org.apache.hadoop.fs.Path;
4. import org.apache.hadoop.hive.conf.HiveConf;
5. import org.apache.hadoop.hive.ql.Context;
6. import org.apache.hadoop.hive.ql.QueryPlan;
7. import org.apache.hadoop.hive.ql.exec.Utilities;
8. import org.apache.hadoop.hive.ql.parse.ASTNode;
9. import org.apache.hadoop.hive.ql.parse.BaseSemanticAnalyzer;
10. import org.apache.hadoop.hive.ql.parse.ParseDriver;
11. import org.apache.hadoop.hive.ql.parse.ParseUtils;
12. import org.apache.hadoop.hive.ql.parse.SemanticAnalyzerFactory;
13. import org.apache.hadoop.hive.ql.session.SessionState;
14.  
15. /**
16. * lxw的大數據田地 -- lxw1234.com
17. * @author lxw1234
18. *
19. */
20. public class HiveQueryPlan {
21. public static void main(String[] args) throws Exception {
22. HiveConf conf = new HiveConf();
23. conf.addResource(new Path("file:///usr/local/apache-hive-0.13.1-bin/conf/hive-site.xml"));
24. conf.addResource(new Path("file:///usr/local/apache-hive-0.13.1-bin/conf/hive-default.xml.template"));
25. conf.set("javax.jdo.option.ConnectionURL",
26. "jdbc:mysql://127.0.0.1:3306/hive?createDatabaseIfNotExist=true&useUnicode=true&characterEncoding=utf-8");
27. conf.set("hive.metastore.local", "true");
28. conf.set("javax.jdo.option.ConnectionDriverName","com.mysql.jdbc.Driver");
29. conf.set("javax.jdo.option.ConnectionUserName", "hive");
30. conf.set("javax.jdo.option.ConnectionPassword", "hive");
31. conf.set("hive.stats.dbclass", "jdbc:mysql");
32. conf.set("hive.stats.jdbcdriver", "com.mysql.jdbc.Driver");
33. conf.set("hive.exec.dynamic.partition.mode", "nonstrict");
35. String command = args[0];
36. SessionState.start(conf);
37. Context ctx = new Context(conf);
38. ParseDriver pd = new ParseDriver();
39. ASTNode tree = pd.parse(command, ctx);
40. tree = ParseUtils.findRootNonNullToken(tree);
41. BaseSemanticAnalyzer sem = SemanticAnalyzerFactory.get(conf, tree);
42. sem.analyze(tree, ctx);
43. sem.validate();
44. QueryPlan queryPlan = new QueryPlan(command,sem,0l);
45. int jobs = Utilities.getMRTasks(queryPlan.getRootTasks()).size();
46. System.out.println("Total jobs = " + jobs);
48. }
49. }
50.  

將上面的代碼打包成testhive.jar，運行該類需要引入Hive的依賴包，在包含Hadoop和Hive客戶端的機器上執行下面的命令：

1. for f in /usr/local/apache-hive-0.13.1-bin/lib/*.jar; do
2. HADOOP_CLASSPATH=${HADOOP_CLASSPATH}:$f;
3. done
4. export HADOOP_CLASSPATH

分別解析下面三條HQL語句：

1. HQL1：SELECT COUNT(1) FROM liuxiaowen.lxw1;
2. HQL2：SELECT COUNT(1) FROM (SELECT url FROM liuxiaowen.lxw1 GROUP BY url) x;
3. HQL3：SELECT COUNT(1) FROM liuxiaowen.lxw1 a join liuxiaowen.lxw2 b ON (a.url = b.domain);

解析HQL1：

1. hadoop jar testhive.jar com.lxw1234.test.HiveQueryPlan "SELECT COUNT(1) FROM liuxiaowen.lxw1"

結果如下：

解析HQL2：

1. hadoop jar testhive.jar com.lxw1234.test.HiveQueryPlan "SELECT COUNT(1) FROM (SELECT url FROM liuxiaowen.lxw1 GROUP BY url) x"

結果如下：

解析HQL3：

1. hadoop jar testhive.jar com.lxw1234.test.HiveQueryPlan "SELECT COUNT(1) FROM liuxiaowen.lxw1 a join liuxiaowen.lxw2 b ON (a.url = b.domain)"

結果如下：

在HQL3中，由於Hive自動優化使用了MapJoin，因此，兩個表的join最終只用了一個job，在Hive中執行驗證一下：

解析HQL中表的血緣關系

在元數據管理中，可能需要知道Hive中有哪些表，以及這些表之間的關聯關系，比如：A表是由B表和C表統計匯總而來。

Hive中本身自帶了一個工具，用來分析一條HQL中的源表和目標表，org.apache.hadoop.hive.ql.tools.LineageInfo

但該類中目標表只能是使用INSERT語句插入數據的目標表，對於使用CREATE TABLE AS語句創建的表分析不出來。

下面的代碼只對org.apache.hadoop.hive.ql.tools.LineageInfo做了小小的修改：

1. package com.lxw1234.test;
2.  
3. import java.io.IOException;
4. import java.util.ArrayList;
5. import java.util.LinkedHashMap;
6. import java.util.Map;
7. import java.util.Stack;
8. import java.util.TreeSet;
9.  
10. import org.apache.hadoop.hive.ql.lib.DefaultGraphWalker;
11. import org.apache.hadoop.hive.ql.lib.DefaultRuleDispatcher;
12. import org.apache.hadoop.hive.ql.lib.Dispatcher;
13. import org.apache.hadoop.hive.ql.lib.GraphWalker;
14. import org.apache.hadoop.hive.ql.lib.Node;
15. import org.apache.hadoop.hive.ql.lib.NodeProcessor;
16. import org.apache.hadoop.hive.ql.lib.NodeProcessorCtx;
17. import org.apache.hadoop.hive.ql.lib.Rule;
18. import org.apache.hadoop.hive.ql.parse.ASTNode;
19. import org.apache.hadoop.hive.ql.parse.BaseSemanticAnalyzer;
20. import org.apache.hadoop.hive.ql.parse.HiveParser;
21. import org.apache.hadoop.hive.ql.parse.ParseDriver;
22. import org.apache.hadoop.hive.ql.parse.ParseException;
23. import org.apache.hadoop.hive.ql.parse.SemanticException;
24.  
25. /**
26. * lxw的大數據田地 -- lxw1234.com
27. * @author lxw1234
28. *
29. */
30. public class HiveLineageInfo implements NodeProcessor {
31.  
32. /**
33. * Stores input tables in sql.
34. */
35. TreeSet inputTableList = new TreeSet();
36. /**
37. * Stores output tables in sql.
38. */
39. TreeSet OutputTableList = new TreeSet();
40.  
41. /**
42. *
43. * @return java.util.TreeSet
44. */
45. public TreeSet getInputTableList() {
46. return inputTableList;
47. }
48.  
49. /**
50. * @return java.util.TreeSet
51. */
52. public TreeSet getOutputTableList() {
53. return OutputTableList;
54. }
55.  
56. /**
57. * Implements the process method for the NodeProcessor interface.
58. */
59. public Object process(Node nd, Stack stack, NodeProcessorCtx procCtx,
60. Object... nodeOutputs) throws SemanticException {
61. ASTNode pt = (ASTNode) nd;
62.  
63. switch (pt.getToken().getType()) {
65. case HiveParser.TOK_CREATETABLE:
66. OutputTableList.add(BaseSemanticAnalyzer.getUnescapedName((ASTNode)pt.getChild(0)));
67. break;
68. case HiveParser.TOK_TAB:
69. OutputTableList.add(BaseSemanticAnalyzer.getUnescapedName((ASTNode)pt.getChild(0)));
70. break;
71.  
72. case HiveParser.TOK_TABREF:
73. ASTNode tabTree = (ASTNode) pt.getChild(0);
74. String table_name = (tabTree.getChildCount() == 1) ?
75. BaseSemanticAnalyzer.getUnescapedName((ASTNode)tabTree.getChild(0)) :
76. BaseSemanticAnalyzer.getUnescapedName((ASTNode)tabTree.getChild(0)) + "." + tabTree.getChild(1);
77. inputTableList.add(table_name);
78. break;
79. }
80. return null;
81. }
82.  
83. /**
84. * parses given query and gets the lineage info.
85. *
86. * @param query
87. * @throws ParseException
88. */
89. public void getLineageInfo(String query) throws ParseException,
90. SemanticException {
91.  
92. /*
93. * Get the AST tree
94. */
95. ParseDriver pd = new ParseDriver();
96. ASTNode tree = pd.parse(query);
97.  
98. while ((tree.getToken() == null) && (tree.getChildCount() > 0)) {
99. tree = (ASTNode) tree.getChild(0);
100. }
101.  
102. /*
103. * initialize Event Processor and dispatcher.
104. */
105. inputTableList.clear();
106. OutputTableList.clear();
107.  
108. // create a walker which walks the tree in a DFS manner while maintaining
109. // the operator stack. The dispatcher
110. // generates the plan from the operator tree
111. Map<Rule, NodeProcessor> rules = new LinkedHashMap<Rule, NodeProcessor>();
112.  
113. // The dispatcher fires the processor corresponding to the closest matching
114. // rule and passes the context along
115. Dispatcher disp = new DefaultRuleDispatcher(this, rules, null);
116. GraphWalker ogw = new DefaultGraphWalker(disp);
117.  
118. // Create a list of topop nodes
119. ArrayList topNodes = new ArrayList();
120. topNodes.add(tree);
121. ogw.startWalking(topNodes, null);
122. }
123.  
124. public static void main(String[] args) throws IOException, ParseException,
125. SemanticException {
126. String query = args[0];
127. HiveLineageInfo lep = new HiveLineageInfo();
128. lep.getLineageInfo(query);
129. System.out.println("Input tables = " + lep.getInputTableList());
130. System.out.println("Output tables = " + lep.getOutputTableList());
131. }
132. }
133.  

將上面的程序打包成testhive.jar，同上面，執行時候需要引入Hive的依賴包：

分析下面兩條HQL語句：

1. HQL1:CREATE TABLE liuxiaowen.lxw1234 AS SELECT * FROM liuxiaowen.lxw1;
2. HQL2:INSERT OVERWRITE TABLE liuxiaowen.lxw3 SELECT a.url FROM liuxiaowen.lxw1 a join liuxiaowen.lxw2 b ON (a.url = b.domain);

執行命令：

1. hadoop jar testhive.jar com.lxw1234.test.HiveLineageInfo "CREATE TABLE liuxiaowen.lxw1234 AS SELECT * FROM liuxiaowen.lxw1"
2. hadoop jar testhive.jar com.lxw1234.test.HiveLineageInfo "INSERT OVERWRITE TABLE liuxiaowen.lxw3 SELECT a.url FROM liuxiaowen.lxw1 a
3. join liuxiaowen.lxw2 b ON (a.url = b.domain)"

分析結果：

HQL中的Input table和Output table已經正確解析出來。