大数据SQL调优专题——Hive执行原理
引入
Apache Hive 是基于Hadoop的数据仓库工具，它可以使用SQL来读取、写入和管理存在分布式文件系统中的海量数据。在Hive中，HQL默认转换成MapReduce程序运行到Yarn集群中，大大降低了非Java开发者数据分析的门槛，并且Hive提供命令行工具和JDBC驱动程序，方便用户连接到Hive进行数据分析操作。
严格意义上，Hive并不属于计算引擎，而是建立在Hadoop生态之上的数据仓库管理工具。它将繁杂的MapReduce作业抽象成SQL，使得开发及维护成本大幅降低。得益于HDFS的存储和MapReduce的读写能力，Hive展现出了强大的兼容能力、数据吞吐能力和服务稳定性，时至今日依然是大数据架构中不可或缺的一部分。
Hive的核心特点
-
Hive是基于Hadoop的数仓工具，底层数据存储在HDFS中；
-
Hive提供标准SQL功能，支持SQL语法访问操作数据；
-
Hive适合OLAP数据分析场景，不适合OLTP数据处理场景，所以适合数据仓库构建；
-
HQL默认转换成MapReduce任务执行，也可以配置转换成Apache Spark、Apache Tez任务运行；
-
Hive中支持定义UDF、UDAF、UDTF函数扩展功能。
Hive的架构设计
Hive用户接口
访问Hive可以通过CLI、Beeline、JDBC/ODBC、WebUI几种方式。在Hive早期版本中可以使用Hive CLI来操作Hive，Hive CLI并发性能差、脚本执行能力有限并缺乏JDBC驱动支持，从Hive 4.x版本起废弃了Hive CLI推荐使用Beeline。Beeline是一个基于JDBC的Hive客户端，支持并发环境、复杂脚本执行、JDBC驱动等，在Hive集群内连接Hive可以使用Beeline方式。在Hive集群外，通过代码或者工具连接操作Hive时可以通过JDBC/ODBC方式。通过WebUI方式可以通过浏览器查看到Hive集群的一些信息。
HiveServer2服务
HiveServer2服务提供JDBC/ODBC接口，主要用于代理远程客户端对Hive的访问，是一种基于Thrift协议的服务。例如通过JDBC或者Beeline连接访问Hive时就需要启动HiveServer2服务，就算Beeline访问本机上的Hive服务也需要启动HiveServer2服务。
HiveServer2代理远程客户端对Hive操作时会涉及到操作HDFS数据，就会有操作权限问题，那么操作HDFS中数据的用户是启动HiveServer2的用户还是远程客户端的用户需要通过“hive.server2.enable.doAs” 参数决定，该参数默认为true，表示HiveServer2操作HDFS时的用户为远程客户端用户，如果设置为false表示操作HDFS数据的用户为启动HiveServer2的用户。
MetaStore服务
MetaStore服务负责存储和管理Hive元数据，为HiverServer2提供元数据访问接口。Hive中的元数据包括表的名字，表的列和分区及其属性，表的属性（表拥有者、是否为外部表等），表的数据所在目录等。
Hive MetaStore可以将元数据存储在mysql、derby数据库中。
Hive Driver
Driver中包含解释器（SQL Parser）、编译器（Compiler）、优化器（Optimizer），负责完成HQL查询语句从词法分析、语法分析、编译、优化以及查询计划的生成。生成的查询计划存储在HDFS中，并在随后有执行器（Executor）调用MapReduce执行。
对于Hive有了一个初步认识，我们下面开始梳理Hive的执行原理。
Hive的执行原理
Hive无论采用哪种调用方式，最终都会辗转到org.apache.hadoop.hive.ql.Driver类。SQL语句在Driver类中，通过Antlr框架进行解析编译，将一条SQL按照如下流程转换成最终执行的MapReduce任务。
如果直接盲目的去看Driver类的代码，会很容易看懵逼，我们需要再往前一点。
SQLOperation
先看org.apache.hive.service.cli.operation.SQLOperation 类，它负责创建Driver对象、编译SQL、异步执行SQL。其中核心的就是 runInternal()方法，主要进行如下两个步骤：
- Driver对象创建并编译SQL，将SQL编译成Query Plan执行计划。
- 对QueryPaln 进行处理，转换成MR 任务执行。
runInternal() 方法源码内容如下：
/**
* 内部运行方法，用于执行SQL操作。
*
* @throws HiveSQLException 如果在执行过程中发生Hive SQL异常。
*/
public void runInternal() throws HiveSQLException {
// 设置操作状态为PENDING
setState(OperationState.PENDING);
// 判断是否应该异步运行
boolean runAsync = shouldRunAsync();
// 判断是否应该异步编译
final boolean asyncPrepare = runAsync
&& HiveConf.getBoolVar(queryState.getConf(),
HiveConf.ConfVars.HIVE_SERVER2_ASYNC_EXEC_ASYNC_COMPILE);
// 如果不是异步编译，则同步准备查询
if (!asyncPrepare) {
//创建Driver对象，编译SQL
//Driver经过：SQL -> AST(抽象语法树) -> QueryBlock(查询块) -> Operator(e逻辑执行计划) -> TaskTree(物理执行计划) -> QueryPlan(查询计划)
prepare(queryState);
}
// 如果不是异步运行，则同步运行查询
if (!runAsync) {
runQuery();
} else {
// 我们将在后台线程中传递ThreadLocals，从前台（处理程序）线程传递。
// 1) ThreadLocal Hive对象需要在后台线程中设置
// 2) Hive中的元数据存储客户端与正确的用户相关联。
// 3) 当前UGI将在元数据存储处于嵌入式模式时被元数据存储使用
Runnable work = new BackgroundWork(getCurrentUGI(), parentSession.getSessionHive(),
SessionState.getPerfLogger(), SessionState.get(), asyncPrepare);
try {
// 如果没有可用的后台线程来运行此操作，此提交将阻塞
Future<?> backgroundHandle = getParentSession().submitBackgroundOperation(work);
// 设置后台操作句柄
setBackgroundHandle(backgroundHandle);
} catch (RejectedExecutionException rejected) {
// 设置操作状态为ERROR
setState(OperationState.ERROR);
// 抛出HiveSQLException异常
throw new HiveSQLException("The background threadpool cannot accept" +
" new task for execution, please retry the operation", rejected);
}
}
}
1.Driver对象创建并编译SQL，将SQL编译成Query Plan执行计划
其中核心的是prepare()方法，它的源码在2.x和3.x、4.x有一些区别，不过其核心功能是没变的，主要是创建Driver对象，并编译SQL，然后通过Driver将SQL最终转换成Query Plan。
prepare()方法3.x的源码如下：
/**
* 准备执行SQL查询的操作。
* 此方法负责初始化Driver，设置查询超时，编译查询语句，并处理可能的异常。
*
* @param queryState 包含查询状态信息的对象。
* @throws HiveSQLException 如果在准备过程中发生Hive SQL异常。
*/
public void prepare(QueryState queryState) throws HiveSQLException {
// 设置操作状态为运行中
setState(OperationState.RUNNING);
try {
// 创建Driver实例，返回的Driver对象是 ReExecDriver
driver = DriverFactory.newDriver(queryState, getParentSession().getUserName(), queryInfo);
// 如果查询超时时间大于0，则启动一个定时任务来取消查询
if (queryTimeout > 0) {
// 创建一个单线程的定时任务执行器
timeoutExecutor = new ScheduledThreadPoolExecutor(1);
// 创建一个定时任务，在查询超时后取消查询
Runnable timeoutTask = new Runnable() {
@Override
public void run() {
try {
// 获取查询ID
String queryId = queryState.getQueryId();
// 记录日志，查询超时并取消执行
LOG.info("Query timed out after: " + queryTimeout
+ " seconds. Cancelling the execution now: " + queryId);
// 取消查询
SQLOperation.this.cancel(OperationState.TIMEDOUT);
} catch (HiveSQLException e) {
// 记录日志，取消查询时发生错误
LOG.error("Error cancelling the query after timeout: " + queryTimeout + " seconds", e);
} finally {
// 关闭定时任务执行器
timeoutExecutor.shutdown();
}
}
};
// 安排定时任务在查询超时后执行
timeoutExecutor.schedule(timeoutTask, queryTimeout, TimeUnit.SECONDS);
}
// 设置查询显示信息
queryInfo.setQueryDisplay(driver.getQueryDisplay());
// 设置操作句柄信息，以便Thrift API用户可以使用操作句柄查找Yarn ATS中的查询信息
String guid64 = Base64.encodeBase64URLSafeString(getHandle().getHandleIdentifier()
.toTHandleIdentifier().getGuid()).trim();
driver.setOperationId(guid64);
// 编译SQL查询并响应 ReExecDriver.compileAndRespond(...) -> Driver.compileAndRespond(...)
response = driver.compileAndRespond(statement);
// 如果响应代码不为0，则抛出异常
if (0 != response.getResponseCode()) {
throw toSQLException("Error while compiling statement", response);
}
// 设置是否有结果集
setHasResultSet(driver.hasResultSet());
} catch (HiveSQLException e) {
// 设置操作状态为错误
setState(OperationState.ERROR);
// 抛出异常
throw e;
} catch (Throwable e) {
// 设置操作状态为错误
setState(OperationState.ERROR);
// 抛出异常
throw new HiveSQLException("Error running query: " + e.toString(), e);
}
}
2.x与3.x源码最核心的区别就是在创建Driver，其对应源码是：
driver = new Driver(queryState, getParentSession().getUserName());而4.x与3.x源码最核心的区别如下：
- 利用 Java 8 的 Lambda 表达式特性，简化代码逻辑，提高代码的可读性和可维护性。
- 通过将 queryTimeout 的类型改为 long，支持了更大的超时值，避免了溢出问题。
- 在资源管理方面，对调度器的生命周期管理也进行了优化，不需要显式的关闭操作。
4.x对应源码是：
if (queryTimeout > 0L) { timeoutExecutor = Executors.newSingleThreadScheduledExecutor(); timeoutExecutor.schedule(() -> { try { final String queryId = queryState.getQueryId(); log.info("Query timed out after: {} seconds. Cancelling the execution now: {}", queryTimeout, queryId); SQLOperation.this.cancel(OperationState.TIMEDOUT); } catch (HiveSQLException e) { log.error("Error cancelling the query after timeout: {} seconds", queryTimeout, e); } return null; }, queryTimeout, TimeUnit.SECONDS); }
在DriverFactory.newDriver()方法中返回 ReExecDriver对象，该对象表示执行过程失败可重试的Driver对象，然后调用 Driver.compileAndRespond() 方法进行编译SQL。
2.对QueryPaln 进行处理，转换成MR 任务执行
BackgroundWork是一个线程，负责异步处理QueryPlan，通过submitBackgroundOperation(work)提交运行，执行到SQLOperator.BackgroundOperation.run()方法，最终调用到Driver.run() 方法。
Driver
下面我们再来Driver类，它在不同版本中也有一些差别，比如2.x版本是直接 implements CommandProcessor，而3.x和4.x版本则是implements IDriver，而IDriver 则是 extends CommandProcessor。本质是为了更好的解耦和扩展性，使得代码更加模块化，易于维护和扩展。同时，通过继承 CommandProcessor 接口，也保持了与旧版本的兼容性，确保了功能的连续性。不过其核心功能是没变的，主要包含编译、优化及执行。
为了方便理解，我们先梳理整个执行步骤如下：
-
通过Antlr解析SQL语法规则和语法解析，将SQL语法转换成AST(抽象语法树) 。
-
遍历AST(抽象语法树) 将其转化成Query Block（查询块，可以看成查询基本执行单元）。
-
将Query Block(查询块) 转换成OperatorTree（逻辑执行计划），并进行优化。
-
OperatorTree(逻辑执行计划)转换成TaskTree(物理执行计划，每个Task对应一个MR Job任务)。
-
TaskTree(物理执行计划)最终包装成Query Plan(查询计划)。
简单总结执行流程如下：
SQL -> AST(抽象语法树) -> QueryBlock(查询块) -> Operator(逻辑执行计划) -> TaskTree(物理执行计划) -> QueryPlan(查询计划)。
下面我们再结合SQLOperation调用的Driver类里面的核心方法，来看看底层源码是如何实现的：
compileAndRespond方法
首先第一个核心方法是
response = driver.compileAndRespond(statement);compileAndRespond()方法2.x源码如下：
/**
* 编译给定的 SQL 命令并返回一个命令处理器响应。
* 此方法调用 compileInternal 方法进行实际的编译操作，并使用编译结果创建一个命令处理器响应。
*
* @param command 要编译的 SQL 命令
* @return 包含编译结果的命令处理器响应
*/
public CommandProcessorResponse compileAndRespond(String command) {
return createProcessorResponse(compileInternal(command, false));
}
3.x和4.x会有些区别，会返回以下方法的调用结果：
coreDriver.compileAndRespond(statement);无论哪个版本，最终compileAndRespond()方法都会调用到 compileInternal()方法，我们继续看2.x版本compileInternal()方法源码如下：
private int compileInternal(String command, boolean deferClose) {
int ret;
// 获取Metrics实例，如果存在则增加等待编译操作的计数器
Metrics metrics = MetricsFactory.getInstance();
if(metrics != null) {
metrics.incrementCounter(MetricsConstant.WAITING_COMPILE_OPS, 1);
}
// 尝试获取编译锁，如果获取失败则返回编译锁超时错误码
final ReentrantLock compileLock = tryAcquireCompileLock(isParallelEnabled, command);
if(compileLock == null) {
return ErrorMsg.COMPILE_LOCK_TIMED_OUT.getErrorCode();
}
try {
// 如果Metrics实例存在，减少等待编译操作的计数器
if(metrics != null) {
metrics.decrementCounter(MetricsConstant.WAITING_COMPILE_OPS, 1);
}
// 进行Hive SQL编译
ret = compile(command, true, deferClose);
} finally {
// 无论编译结果如何，最终都要释放编译锁
compileLock.unlock();
}
// 如果编译失败，尝试释放锁并回滚事务
if(ret != 0) {
try {
releaseLocksAndCommitOrRollback(false, null);
} catch(LockException e) {
// 记录释放锁时的异常信息
LOG.warn("Exception in releasing locks. " + org.apache.hadoop.util.StringUtils.stringifyException(e));
}
}
// 保存编译时的性能日志，用于WebUI展示
// 执行时的性能日志由另一个线程的PerfLogger或重置后的PerfLogger完成
PerfLogger perfLogger = SessionState.getPerfLogger();
queryDisplay.setPerfLogStarts(QueryDisplay.Phase.COMPILATION, perfLogger.getStartTimes());
queryDisplay.setPerfLogEnds(QueryDisplay.Phase.COMPILATION, perfLogger.getEndTimes());
return ret;
}
3.x有一些区别，但是都是通过执行Driver.compile()方法，而4.x则是解耦了，执行的是Compiler.compile()。
compile方法
核心都是compile()方法，compile()方法2.x源码如下：
/**
* 编译一个新的查询，可选择重置任务ID计数器并决定是否延迟关闭。
*
* @param command 要编译的HiveQL查询。
* @param resetTaskIds 如果为true，则重置任务ID计数器。
* @param deferClose 如果为true，则在编译过程被中断时延迟关闭/销毁操作。
* @return 0表示编译成功，否则返回错误代码。
*/
public int compile(String command, boolean resetTaskIds, boolean deferClose) {
// 获取性能日志记录器，并开始记录编译过程的性能
PerfLogger perfLogger = SessionState.getPerfLogger(true);
perfLogger.PerfLogBegin(CLASS_NAME, PerfLogger.DRIVER_RUN);
perfLogger.PerfLogBegin(CLASS_NAME, PerfLogger.COMPILE);
// 锁定驱动状态，将驱动状态设置为编译中
lDrvState.stateLock.lock();
try {
lDrvState.driverState = DriverState.COMPILING;
} finally {
lDrvState.stateLock.unlock();
}
// 对查询命令进行变量替换
command = new VariableSubstitution(new HiveVariableSource() {
@Override
public Map<String, String> getHiveVariable() {
return SessionState.get().getHiveVariables();
}
}).substitute(conf, command);
// 存储查询字符串
String queryStr = command;
try {
// 对查询命令进行脱敏处理，避免记录敏感数据
queryStr = HookUtils.redactLogString(conf, command);
} catch(Exception e) {
// 若脱敏失败，记录警告信息
LOG.warn("WARNING! Query command could not be redacted." + e);
}
// 检查编译过程是否被中断，若中断则处理中断并返回错误代码
if(isInterrupted()) {
return handleInterruption("at beginning of compilation."); //indicate if need clean resource
}
// 如果上下文不为空且解释分析状态不为运行中，则关闭现有上下文
if(ctx != null && ctx.getExplainAnalyze() != AnalyzeState.RUNNING) {
// close the existing ctx etc before compiling a new query, but does not destroy driver
closeInProcess(false);
}
// 如果需要重置任务ID，则重置任务工厂的ID
if(resetTaskIds) {
TaskFactory.resetId();
}
// 获取查询ID
String queryId = conf.getVar(HiveConf.ConfVars.HIVEQUERYID);
// 保存查询信息，用于Web UI显示
this.queryDisplay.setQueryStr(queryStr);
this.queryDisplay.setQueryId(queryId);
// 记录编译开始信息
LOG.info("Compiling command(queryId=" + queryId + "): " + queryStr);
// 设置查询的当前时间戳
SessionState.get().setupQueryCurrentTimestamp();
// 标记编译过程中是否发生错误
boolean compileError = false;
try {
// 初始化事务管理器
final HiveTxnManager txnManager = SessionState.get().initTxnMgr(conf);
// 移除旧的关闭hook
ShutdownHookManager.removeShutdownHook(shutdownRunner);
// 创建新的关闭hook，用于在JVM关闭时释放锁
shutdownRunner = new Runnable() {
@Override
public void run() {
try {
releaseLocksAndCommitOrRollback(false, txnManager);
} catch(LockException e) {
// 若释放锁时发生异常，记录警告信息
LOG.warn("Exception when releasing locks in ShutdownHook for Driver: " + e.getMessage());
}
}
};
// 添加新的关闭hook
ShutdownHookManager.addShutdownHook(shutdownRunner, SHUTDOWN_HOOK_PRIORITY);
// 再次检查编译过程是否被中断
if(isInterrupted()) {
return handleInterruption("before parsing and analysing the query");
}
// 如果上下文为空，则创建新的上下文
if(ctx == null) {
ctx = new Context(conf);
}
// 设置上下文的重试次数、命令和HDFS清理标志
ctx.setTryCount(getTryCount());
ctx.setCmd(command);
ctx.setHDFSCleanup(true);
/**
* 把 HQL命令 翻译成一个 ASTNode Tree
* 封装了 ParseDriver 对 HQL 的解析工作
* ParseDriver 对 command 进行词法分析和语法解析（统称为语法分析），返回一个抽象语法树AST
*/
// 开始记录解析过程的性能
perfLogger.PerfLogBegin(CLASS_NAME, PerfLogger.PARSE);
// 解析查询命令，得到抽象语法树
ASTNode tree = ParseUtils.parse(command, ctx);
// 结束记录解析过程的性能
perfLogger.PerfLogEnd(CLASS_NAME, PerfLogger.PARSE);
// 加载查询hook
queryHooks = loadQueryHooks();
// 如果查询hook不为空且不为空列表，则触发查询hook的编译前操作
if(queryHooks != null && !queryHooks.isEmpty()) {
QueryLifeTimeHookContext qhc = new QueryLifeTimeHookContextImpl();
qhc.setHiveConf(conf);
qhc.setCommand(command);
for(QueryLifeTimeHook hook : queryHooks) {
hook.beforeCompile(qhc);
}
}
// 开始记录语义分析过程的性能
perfLogger.PerfLogBegin(CLASS_NAME, PerfLogger.ANALYZE);
// 获取语义分析器
BaseSemanticAnalyzer sem = SemanticAnalyzerFactory.get(queryState, tree);
// 获取语义分析hook
List<HiveSemanticAnalyzerHook> saHooks = getHooks(HiveConf.ConfVars.SEMANTIC_ANALYZER_HOOK, HiveSemanticAnalyzerHook.class);
// 刷新元数据存储缓存，确保获取最新的元数据
Hive.get().getMSC().flushCache();
// 进行语义分析和计划生成
if(saHooks != null && !saHooks.isEmpty()) {
HiveSemanticAnalyzerHookContext hookCtx = new HiveSemanticAnalyzerHookContextImpl();
hookCtx.setConf(conf);
hookCtx.setUserName(userName);
hookCtx.setIpAddress(SessionState.get().getUserIpAddress());
hookCtx.setCommand(command);
hookCtx.setHiveOperation(queryState.getHiveOperation());
// 触发语义分析hook的预分析操作
for(HiveSemanticAnalyzerHook hook : saHooks) {
tree = hook.preAnalyze(hookCtx, tree);
}
/**
* sem 是一个 SemanticAnalyzer(语义分析器) 对象
* 主要的工作是将 ASTNode 转化为 TaskTree，包括可能的 optimize，过程比较复杂
*
* tree: AST 抽象语法树 ===> TaskTree
* TaskTree ： 物理执行计划
*
* 把抽象语法树交给 SemanticAnalyzer 执行语法解析
* 1、从 AST 转成 解析树
* 2、通过解析树 再生成 QB 在查询快
* 3、从 QB 树在生成 OperatorTree (Logical Plan)
* 4、逻辑执行计划的优化
* 5、OperatorTree转变成TaskTree
* 6、再针对物理执行计划执行优化
* 7、生成QueryPlan
*/
// 进行语义分析
sem.analyze(tree, ctx);
// 更新hook上下文
hookCtx.update(sem);
// 触发语义分析hook的后分析操作
for(HiveSemanticAnalyzerHook hook : saHooks) {
hook.postAnalyze(hookCtx, sem.getAllRootTasks());
}
} else {
// 若没有语义分析hook，直接进行语义分析
sem.analyze(tree, ctx);
}
// 记录查询中发现的ACID文件接收器
acidSinks = sem.getAcidFileSinks();
// 记录语义分析完成信息
LOG.info("Semantic Analysis Completed");
// 验证语义分析生成的计划是否有效
sem.validate();
// 检查查询中是否包含ACID操作
acidInQuery = sem.hasAcidInQuery();
// 结束语义分析阶段的性能日志记录
perfLogger.PerfLogEnd(CLASS_NAME, PerfLogger.ANALYZE);
// 检查编译过程是否被中断，如果中断则处理中断情况并返回
if(isInterrupted()) {
return handleInterruption("after analyzing query.");
}
// 根据语义分析结果和配置信息获取查询的输出模式
schema = getSchema(sem, conf);
/**
* 把 TaskTree 生成一个 QueryPlan
* 通过 Exeuctor 提交的方法，要接受的参数就是 QueryPlan
*/
// 根据查询字符串、语义分析器、开始时间、查询ID、操作类型和输出模式创建查询计划
plan = new QueryPlan(queryStr, sem, perfLogger.getStartTime(PerfLogger.DRIVER_RUN), queryId, queryState.getHiveOperation(), schema);
// 设置查询字符串到配置中
conf.setQueryString(queryStr);
// 设置MapReduce工作流ID到配置中
conf.set("mapreduce.workflow.id", "hive_" + queryId);
// 设置MapReduce工作流名称到配置中
conf.set("mapreduce.workflow.name", queryStr);
// 如果查询计划中包含FetchTask，则对其进行初始化
if(plan.getFetchTask() != null) {
plan.getFetchTask().initialize(queryState, plan, null, ctx.getOpContext());
}
// 进行授权检查，如果语义分析不跳过授权且开启了授权功能
if(!sem.skipAuthorization() && HiveConf.getBoolVar(conf, HiveConf.ConfVars.HIVE_AUTHORIZATION_ENABLED)) {
try {
// 开始记录授权过程的性能日志
perfLogger.PerfLogBegin(CLASS_NAME, PerfLogger.DO_AUTHORIZATION);
// 执行授权操作
doAuthorization(queryState.getHiveOperation(), sem, command);
} catch(AuthorizationException authExp) {
// 如果授权失败，打印错误信息并设置错误状态和返回码
console.printError("Authorization failed:" + authExp.getMessage() + ". Use SHOW GRANT to " + "get" + " more details.");
errorMessage = authExp.getMessage();
SQLState = "42000";
return 403;
} finally {
// 结束记录授权过程的性能日志
perfLogger.PerfLogEnd(CLASS_NAME, PerfLogger.DO_AUTHORIZATION);
}
}
// 如果配置中开启了记录EXPLAIN输出的功能
if(conf.getBoolVar(ConfVars.HIVE_LOG_EXPLAIN_OUTPUT)) {
// 获取查询的EXPLAIN输出
String explainOutput = getExplainOutput(sem, plan, tree);
if(explainOutput != null) {
if(conf.getBoolVar(ConfVars.HIVE_LOG_EXPLAIN_OUTPUT)) {
// 记录EXPLAIN输出到日志中
LOG.info("EXPLAIN output for queryid " + queryId + " : " + explainOutput);
}
if(conf.isWebUiQueryInfoCacheEnabled()) {
// 如果开启了Web UI查询信息缓存，将EXPLAIN计划设置到查询显示信息中
queryDisplay.setExplainPlan(explainOutput);
}
}
}
// 编译成功，返回0
return 0;
} catch(Exception e) {
// 如果编译过程中被中断，处理中断情况并返回
if(isInterrupted()) {
return handleInterruption("during query compilation: " + e.getMessage());
}
// 标记编译过程出现错误
compileError = true;
// 获取错误信息
ErrorMsg error = ErrorMsg.getErrorMsg(e.getMessage());
// 构建错误消息
errorMessage = "FAILED: " + e.getClass().getSimpleName();
if(error != ErrorMsg.GENERIC_ERROR) {
errorMessage += " [Error " + error.getErrorCode() + "]:";
}
// HIVE-4889
if((e instanceof IllegalArgumentException) && e.getMessage() == null && e.getCause() != null) {
errorMessage += " " + e.getCause().getMessage();
} else {
errorMessage += " " + e.getMessage();
}
if(error == ErrorMsg.TXNMGR_NOT_ACID) {
errorMessage += ". Failed command: " + queryStr;
}
// 设置SQL状态码
SQLState = error.getSQLState();
// 记录下游错误信息
downstreamError = e;
// 打印错误信息和详细堆栈跟踪
console.printError(errorMessage, "\n" + org.apache.hadoop.util.StringUtils.stringifyException(e));
// 返回错误代码
return error.getErrorCode();
// since it exceeds valid range of shell return values
} finally {
// 触发编译后的hook函数
try {
if(queryHooks != null && !queryHooks.isEmpty()) {
QueryLifeTimeHookContext qhc = new QueryLifeTimeHookContextImpl();
qhc.setHiveConf(conf);
qhc.setCommand(command);
for(QueryLifeTimeHook hook : queryHooks) {
hook.afterCompile(qhc, compileError);
}
}
} catch(Exception e) {
// 如果触发hook函数时出现异常，记录警告信息
LOG.warn("Failed when invoking query after-compilation hook.", e);
}
/**
* 计算任务总耗时
*/
// 结束编译阶段的性能日志记录并计算耗时
double duration = perfLogger.PerfLogEnd(CLASS_NAME, PerfLogger.COMPILE) / 1000.00;
// 获取编译过程中HMS调用的时间统计信息
ImmutableMap<String, Long> compileHMSTimings = dumpMetaCallTimingWithoutEx("compilation");
// 设置查询显示信息中的HMS时间统计信息
queryDisplay.setHmsTimings(QueryDisplay.Phase.COMPILATION, compileHMSTimings);
// 检查编译过程是否被中断
boolean isInterrupted = isInterrupted();
if(isInterrupted && !deferClose) {
// 如果被中断且不延迟关闭，关闭正在进行的操作
closeInProcess(true);
}
// 锁定驱动状态
lDrvState.stateLock.lock();
try {
if(isInterrupted) {
// 如果被中断，根据是否延迟关闭设置驱动状态
lDrvState.driverState = deferClose ? DriverState.EXECUTING : DriverState.ERROR;
} else {
// 如果未被中断，根据编译是否出错设置驱动状态
lDrvState.driverState = compileError ? DriverState.ERROR : DriverState.COMPILED;
}
} finally {
// 解锁驱动状态
lDrvState.stateLock.unlock();
}
if(isInterrupted) {
// 如果编译过程被中断，记录中断信息
LOG.info("Compiling command(queryId=" + queryId + ") has been interrupted after " + duration + " seconds");
} else {
// 如果编译过程未被中断，记录编译完成信息
LOG.info("Completed compiling command(queryId=" + queryId + "); Time taken: " + duration + " " + "seconds");
}
}
}
compile()方法在3.x和4.x有一些区别，但是都有以下几个核心方法：
- 2.x和3.x是通过ParseUtils.parse(command, ctx)，而4.x是通过parse()将Hive SQL转换成AST（抽象语法树），即：HQL -> AST 转换；
- 然后无论哪个版本，都会通过BaseSemanticAnalyzer.analyze()方法将AST解析生成TaskTree(物理执行计划)；
- 最后2.x和3.x版本都会将BaseSemanticAnalyzer传入QueryPlan构造函数来创建QueryPlan(查询计划)，而4.x版本则是传入createPlan()方法创建QueryPlan。
总结
本文介绍了Hive，并通过源码梳理了Hive的执行原理，其核心正是引入篇我们提到的解析（Parsing）、校验（Validation）、优化（Optimization）和执行（Execution）。
总结起来主要有以下四个步骤：
- 词法解析
将SQL语法转换成AST(抽象语法树)
核心是parse()方法中调用的HiveLexer和HiveParser这两个类，它们分别负责SQL的词法分析和语法解析。 - 语义分析
这一步是对AST进行进一步的抽象和结构化处理，通过遍历AST(抽象语法树) 将其转化成Query Block（查询块，可以看成查询基本执行单元，它包含了输入源、计算过程和输出结果这三个基本组成部分。）
核心是通过BaseSemanticAnalyzer子类SemanticAnalyzer的analyzeInternal()方法，核心逻辑是首先将SQL语句中涉及的各类信息提取出来，并存储到QueryBlock中，在完成后，通过genOPTree()方法将Query Block(查询块) 转换成OperatorTree（逻辑执行计划） - 逻辑优化
到了第三步时，操作符树虽然已经勾勒出执行任务的先后顺序和上下游依赖，但细节还比较粗糙，例如存在重复的数据扫描、不必要的Shuffle操作等，因此还需要进行进一步优化。通过优化，Hive可以改进查询的执行计划，并生成更高效的作业图以在分布式计算框架中执行。这些优化可以提高查询的性能和效率，并减少资源开销。
核心是通过Optimizer类的方法完成的，从源码可以看到，优化器的种类非常繁杂。总体而言，优化的目的是通过匹配相应的规则来减少MapReduce作业的数量，降低数据传输和Shuffle的数据量。 - 物理优化
在逻辑优化阶段结束后，输入的SQL语句也逐步转换为优化后的逻辑计划，不过此时的逻辑计划仍然不能直接执行，还需要进一步转换成可以识别并执行的MapReduce Task，因此物理优化实际上分为两个执行步骤：首先将优化后的OperatorTree(逻辑执行计划)转换成TaskTree(物理执行计划，每个Task对应一个MR Job任务)，并对物理执行计划进行一些优化，然后依次调用执行。
因为篇幅问题，本文有很多有意思的方法源码没有深入探索，比如compile()方法中的核心方法；比如Hive通过Antler实现的词法分析和语法解析，是我们去梳理hive任务血缘的核心技术等，都是很有意思的东西，感兴趣的小伙伴可以深入了解一下。