hive执行流程(1)-hive入口CliDriver类分析
在运行hive cli命令时,调用hadoop jar hive-cli-0.13.1.jar org.apache.hadoop.hive.cli.CliDriver xxxx 命令,而org.apache.hadoop.util.RunJar方法其实是封装了反射调用,最终是调用org.apache.hadoop.hive.cli.CliDriver类的main方法. CliDriver类是hive的入口类。 首先CliDriver类会通过OptionsProcessor类来parse输入的命令。比如解析-e,-s,-h等参数,然后把对应的值存放到对应的CliSessionState类的属性中,最后应用于CliDriver类中。 比如在executeDriver方法中,根据CliSessionState的属性对命令进行处理 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 CliDrivercli= new CliDriver(); cli.setHiveVariables(oproc.getHiveVariables()); //有变量相关的设置时 //usethespecifieddatabaseifspecified cli.processSelectDatabase(ss); //Execute-iinitfiles(alwaysinsilentmode) cli.processInitFiles(ss); //指定了-i和加载.hiverc文件 if (ss.execString!= null ){ //指定了-e时 int cmdProcessStatus=cli.processLine(ss.execString); return cmdProcessStatus; } try { //指定了-f时 if (ss.fileName!= null ){ return cli.processFile(ss.fileName); } } catch (FileNotFoundExceptione){ System.err.println( "Couldnotopeninputfileforreading.(" +e.getMessage()+ ")" ); return 3 ; } 在CliDriver类方法的调用顺序主要有下面几种 1)add xxx/set/compile/reset等命令 1 main-->run--->executeDriver---->processLine--->processCmd--->processLocalCmd--对应processor类的run方法 2)sql命令 1 main-->run--->executeDriver---->processLine--->processCmd--->processLocalCmd---->Driver类run方法 3)shell命令 1 main-->run--->executeDriver---->processLine--->processCmd 其中CliDriver类中最重要的方法是processCmd,其定义了不同的命令不同的执行方式: 具体实现: 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 public int processCmd(Stringcmd){ CliSessionStatess=(CliSessionState)SessionState.get(); ss.setLastCommand(cmd); //Flushtheprintstream,soitdoesn'tincludeoutputfromthelastcommand ss.err.flush(); Stringcmd_trimmed=cmd.trim(); String[]tokens=tokenizeCmd(cmd_trimmed); int ret= 0 ; if (cmd_trimmed.toLowerCase().equals( "quit" )||cmd_trimmed.toLowerCase().equals( "exit" )){ //如果是quit或者是exit,则直接退出jvm ss.close(); System.exit( 0 ); } else if (tokens[ 0 ].equalsIgnoreCase( "source" )){ //如果是sourcexxx的情况,则按文件处理(调用processFile方法) Stringcmd_1=getFirstCmd(cmd_trimmed,tokens[ 0 ].length()); FilesourceFile= new File(cmd_1); if (!sourceFile.isFile()){ console.printError( "File:" +cmd_1+ "isnotafile." ); ret= 1 ; } else { try { this .processFile(cmd_1); } catch (IOExceptione){ console.printError( "Failedprocessingfile" +cmd_1+ "" +e.getLocalizedMessage(), stringifyException(e)); ret= 1 ; } } } else if (cmd_trimmed.startsWith( "!" )){ //以!开头的,做为shell命令执行,最终调用Runtime.getRuntime().exec(shell_cmd) Stringshell_cmd=cmd_trimmed.substring( 1 ); shell_cmd= new VariableSubstitution().substitute(ss.getConf(),shell_cmd); //这里也会进行变量替换 //shell_cmd="/bin/bash-c\'"+shell_cmd+"\'"; try { Processexecutor=Runtime.getRuntime().exec(shell_cmd); StreamPrinteroutPrinter= new StreamPrinter(executor.getInputStream(), null ,ss.out); StreamPrintererrPrinter= new StreamPrinter(executor.getErrorStream(), null ,ss.err); outPrinter.start(); errPrinter.start(); ret=executor.waitFor(); if (ret!= 0 ){ console.printError( "Commandfailedwithexitcode=" +ret); } } catch (Exceptione){ console.printError( "ExceptionraisedfromShellcommand" +e.getLocalizedMessage(), stringifyException(e)); ret= 1 ; } } else if (tokens[ 0 ].toLowerCase().equals( "list" )){ //list命令时,调用SessionState的list_resource方法 SessionState.ResourceTypet; if (tokens.length< 2 ||(t=SessionState.find_resource_type(tokens[ 1 ]))== null ){ console.printError( "Usage:list[" +StringUtils.join(SessionState.ResourceType.values(), "|" )+ "][<value>[<value>]*]" ); ret= 1 ; } else { List<String>filter= null ; if (tokens.length>= 3 ){ System.arraycopy(tokens, 2 ,tokens, 0 ,tokens.length- 2 ); filter=Arrays.asList(tokens); } Set<String>s=ss.list_resource(t,filter); if (s!= null &&!s.isEmpty()){ ss.out.println(StringUtils.join(s, "\n" )); } } } else if (ss.isRemoteMode()){ //remotemode--connectingtoremotehiveserver//如果是远程模式,即hiveserver,调用HiveClient类的execute方法 HiveClientclient=ss.getClient(); PrintStreamout=ss.out; PrintStreamerr=ss.err; try { client.execute(cmd_trimmed); List<String>results; do { results=client.fetchN(LINES_TO_FETCH); for (Stringline:results){ out.println(line); } } while (results.size()==LINES_TO_FETCH); } catch (HiveServerExceptione){ ret=e.getErrorCode(); if (ret!= 0 ){ //OKifret==0--reachedtheEOF StringerrMsg=e.getMessage(); if (errMsg== null ){ errMsg=e.toString(); } ret=e.getErrorCode(); err.println( "[HiveError]:" +errMsg); } } catch (TExceptione){ StringerrMsg=e.getMessage(); if (errMsg== null ){ errMsg=e.toString(); } ret=- 10002 ; err.println( "[ThriftError]:" +errMsg); } finally { try { client.clean(); } catch (TExceptione){ StringerrMsg=e.getMessage(); if (errMsg== null ){ errMsg=e.toString(); } err.println( "[ThriftError]:Hiveserverisnotcleanedduetothriftexception:" +errMsg); } } } else { //localmode//剩下的情况都作为local模式,比如addxxx,setxxxx,select/insertxxx/showtables/createtable,databse/usexxx等命令。 try { CommandProcessorproc=CommandProcessorFactory.get(tokens,(HiveConf)conf); //会先根据命令获取对应的CommandProcessor实现类 ret=processLocalCmd(cmd,proc,ss); //并调用processLocalCmd方法 } catch (SQLExceptione){ console.printError( "Failedprocessingcommand" +tokens[ 0 ]+ "" +e.getLocalizedMessage(), org.apache.hadoop.util.StringUtils.stringifyException(e)); ret= 1 ; } } return ret; } 而processLocalCmd方法会将CommandProcessor的实例作为参数传入,并根据不同的CommandProcessor实现类,来调用不同的类的run方法。 1 int processLocalCmd(Stringcmd,CommandProcessorproc,CliSessionStatess) 本文转自菜菜光 51CTO博客,原文链接:http://blog.51cto.com/caiguangguang/1542275,如需转载请自行联系原作者