Hive源码分析：Driver类运行过程

说明：

本文的源码分析基于hive-0.12.0-cdh5.0.1。

概括

从《hive cli的入口类》中可以知道hive中处理hive命令的处理器一共有以下几种：

（1）set       SetProcessor，设置修改参数,设置到SessionState的HiveConf里。
（2）dfs       DfsProcessor，使用hadoop的FsShell运行hadoop的命令。
（3）add       AddResourceProcessor，添加到SessionState的resource_map里，运行提交job的时候会写入Hadoop的Distributed Cache。
（4）delete    DeleteResourceProcessor，从SessionState的resource_map里删除。
（5）reset     RestResourceProcessor，重置终端输出
（6）其他命令   Driver

Driver类的主要作用是用来编译并执行hive命令，然后返回执行结果。这里主要分析Driver类的运行逻辑，其时序图如下：

从时序图上可以看出有以下步骤：

  • run方法调用内部方法runInternal
  • 在runInternal方法内部先，调用HiveDriverRunHookContext的preDriverRun方法
  • 调用compileInternal方法
  • compileInternal方法内部调用compile方法
  • compile方法内，先调用HiveSemanticAnalyzerHookContext的preAnalyze方法
  • 再进行语法分析，调用BaseSemanticAnalyzer的analyze方法
  • 调用HiveSemanticAnalyzerHookContext的postAnalyze方法
  • 再进行语法校验，调用BaseSemanticAnalyzer的validate方法
  • compileInternal方法运行完成之后，调用checkConcurrency方法
  • 再来运行execute方法，该方法用于运行任务
  • 最后，调用HiveDriverRunHookContext的postDriverRun方法

Driver初始化

在继续分析之前，需要弄清楚Driver类初始化时做了什么事情。

在CliDriver的processCmd(String cmd)方法中可以看到proc是在CommandProcessorFactory类中new出来的并调用了init方法。

} else { // local mode
      CommandProcessor proc = CommandProcessorFactory.get(tokens[0], (HiveConf) conf);
      ret = processLocalCmd(cmd, proc, ss);
}

CommandProcessorFactory.get方法代码片段：

if (conf == null) {
return new Driver();
}

Driver drv = mapDrivers.get(conf);
if (drv == null) {
drv = new Driver();
mapDrivers.put(conf, drv);
}
drv.init();

init方法和构造方法代码如下：

public void init() {
	Operator.resetId();
}

public Driver() {
    if (SessionState.get() != null) {
      conf = SessionState.get().getConf();
    }
  }

从上可以看出仅仅是初始化了conf属性和重置了Operator的id。

run方法过程

1、调用runInternal方法，根据该方法返回值判断是否出错。

2、runInternal方法内，运行HiveDriverRunHook的前置方法preDriverRun

3、判断是否需要编译，如果需要，则运行compileInternal(command)方法，并根据返回值判断是否该释放Hive锁。hive中可以配置hive.support.concurrency值为true并设置zookeeper的服务器地址和端口，基于zookeeper实现分布式锁以支持hive的多并发访问。这部分内容不是本文重点故不做介绍。

compileInternal(command)方法内部代码说明见下文。

4、判断是否需要对Task加锁。如果需要，则调用checkConcurrency方法。

5、调用execute()方法执行任务。

  • 执行计划开始：plan.setStarted();
  • 先运行ExecuteWithHookContext的前置hook方法，ExecuteWithHookContext类型有三种：前置、运行失败、后置。
  • 然后创建DriverContext用于维护正在运行的task任务，正在运行的task任务会添加到队列runnable中去。
  • 其次，在while循环中遍历队列中的任务，然后启动任务让其执行，并且轮训任务执行结果，如果任务运行完成，则将其从running中删除并将当前任务的子任务加入队列中；如果运行失败，则会启动备份的任务，并运行失败的hook。

while (runnable.peek() != null && running.size() < maxthreads) {
  Task<? extends Serializable> tsk = runnable.remove();
  launchTask(tsk, queryId, noName, running, jobname, jobs, driverCxt);
}
  • 在launchTask方法中，先判断是否支持并发执行，如果支持则调用TaskRunner的start()方法，否则调用tskRun.runSequential()方法顺序执行，只有当是MapReduce任务时，才执行并发执行：

	public void launchTask(Task<? extends Serializable> tsk, String queryId, boolean noName,
      Map<TaskResult, TaskRunner> running, String jobname, int jobs, DriverContext cxt) {

    if (SessionState.get() != null) {
      SessionState.get().getHiveHistory().startTask(queryId, tsk, tsk.getClass().getName());
    }
    if (tsk.isMapRedTask() && !(tsk instanceof ConditionalTask)) {
      if (noName) {
        conf.setVar(HiveConf.ConfVars.HADOOPJOBNAME, jobname + "(" + tsk.getId() + ")");
      }
      cxt.incCurJobNo(1);
      console.printInfo("Launching Job " + cxt.getCurJobNo() + " out of " + jobs);
    }
    tsk.initialize(conf, plan, cxt);
    TaskResult tskRes = new TaskResult();
    TaskRunner tskRun = new TaskRunner(tsk, tskRes);

    // Launch Task
    if (HiveConf.getBoolVar(conf, HiveConf.ConfVars.EXECPARALLEL) && tsk.isMapRedTask()) {
      // Launch it in the parallel mode, as a separate thread only for MR tasks
      tskRun.start();
    } else {
      tskRun.runSequential();
    }
    running.put(tskRes, tskRun);
    return;
  }

