hive执行流程分析

转自：http://blog.csdn.net/gexiaobaohelloworld/article/details/7719163

入口：
bin/hive脚本中，环境检查后执行ext中的cli.sh，进入主类：CliDriver.main。

CliDriver.main:
进入cli.processLine，处理分号";"分割为一条一条语句，再进processCmd。

processCmd:
处理quit/exit，再处理source，处理！，处理list；else建立CommandProcessor(实现有Driver和各种Processor)，set/dfs/add/delete命令有单独的Processor，剩下的走Driver。

如果是Driver类型的Processor：
把cmd发到这个driver的run，再进到compile，在compile中，用一个parseDriver去生成ASTNode(生成使用了antlr，主要过程：经过文法分析器切割，进解析器，出来一个TREE)，这里有细节的compile的过程说明 http://fromheartgo.wordpress.com/2010/04/02/hive%E7%9A%84compile%E8%BF%87%E7%A8%8B%EF%BC%881%EF%BC%89/ ；
根据得到的ASTNode，开始语义分析，把结果设置到一个QueryPlan对象中，初始化一些task放在QueryPlan中；

run里的test only代码读了test.serialize.qplan的设置，test状态会把这些查询记录写到文件里；权限检查。

退出complie，在Driver的run中分解执行MR后，退出来到了processCmd:
如果装填一切正常，通过getResults取到MR运行结果。

全过程如下：

CliDriver.main > processLine > processCmd >> Driver.run(cmd) > compile >> BaseSemanticAnalyzer >> xxxSemanticAnalyzer（常规select走SemanticAnalyzer） > analyze(sem.analyze) >> SemanticAnalyzer的analyzeInternal方法 >> new Optimizer.optimize（进行列剪裁等优化后生成Task） > genMapRedTasks >> 返回到Driver.run(cmd) >>ret = execute() >> launchTask >> TaskRunner.run > Task.executeTask > ExecDriver.execute > 执行MR（submitJob） >> getResults.

http://blog.csdn.net/bupt041137/article/details/6553762

即：

HiveCLI [Java Application]

org.apache.hadoop.hive.cli.CliDriver at localhost:38723

Thread [main] (Suspended)

Driver.execute() line: 1344

Driver.runExecute() line: 1219

Driver.run(String, Map<String,Object[]>) line: 1177

Driver.run(String) line: 1159

CliDriver.processLocalCmd(String, CommandProcessor, CliSessionState) line: 258

CliDriver.processCmd(String) line: 215

CliDriver.processLine(String, boolean) line: 411

CliDriver.run(String[]) line: 679

CliDriver.main(String[]) line: 562

/usr/lib/jvm/java-1.6.0-openjdk-1.6.0.0.x86_64/bin/java (2012-7-6 上午11:36:07)

主函数是CliDriver类的main函数，然后走run函数，再做了一些初始化和检测后，再调用processLine，再调用processCmd。processLocalCmd则调用了Driver类的run函数和runExcute函数。

直到：

