HiveSQL解析/执行计划生成流程分析

最近在研究impala，还是先回顾下hive的sql执行流程吧。 hive有三种用户接口： cli (command line interface) bin/hive或bin/hive –service cli 命令行方式（默认） hive-server/hive-server2 bin/hive –service hiveserver 或bin/hive –service hiveserve

最近在研究impala，还是先回顾下hive的sql执行流程吧。

hive有三种用户接口：

cli (command line interface)	bin/hive或bin/hive –service cli	命令行方式（默认）
hive-server/hive-server2	bin/hive –service hiveserver 或bin/hive –service hiveserver2	通过jdbc/odbc和thrift访问（impala通过这种方式借用hive-metastore）
hwi (hive web interface)	bin/hive –service hwi	通过浏览器访问

在hive shell中输入“show tables;”实际执行的是：

bin/hadoop jar hive/lib/hive-cli-0.9.0.jar org.apache.hadoop.hive.cli.clidriver -e 'show tables;'

cli入口函数：cli.clidriver.main()

读入参数->建立sessionstate并导入配置->处理输入文件中指令clidriver.processfile()；或交互型指令clidriver.processline()->解析输入clidriver.processcmd()

(1) 如果是quit或者exit，退出

(2) 以source开头的，读取外部文件并执行文件中的hiveql

(3) ！开头的命令，执行操作系统命令（如!ls，列出当前目录的文件信息）

(4) list，列出jar/file/archive

(5) 其他命令，则生成调用相应的commandprocessor处理，进入clidriver.processlocalcmd()

下面看看clidriver.processlocalcmd()这个函数：

set/dfs/add/delete指令交给指定的commandprocessor处理，其余的交给org.apache.hadoop.hive.ql.driver.run()处理

org.apache.hadoop.hive.ql.driver类是查询的起点，run()方法会先后调用compile()和execute()两个函数来完成查询，所以一个command的查询分为compile和execute两个阶段。

compile

(1)利用antlr生成的hivelexer.java和hiveparser.java类，将hiveql转换成抽象语法树（ast）。

首先使用antlr工具将srcqlsrcjavaorgapachehadoophiveqlparsehive.g编译成以下几个文件：hiveparser.java,?hive.tokens,?hive__.g,?hivelexer.java

hivelexer.java和hiveparser.java分别是词法和语法分析类文件，hive__.g是hivelexer.java对应的词法分析规范，hive.tokens定义了词法分析后所有的token。

然后沿着“driver.compile()->parsedriver.parse(command, ctx)->hiveparserx.statement()->antlr中的api”这个调用关系把输入的hiveql转化成astnode类型的语法树。hiveparserx是由antlr生成的hiveparser类的子类。

(2)利用对应的semanticanalyzer类，将ast树转换成map-reduce task。主要分为三个步骤：

a) ast -> operator dag

b) optimize operator dag

c) oprator dag -> map-reduce task

首先接着上一步生成的语法树astnode，?semanticanalyzerfactory会根据astnode的token类型生成不同的semanticanalyzer (所有这些semanticanalyzer都继承自basesemanticanalyzer)

1) explainsemanticanalyzer

2) loadsemanticanalyzer

3) exportsemanticanalyzer

4) ddlsemanticanalyzer

5) functionsemanticanalyzer

6) semanticanalyzer

然后调用basesemanticanalyzer.analyze()->basesemanticanalyzer. analyzeinternal()。

下面以最常见的select * from table类型的查询为例，进入的子类是semanticanalyzer. analyzeinternal()，这个函数的逻辑如下：

1) dophase1()：将sql语句中涉及到的各种信息存储起来，存到qb中去，留着后面用。

2) getmetadata()：获取元数据信息，主要是sql中涉及到的 表 和 元数据 的关联

3) genplan()：生成operator tree/dag

4) optimize：优化，对operator tree/dag 进行一些优化操作，例如列剪枝等（目前只能做rule-based optimize，不能做cost-based optimize）

5) genmapredtasks()：将operator tree/dag 通过一定的规则生成若干相互依赖的mr任务

execute

将compile阶段生成的task信息序列化到plan.xml，然后启动map-reduce，在configure时反序列化plan.xml

实例分析：

在hive中有这样一张表：

uid	fruit_name	count
a	apple	5
a	orange	3
a	apple	2
b	banana	1

执行如下的查询：

select uid, sum(count) from logs group by uid

通过explain命令可以查看执行计划：

explain select uid, sum(count) from logs group by uid;

依照hive.g的语法规则，生成ast如下:

abstract syntax tree: ( tok_query (tok_from (tok_tabref (tok_tabname logs))) ( tok_insert (tok_destination (tok_dir tok_tmp_file)) ( tok_select (tok_selexpr (tok_table_or_col uid)) (tok_selexpr (tok_function sum (tok_table_or_col count))) ) (tok_groupby (tok_table_or_col uid)) ) )

生成的执行计划operator tree/dag如下：

stage dependencies: stage-1 is a root stage stage-0 is a root stage stage plans: stage: stage-1 map reduce alias -> map operator tree: logs tablescan // 扫描表 alias: logs select operator //选择字段 expressions: expr: uid type: string expr: count type: int outputcolumnnames: uid, count group by operator //在map端先做一次聚合，减少shuffle数据量 aggregations: expr: sum(count) //聚合函数 bucketgroup: false keys: expr: uid type: string mode: hash outputcolumnnames: _col0, _col1 reduce output operator //输出key，value给reduce key expressions: expr: _col0 type: string sort order: + map-reduce partition columns: expr: _col0 type: string tag: -1 value expressions: expr: _col1 type: bigint reduce operator tree: group by operator aggregations: expr: sum(value._col0) //聚合 bucketgroup: false keys: expr: key._col0 type: string mode: mergepartial outputcolumnnames: _col0, _col1 select operator //选择字段 expressions: expr: _col0 type: string expr: _col1 type: bigint outputcolumnnames: _col0, _col1 file output operator //输出到文件 compressed: false globaltableid: 0 table: input format: org.apache.hadoop.mapred.textinputformat output format: org.apache.hadoop.hive.ql.io.hiveignorekeytextoutputformat stage: stage-0 fetch operator limit: -1

hive优化策略：

1. 去除查询中不需要的column

2. where条件判断等在tablescan阶段就进行过滤

3. 利用partition信息，只读取符合条件的partition

4. map端join，以大表作驱动，小表载入所有mapper内存中

5. 调整join顺序，确保以大表作为驱动表

6. 对于数据分布不均衡的表group by时，为避免数据集中到少数的reducer上，分成两个map-reduce阶段。第一个阶段先用distinct列进行shuffle，然后在reduce端部分聚合，减小数据规模，第二个map-reduce阶段再按group-by列聚合。

7. 在map端用hash进行部分聚合，减小reduce端数据处理规模。

参考文献：

http://fatkun.com/2013/01/hive-group-by.html

【说明】：本文章由站长整理发布，文章内容不代表本站观点，如文中有侵权行为，请与本站客服联系(QQ:)！