最后任务的执行情况，就要看具体的Task<? extends Serializable>的实现类的逻辑了。

  • 执行计划完成：plan.setDone();

6、运行HiveDriverRunHook的后置方法postDriverRun

compileInternal方法过程

1、保存当前查询状态

QueryState queryState = new QueryState();

if (plan != null) {
  close();
  plan = null;
}

if (resetTaskIds) {
  TaskFactory.resetId();
}
saveSession(queryState);

QueryState中保存了HiveOperation以及当前查询语句或者命令。

2、创建Context上下文

command = new VariableSubstitution().substitute(conf,command);
ctx = new Context(conf);
ctx.setTryCount(getTryCount());
ctx.setCmd(command);
ctx.setHDFSCleanup(true);

3、创建ParseDriver对象，然后解析命令、生成AST树。语法和词法分析内容，不是本文重点故不做介绍。

ParseDriver pd = new ParseDriver();
ASTNode tree = pd.parse(command, ctx);
tree = ParseUtils.findRootNonNullToken(tree);

简单归纳来说，解析程包括如下：

  • 词法分析，生成AST树，ParseDriver完成。
  • 分析AST树，AST拆分成查询子块，信息记录在QB，这个QB在下面几个阶段都需要用到，SemanticAnalyzer.doPhase1完成。
  • 从metastore中获取表的信息，SemanticAnalyzer.getMetaData完成。
  • 生成逻辑执行计划，SemanticAnalyzer.genPlan完成。
  • 优化逻辑执行计划，Optimizer完成，ParseContext作为上下文信息进行传递。
  • 生成物理执行计划，SemanticAnalyzer.genMapRedTasks完成。
  • 物理计划优化，PhysicalOptimizer完成，PhysicalContext作为上下文信息进行传递。

4、读取环境变量，如果配置了语法分析的hook，参数为：hive.semantic.analyzer.hook，则:先用反射得到AbstractSemanticAnalyzerHook的集合，调用hook.preAnalyze(hookCtx, tree)方法,然后再调用sem.analyze(tree, ctx)方法，该方法才是用来作语法分析的,最后再调用hook.postAnalyze(hookCtx, tree)方法执行一些用户定义的后置操作；

否则，直接调用sem.analyze(tree, ctx)进行语法分析。

  BaseSemanticAnalyzer sem = SemanticAnalyzerFactory.get(conf, tree);
  List<AbstractSemanticAnalyzerHook> saHooks =
      getHooks(HiveConf.ConfVars.SEMANTIC_ANALYZER_HOOK,
               AbstractSemanticAnalyzerHook.class);

  // Do semantic analysis and plan generation
  if (saHooks != null) {
    HiveSemanticAnalyzerHookContext hookCtx = new HiveSemanticAnalyzerHookContextImpl();
    hookCtx.setConf(conf);
    hookCtx.setUserName(userName);
    for (AbstractSemanticAnalyzerHook hook : saHooks) {
      tree = hook.preAnalyze(hookCtx, tree);
    }
    sem.analyze(tree, ctx);
    hookCtx.update(sem);
    for (AbstractSemanticAnalyzerHook hook : saHooks) {
      hook.postAnalyze(hookCtx, sem.getRootTasks());
    }
  } else {
    sem.analyze(tree, ctx);
  }

5、校验执行计划：sem.validate()

6、创建查询计划QueryPlan。

plan = new QueryPlan(command, sem, perfLogger.getStartTime(PerfLogger.DRIVER_RUN),
           SessionState.get().getCommandType());

7、初始化FetchTask。

if (plan.getFetchTask() != null) {
   plan.getFetchTask().initialize(conf, plan, null);
}

8、得到schema

schema = getSchema(sem, conf);

9、授权校验工作。

hive中支持的hook

上面分析中，提到了hive的hook机制，hive中一共存在以下几种hook。

hive.semantic.analyzer.hook
hive.exec.filter.hook
hive.exec.driver.run.hooks
hive.server2.session.hook
hive.exec.pre.hooks
hive.exec.post.hooks
hive.exec.failure.hooks
hive.client.stats.publishers
hive.metastore.ds.connection.url.hook
hive.metastore.init.hooks

通过hook机制，可以在运行前后做一些用户想做的事情。如：你可以在语法分析的hook中对hive的操作做一些超级管理员级别的权限判断；你可以对hive-server2做一些session级别的控制。

cloudera的github仓库access中关于hive的访问控制就是使用了hive的hook机制。

twitter的mapreduce可视化项目监控项目ambrose也利用了hive的hook机制，有兴趣的话，你可以去看看其是如何使用hive的hook并且你也可以扩增hook做些自己想做的事情。