while ((line = reader.readLine(curPrompt + "> ")) != null) {

表示重复请求读入 SQL>

1，cli/src/java   CliDriver.main是主函数。

[java] view plaincopy

1. public static void main(String[] args) throws Exception {
2. int ret = run(args);
3. System.exit(ret);
4. }

2，进入run函数

[java] view plaincopy

1. public static int run(String[] args) throws Exception {
2. OptionsProcessor oproc = new OptionsProcessor();

[java] view plaincopy

1. //（1） 解析(Parse)args，放入cmdLine，处理 –hiveconf var=val  用于增加或者覆盖hive/hadoop配置，设置到System的属性中。
2. if (!oproc.process_stage1(args)) {
3. return 1;
4. }
5. //（2） 配置log4j，加载hive-log4j.properties里的配置信息。

[java] view plaincopy

1. // NOTE: It is critical to do this here so that log4j is reinitialized
2. // before any of the other core hive classes are loaded
3. boolean logInitFailed = false;
4. String logInitDetailMessage;
5. try {
6. logInitDetailMessage = LogUtils.initHiveLog4j();
7. } catch (LogInitializationException e) {
8. logInitFailed = true;
9. logInitDetailMessage = e.getMessage();
10. }
11. //（3） 创建一个CliSessionState（SessionState）
12. CliSessionState ss = new CliSessionState(new HiveConf(SessionState.class));
13. ss.in = System.in;
14. try {
15. ss.out = new PrintStream(System.out, true, "UTF-8");
16. ss.err = new PrintStream(System.err, true, "UTF-8");
17. } catch (UnsupportedEncodingException e) {
18. return 3;
19. }

[java] view plaincopy

1. //（4） 处理-S， -e， -f， -h，-i等信息，保存在SessionState中。如果是-h，打印提示信息，并退出。
2. if (!oproc.process_stage2(ss)) {
3. return 2;
4. }
5. //(5) 如果不是-S，就是说不是静默状态，就输出一些提示信息，表示初始化好了。
6. if (!ss.getIsSilent()) {
7. if (logInitFailed) {
8. System.err.println(logInitDetailMessage);
9. } else {

[java] view plaincopy

1. //（5）输出一些信息：12/07/05 16:52:34 INFO SessionState:

[java] view plaincopy

1. SessionState.getConsole().printInfo(logInitDetailMessage);
2. }
3. }
4. //（6）创建一个HiveConf，通过命令行配置所有属性。
5. // set all properties specified via command line
6. HiveConf conf = ss.getConf();
7. for (Map.Entry<Object, Object> item : ss.cmdProperties.entrySet()) {
8. conf.set((String) item.getKey(), (String) item.getValue());
9. }
10. //（7）启动CliSessionState ss。
11. SessionState.start(ss);
12. // （8）连接到 Hive Server
13. if (ss.getHost() != null) {
14. ss.connect();
15. if (ss.isRemoteMode()) {
16. prompt = "[" + ss.host + ‘:‘ + ss.port + "] " + prompt;
17. char[] spaces = new char[prompt.length()];
18. Arrays.fill(spaces, ‘ ‘);
19. prompt2 = new String(spaces);
20. }
21. }

[java] view plaincopy

1. //（9） ShimLoader，load  HadoopShims
2. // CLI remote mode is a thin client: only load auxJars in local mode
3. if (!ss.isRemoteMode() && !ShimLoader.getHadoopShims().usesJobShell()) {
4. // hadoop-20 and above - we need to augment classpath using hiveconf
5. // components
6. // see also: code in ExecDriver.java
7. ClassLoader loader = conf.getClassLoader();

[java] view plaincopy

1. //（9）设置hiveJar= hive-exec-0.6.0.jar ，初始化加载hive-default.xml、 hive-site.xml。
2. String auxJars = HiveConf.getVar(conf, HiveConf.ConfVars.HIVEAUXJARS);
3. if (StringUtils.isNotBlank(auxJars)) {
4. loader = Utilities.addToClassPath(loader, StringUtils.split(auxJars, ","));
5. }
6. conf.setClassLoader(loader);
7. Thread.currentThread().setContextClassLoader(loader);
8. }
9. //（10） 创建CliDriver.
10. CliDriver cli = new CliDrive();
11. cli.setHiveVariables(oproc.getHiveVariables());
12. //（10）在接受hivesql命令前，执行一些初始化命令，这些命令存在文件中，文件可以通过-i选项设置，如果没有设置就去查找是否有$HIVE_HOME/bin/.hiverc和System.getProperty("user.home")/.hiverc两个文件，如果有就执行这两个文件中的命令。
13. // Execute -i init files (always in silent mode)
14. cli.processInitFiles(ss);
15. //（10） 如果是–e，执行命令并退出，如果是-f，执行文件中的命令并退出。
16. if (ss.execString != null) {
17. return cli.processLine(ss.execString);
18. }
19. try {
20. if (ss.fileName != null) {
21. return cli.processFile(ss.fileName);
22. }
23. } catch (FileNotFoundException e) {
24. System.err.println("Could not open input file for reading. (" + e.getMessage() + ")");
25. return 3;
26. }
27. //（11）创建ConsoleReader，读取用户输入，遇到“；”为一个完整的命令，执行该命令(CliDriver.processLine )，接着读取处理用户的输入。用户输入的命令记录在user.home/.hivehistory文件中。
28. ConsoleReader reader = new ConsoleReader();
29. reader.setBellEnabled(false);
30. // reader.setDebug(new PrintWriter(new FileWriter("writer.debug", true)));
31. reader.addCompletor(getCommandCompletor());
32. String line;
33. final String HISTORYFILE = ".hivehistory";
34. String historyFile = System.getProperty("user.home") + File.separator + HISTORYFILE;
35. reader.setHistory(new History(new File(historyFile)));
36. int ret = 0;
37. String prefix = "";
38. String curDB = getFormattedDb(conf, ss);
39. String curPrompt = prompt + curDB;
40. String dbSpaces = spacesForString(curDB);
41. while ((line = reader.readLine(curPrompt + "> ")) != null) {
42. if (!prefix.equals("")) {
43. prefix += ‘\n‘;
44. }
45. if (line.trim().endsWith(";") && !line.trim().endsWith("\\;")) {
46. line = prefix + line;
47. ret = cli.processLine(line, true);
48. prefix = "";
49. curDB = getFormattedDb(conf, ss);
50. curPrompt = prompt + curDB;
51. dbSpaces = dbSpaces.length() == curDB.length() ? dbSpaces : spacesForString(curDB);
52. } else {
53. prefix = prefix + line;
54. curPrompt = prompt2 + dbSpaces;
55. continue;
56. }
57. }
58. ss.close();
59. return ret;

3，主要是调用了 processLine。

ProcessLine又调用了 processCmd。

CliDriver.processLine   去掉命令末尾的;，

[java] view plaincopy

1. public int processLine(String line, boolean allowInterupting) {
2. SignalHandler oldSignal = null;
3. Signal interupSignal = null;
4. //(1)整理允许中断 ctrl+C
5. if (allowInterupting) {
6. // Remember all threads that were running at the time we started line processing.
7. // Hook up the custom Ctrl+C handler while processing this line
8. interupSignal = new Signal("INT");
9. oldSignal = Signal.handle(interupSignal, new SignalHandler() {
10. private final Thread cliThread = Thread.currentThread();
11. private boolean interruptRequested;
12. @Override
13. public void handle(Signal signal) {
14. boolean initialRequest = !interruptRequested;
15. interruptRequested = true;
16. // Kill the VM on second ctrl+c
17. if (!initialRequest) {
18. console.printInfo("Exiting the JVM");
19. System.exit(127);
20. }
21. // Interrupt the CLI thread to stop the current statement and return
22. // to prompt
23. console.printInfo("Interrupting... Be patient, this might take some time.");
24. console.printInfo("Press Ctrl+C again to kill JVM");
25. // First, kill any running MR jobs
26. HadoopJobExecHelper.killRunningJobs();
27. HiveInterruptUtils.interrupt();
28. this.cliThread.interrupt();
29. }
30. });
31. }
32. try {
33. int lastRet = 0, ret = 0;
34. String command = "";

[java] view plaincopy

1. //（2）循环处理每一个以分号结尾的语句。
2. for (String oneCmd : line.split(";")) {
3. if (StringUtils.endsWith(oneCmd, "\\")) {
4. command += StringUtils.chop(oneCmd) + ";";
5. continue;
6. } else {
7. command += oneCmd;
8. }
9. if (StringUtils.isBlank(command)) {
10. continue;
11. }
12. //（3）执行处理命令
13. ret = processCmd(command);
14. //（4）清除query State的状态。wipe cli query state
15. SessionState ss = SessionState.get();
16. ss.setCommandType(null);
17. command = "";
18. lastRet = ret;
19. boolean ignoreErrors = HiveConf.getBoolVar(conf, HiveConf.ConfVars.CLIIGNOREERRORS);
20. if (ret != 0 && !ignoreErrors) {
21. CommandProcessorFactory.clean((HiveConf) conf);
22. return ret;
23. }
24. }
25. CommandProcessorFactory.clean((HiveConf) conf);
26. return lastRet;
27. } finally {
28. // Once we are done processing the line, restore the old handler
29. if (oldSignal != null && interupSignal != null) {
30. Signal.handle(interupSignal, oldSignal);
31. }
32. }
33. }

4，processCmd

CliDriver.processCmd

Split命令，分析第一个单词： 
（1）如果是quit或者exit，退出。 
（2）source，执行文件中的HiveQL 
（3）！，执行命令，如!ls，列出当前目录的文件信息。 
（4）list，列出jar/file/archive。 
（5）如果是其他，则生成调用相应的CommandProcessor处理。

[java] view plaincopy

1. public int processCmd(String cmd) {
2. CliSessionState ss = (CliSessionState) SessionState.get();
3. String cmd_trimmed = cmd.trim();
4. String[] tokens = tokenizeCmd(cmd_trimmed);
5. int ret = 0;
6. //（1）如果是quit或者exit，退出。
7. if (cmd_trimmed.toLowerCase().equals("quit") || cmd_trimmed.toLowerCase().equals("exit")) {
8. // if we have come this far - either the previous commands
9. // are all successful or this is command line. in either case
10. // this counts as a successful run
11. ss.close();
12. System.exit(0);
13. //（2）source，执行文件中的HiveQL
14. } else if (tokens[0].equalsIgnoreCase("source")) {
15. String cmd_1 = getFirstCmd(cmd_trimmed, tokens[0].length());
16. File sourceFile = new File(cmd_1);
17. if (! sourceFile.isFile()){
18. console.printError("File: "+ cmd_1 + " is not a file.");
19. ret = 1;
20. } else {
21. try {
22. this.processFile(cmd_1);
23. } catch (IOException e) {
24. console.printError("Failed processing file "+ cmd_1 +" "+ e.getLocalizedMessage(),
25. org.apache.hadoop.util.StringUtils.stringifyException(e));
26. ret = 1;
27. }
28. }//（3）！，执行命令，如!ls，列出当前目录的文件信息。
29. } else if (cmd_trimmed.startsWith("!")) {
30. String shell_cmd = cmd_trimmed.substring(1);
31. shell_cmd = new VariableSubstitution().substitute(ss.getConf(), shell_cmd);
32. // shell_cmd = "/bin/bash -c \‘" + shell_cmd + "\‘";
33. try {
34. Process executor = Runtime.getRuntime().exec(shell_cmd);
35. StreamPrinter outPrinter = new StreamPrinter(executor.getInputStream(), null, ss.out);
36. StreamPrinter errPrinter = new StreamPrinter(executor.getErrorStream(), null, ss.err);
37. outPrinter.start();
38. errPrinter.start();
39. ret = executor.waitFor();
40. if (ret != 0) {
41. console.printError("Command failed with exit code = " + ret);
42. }
43. } catch (Exception e) {
44. console.printError("Exception raised from Shell command " + e.getLocalizedMessage(),
45. org.apache.hadoop.util.StringUtils.stringifyException(e));
46. ret = 1;
47. }
48. //（4）list，列出jar/file/archive。
49. } else if (tokens[0].toLowerCase().equals("list")) {
50. SessionState.ResourceType t;
51. if (tokens.length < 2 || (t = SessionState.find_resource_type(tokens[1])) == null) {
52. console.printError("Usage: list ["
53. + StringUtils.join(SessionState.ResourceType.values(), "|") + "] [<value> [<value>]*]");
54. ret = 1;
55. } else {
56. List<String> filter = null;
57. if (tokens.length >= 3) {
58. System.arraycopy(tokens, 2, tokens, 0, tokens.length - 2);
59. filter = Arrays.asList(tokens);
60. }
61. Set<String> s = ss.list_resource(t, filter);
62. if (s != null && !s.isEmpty()) {
63. ss.out.println(StringUtils.join(s, "\n"));
64. }
65. }//（5）如果是其他，则生成调用相应的CommandProcessor处理。//如果是远端
66. } else if (ss.isRemoteMode()) { // remote mode -- connecting to remote hive server
67. HiveClient client = ss.getClient();
68. PrintStream out = ss.out;
69. PrintStream err = ss.err;
70. try {
71. client.execute(cmd_trimmed);
72. List<String> results;
73. do {
74. results = client.fetchN(LINES_TO_FETCH);
75. for (String line : results) {
76. out.println(line);
77. }
78. } while (results.size() == LINES_TO_FETCH);
79. } catch (HiveServerException e) {
80. ret = e.getErrorCode();
81. if (ret != 0) { // OK if ret == 0 -- reached the EOF
82. String errMsg = e.getMessage();
83. if (errMsg == null) {
84. errMsg = e.toString();
85. }
86. ret = e.getErrorCode();
87. err.println("[Hive Error]: " + errMsg);
88. }
89. } catch (TException e) {
90. String errMsg = e.getMessage();
91. if (errMsg == null) {
92. errMsg = e.toString();
93. }
94. ret = -10002;
95. err.println("[Thrift Error]: " + errMsg);
96. } finally {
97. try {
98. client.clean();
99. } catch (TException e) {
100. String errMsg = e.getMessage();
101. if (errMsg == null) {
102. errMsg = e.toString();
103. }
104. err.println("[Thrift Error]: Hive server is not cleaned due to thrift exception: "
105. + errMsg);
106. }
107. }//如果是本地
108. } else { // local mode
109. CommandProcessor proc = CommandProcessorFactory.get(tokens[0], (HiveConf) conf);
110. ret = processLocalCmd(cmd, proc, ss);
111. }
112. return ret;
113. }

[java] view plaincopy

1. <span style="font-family:Arial, Helvetica, sans-serif;"><span style="white-space: normal;">
2. </span></span>

[java] view plaincopy

1. <span style="font-family:Arial, Helvetica, sans-serif;"><span style="white-space: normal;">
2. </span></span>

5，processLoacalCmd

[java] view plaincopy

1. int processLocalCmd(String cmd, CommandProcessor proc, CliSessionState ss) {
2. int tryCount = 0;
3. boolean needRetry;
4. int ret = 0;
5. do {
6. try {
7. needRetry = false;
8. if (proc != null) {
9. if (proc instanceof Driver) {
10. Driver qp = (Driver) proc;
11. PrintStream out = ss.out;
12. long start = System.currentTimeMillis();
13. if (ss.getIsVerbose()) {
14. out.println(cmd);
15. }
16. qp.setTryCount(tryCount);
17. ret = qp.run(cmd).getResponseCode();
18. if (ret != 0) {
19. qp.close();
20. return ret;
21. }
22. ArrayList<String> res = new ArrayList<String>();
23. printHeader(qp, out);
24. try {
25. while (qp.getResults(res)) {
26. for (String r : res) {
27. out.println(r);
28. }
29. res.clear();
30. if (out.checkError()) {
31. break;
32. }
33. }
34. } catch (IOException e) {
35. console.printError("Failed with exception " + e.getClass().getName() + ":"
36. + e.getMessage(), "\n"
37. + org.apache.hadoop.util.StringUtils.stringifyException(e));
38. ret = 1;
39. }
40. int cret = qp.close();
41. if (ret == 0) {
42. ret = cret;
43. }
44. long end = System.currentTimeMillis();
45. if (end > start) {
46. double timeTaken = (end - start) / 1000.0;
47. console.printInfo("Time taken: " + timeTaken + " seconds", null);
48. }
49. } else {
50. String firstToken = tokenizeCmd(cmd.trim())[0];
51. String cmd_1 = getFirstCmd(cmd.trim(), firstToken.length());
52. if (ss.getIsVerbose()) {
53. ss.out.println(firstToken + " " + cmd_1);
54. }
55. ret = proc.run(cmd_1).getResponseCode();
56. }
57. }
58. } catch (CommandNeedRetryException e) {
59. console.printInfo("Retry query with a different approach...");
60. tryCount++;
61. needRetry = true;
62. }
63. } while (needRetry);
64. return ret;
65. }

6，Driver 类 的run 方法。

Driver 
Driver.run(String command) // 处理一条命令 
{ 
  int ret =compile(command);  // 分析命令，生成Task。 
  ret = execute();  // 运行Task。 
}

Driver.compile 
 
Driver.compile(String command) // 处理一条命令 
{ 
(1) Parser(antlr):HiveQL->Abstract?Syntax?Tree?(AST) 
      ParseDriver pd = new ParseDriver(); 
      ASTNode tree = pd.parse(command, ctx); 
(2) SemanticAnalyzer 
      BaseSemanticAnalyzer sem = SemanticAnalyzerFactory.get(conf, tree); 
      // Do semantic analysis and plan generation 
      sem.analyze(tree, ctx); 
}

7，plan生成位置

可以通过跟踪到Driver.java文件的Line 663可知其路径为：

/tmp/hive-gexing111/hive_2012-07-09_10-37-27_511_5073252372102100766/test2.py

如果系统自己的plan：

/tmp/hive-gexing111/hive_2012-07-09_12-45-55_479_6444298560478274273/-local-10002/plan.xml

8，Debug  show tables "ge*";

在hive/   metastore/src/java   /com.aliyun.apsara.odps.metastore.ots   /OTSObjectStore.java

中的Gettables;

9，配置文件位置

hive/build/dist/conf/hive-site.xml

设置为改写版本：

<property>
  <name>com.aliyun.odps.mode</name>
  <value>true</value>
</property>

<property>
<name>native.hive.mode</name>
<value>false</value>
</property>