总结

本文主要介绍了hive运行过程，包括hive语法词法解析以及hook机制，任务的最后运行过程取决于具体的Task<? extends Serializable>的实现类的逻辑。关于hive语法词法解析，这一部分没有做详细的解释。

hive Driver类的执行过程如下（该图是根据hive-0.11版本画出来的）：

时间： 2024-04-24 22:43:26

Hive源码分析：Driver类运行过程的相关文章

Hive源码分析：CLI入口类

说明: 本文的源码分析基于hive-0.10.0-cdh4.3.0. 启动脚本 从shell脚本/usr/lib/hive/bin/ext/cli.sh可以看到hive cli的入口类为org.apache.hadoop.hive.cli.CliDriver cli () { CLASS=org.apache.hadoop.hive.cli.CliDriver execHiveCmd $CLASS "$@" } cli_help () { CLASS=org.apache.hadoop

Spark 源码分析 -- task实际执行过程

Spark源码分析 – SparkContext 中的例子, 只分析到sc.runJob 那么最终是怎么执行的? 通过DAGScheduler切分成Stage, 封装成taskset, 提交给TaskScheduler, 然后等待调度, 最终到Executor上执行 val sc = new SparkContext(--) val textFile = sc.textFile("README.md") textFile.filter(line => line.contains(

Openstack Nova 源码分析 — RPC 远程调用过程

目录 目录 Nova Project Services Project 的程序入口 setuppy Nova中RPC远程过程调用 nova-compute RPC API的实现 novacomputemanager 模块 最后 Nova Project Services nova-api:捕获novaclient发送过来的HTTP请求,并且将它转换为AMQP消息,通过Queue来与别的services通信. nova-conductor:为数据库访问提供了一层安全保障. NOTE:除了nova-

HBase源码分析：HTable put过程

HBase版本:0.94.15-cdh4.7.0 在 HBase中,大部分的操作都是在RegionServer完成的,Client端想要插入.删除.查询数据都需要先找到相应的 RegionServer.什么叫相应的RegionServer?就是管理你要操作的那个Region的RegionServer.Client本身并 不知道哪个RegionServer管理哪个Region,那么它是如何找到相应的RegionServer的?本文就是在研究源码的基础上了解这个过程. 首先来看看写过程的序列图: 客

Jquery源码分析---jQuery类数组的分析

4.1.类数组构建 从上节可以看出jquery构建函数完成了查找或转换 或其它的功能,其结果就是查找到元素.Dom树查找,html转换成Dom元素,还是 直接传入Dom元素都不过是方式而已.找到这些元素就得找个地方去存储起来. 存储有序数据的地方(集合)在JS中最好的当然是数组.那么又如何在 jQuery内面实现数组呢?可以采用如下的方式: jQuery.fn.prototype=new Array(); 在上一节中 的this.setArray(arr)函数中加上 Array.apply (t

spark源码分析之Checkpoint的过程

概述 checkpoint 的机制保证了需要访问重复数据的应用 Spark 的DAG执行行图可能很庞大,task 中计算链可能会很长,这时如果 task 中途运行出错,那么 task 的整个需要重算非常耗时,因此,有必要将计算代价较大的 RDD checkpoint 一下,当下游 RDD 计算出错时,可以直接从 checkpoint 过的 RDD 那里读取数据继续算. 我们先来看一个例子,checkpoint的使用 import org.apache.spark.SparkContext imp

Appium Server源码分析之作为Bootstrap客户端

Appium Server拥有两个主要的功能: 它是个http服务器,它专门接收从客户端通过基于http的REST协议发送过来的命令 他是bootstrap客户端:它接收到客户端的命令后,需要想办法把这些命令发送给目标安卓机器的bootstrap来驱动uiatuomator来做事情 通过上一篇文章<Appium Server 源码分析之启动运行Express http服务器>我们分析了Appium Server是如何作为一个http服务器进行工作的.那么今天我们就要分析第二点,Appium S

Spring AOP源码分析（七）ProxyFactoryBean介绍

这篇文章里面就要说说Spring自己的AOP,搞清楚哪种方式是Spring自己实现的AOP,哪种方式是Spring引入aspectj的AOP.  Spring自己的AOP实现在于ProxyFactoryBean.先看下使用案例(仍和之前的案例是一样的):接口AService.实现类AServiceImpl.通知MyBeforeAdvice  ? 1 2 3 4 5 6 public interface AService {       public void fooA(String _msg);

Android源码分析-Alarm机制与Binder的交互

转载请注明出处:http://blog.csdn.net/singwhatiwanna/article/details/18448997 前言 本次给大家分析的是Android中Alarm的机制以及它和Binder的交互,所用源码为最新的Android4.4.因为Alarm的功能都是通过Binder来完成的,所以,介绍Alarm之前必须要先介绍下它是如何调用Binder来完成定时功能的.由于内容较多,本文会比较长,在文章结构安排上是这样的:首先简单介绍如何使用Alarm并给出其工作原理,接着分析