hive执行流程(3)-Driver类分析1Driver类整体流程-低调大师

hive执行流程(3)-Driver类分析1Driver类整体流程

2017-11-14 623

Driver类是对

 
        org.apache.hadoop.hive.ql.processors.CommandProcessor.java

接口的实现，重写了run方法，定义了常见sql的执行方式.

 
        public 
        class 
        Driver 
        implements 
        CommandProcessor

具体的方法调用顺序：

 
        run--->runInternal--->(createTxnManager+recordValidTxns)----->compileInternal--->
       
        compile--analyzer(BaseSemanticAnalyzer)--->execute

其中compile和execute是两个比较重要的方法：

compile用来完成语法和语义的分析，生成执行计划

execute执行物理计划，即提交相应的mapredjob

通过打印perflog可以看到Driver类的简单地时序图：

下面来看下Driver类的几个常用的方法实现：

1)createTxnManager 用来获取目前设置的用于实现lock的类，比如:

 
        org.apache.hadoop.hive.ql.lockmgr.DummyTxnManager

2)checkConcurrency 用来判断当前hive设置是否支持并发控制:

 
        boolean 
        supportConcurrency = conf.getBoolVar(HiveConf.ConfVars.HIVE_SUPPORT_CONCURRENCY);

主要是通过判断hive.support.concurrency参数，默认是false

3)getClusterStatus 调用JobClient类的getClusterStatus方法来获取集群的状态:

 
        public 
        ClusterStatus getClusterStatus() 
        throws 
        Exception { 
       
        ClusterStatus cs; 
       
        try 
        { 
       
        JobConf job = 
        new 
        JobConf(conf , ExecDriver.
        class
        ); 
       
        JobClient jc = 
        new 
        JobClient(job); 
       
        cs = jc.getClusterStatus(); 
       
        } 
        catch 
        (Exception e) { 
       
        e.printStackTrace(); 
       
        throw 
        e; 
       
        } 
       
        LOG.info( 
        "Returning cluster status: " 
        + cs.toString()); 
       
        return 
        cs; 
       
        }

4）getSchema //返回表的schema信息

5）

 
        doAuthorization/doAuthorizationV2/getHivePrivObjects

用来在开启权限验证情况下对sql的权限检测操作

6）

 
        getLockObjects/acquireReadWriteLocks/releaseLocks

都是和锁相关的方法 ,其中getLockObjects用来获取锁的对象（锁的路径，锁的模式等），最终返回一个包含所有锁的list,acquireReadWriteLocks用来控制获取锁，releaseLocks用来释放锁:

 
        getLockObjects:
       
        private 
        List<HiveLockObj> getLockObjects(Database d, Table t, Partition p, HiveLockMode mode) 
       
        throws 
        SemanticException { 
       
        List<HiveLockObj> locks = 
        new 
        LinkedList<HiveLockObj>(); 
       
        HiveLockObjectData lockData = 
       
        new 
        HiveLockObjectData( plan.getQueryId(), 
       
        String. valueOf(System.currentTimeMillis ()), 
       
        "IMPLICIT"
        , 
       
        plan.getQueryStr()); 
       
        if 
        (d != 
        null
        ) { 
       
        locks.add( 
        new 
        HiveLockObj(
        new 
        HiveLockObject(d.getName(), lockData), mode));  
        //数据库层面的锁 
       
        return 
        locks; 
       
        } 
       
        if 
        (t != 
        null
        ) {  
        // 表层面的锁 
       
        locks.add( 
        new 
        HiveLockObj(
        new 
        HiveLockObject(t.getDbName(), lockData), mode)); 
       
        locks.add( 
        new 
        HiveLockObj(
        new 
        HiveLockObject(t, lockData), mode)); 
       
        mode = HiveLockMode.SHARED; 
       
        locks.add( 
        new 
        HiveLockObj(
        new 
        HiveLockObject(t.getDbName(), lockData), mode)); 
       
        return 
        locks; 
       
        } 
       
        if 
        (p != 
        null
        ) { 
        //分区层面的锁 
       
        locks.add( 
        new 
        HiveLockObj(
        new 
        HiveLockObject(p.getTable().getDbName(), lockData), mode)); 
       
        if 
        (!(p 
        instanceof 
        DummyPartition)) { 
       
        locks.add( 
        new 
        HiveLockObj(
        new 
        HiveLockObject(p, lockData), mode)); 
       
        } 
       
        // All the parents are locked in shared mode 
       
        mode = HiveLockMode.SHARED; 
       
        // For dummy partitions, only partition name is needed 
       
        String name = p.getName(); 
       
        if 
        (p 
        instanceof 
        DummyPartition) { 
       
        name = p.getName().split( 
        "@"
        )[
        2
        ]; 
       
        } 
       
        String partialName = 
        ""
        ; 
       
        String[] partns = name.split( 
        "/"
        ); 
       
        int 
        len = p 
        instanceof 
        DummyPartition ? partns.length : partns.length - 
        1
        ; 
       
        Map<String, String> partialSpec = 
        new 
        LinkedHashMap<String, String>(); 
       
        for 
        ( 
        int 
        idx = 
        0
        ; idx < len; idx++) { 
       
        String partn = partns[idx]; 
       
        partialName += partn; 
       
        String[] nameValue = partn.split( 
        "="
        ); 
       
        assert
        (nameValue.length == 
        2
        ); 
       
        partialSpec.put(nameValue[
        0
        ], nameValue[
        1
        ]); 
       
        try 
        { 
       
        locks.add( 
        new 
        HiveLockObj( 
       
        new 
        HiveLockObject(
        new 
        DummyPartition(p.getTable(), p.getTable().getDbName() 
       
        + 
        "/" 
        + p.getTable().getTableName() 
       
        + 
        "/" 
        + partialName, 
       
        partialSpec), lockData), mode)); 
       
        partialName += 
        "/"
        ; 
       
        } 
        catch 
        (HiveException e) { 
       
        throw 
        new 
        SemanticException(e.getMessage()); 
       
        } 
       
        } 
       
        locks.add( 
        new 
        HiveLockObj(
        new 
        HiveLockObject(p.getTable(), lockData), mode)); 
       
        locks.add( 
        new 
        HiveLockObj(
        new 
        HiveLockObject(p.getTable().getDbName(), lockData), mode)); 
       
        } 
       
        return 
        locks; 
       
        }

acquireReadWriteLocks调用了锁具体实现类的acquireLocks方法

releaseLocks调用了锁具体实现类的releaseLocks方法

7）

run方法是Driver类的入口方法，调用了runInternal方法，我们主要来看runInternal的方法，大体步骤：

 
        运行hive.exec.driver.run.hooks中设置的hook，
       
        运行HiveDriverRunHook相关类的的preDriverRun方法---->检测是否支持并发，并获取并发实现的类
       
        --->compileInternal---->运行锁相关的操作（判断是否只对mapred job进行锁，获取锁等）
       
        ---->调用execute---->释放锁--->运行HiveDriverRunHook相关类的的postDriverRun方法
       
        ---->返回CommandProcessorResponse对象

相关代码：

 
        private 
        CommandProcessorResponse runInternal(String command, 
        boolean 
        alreadyCompiled) 
       
        throws 
        CommandNeedRetryException { 
       
        errorMessage = 
        null
        ; 
       
        SQLState = 
        null
        ; 
       
        downstreamError = 
        null
        ; 
       
        if 
        (!validateConfVariables()) { 
       
        return 
        new 
        CommandProcessorResponse(
        12
        , errorMessage , SQLState ); 
       
        } 
       
        HiveDriverRunHookContext hookContext = 
        new 
        HiveDriverRunHookContextImpl(conf , command); 
       
        // Get all the driver run hooks and pre-execute them. 
       
        List<HiveDriverRunHook> driverRunHooks; 
       
        try 
        {               
        //运行hive.exec.driver.run.hooks中设置的hook 
       
        driverRunHooks = getHooks(HiveConf.ConfVars.HIVE_DRIVER_RUN_HOOKS,  
       
        HiveDriverRunHook. 
        class
        );          
       
        for 
        (HiveDriverRunHook driverRunHook : driverRunHooks) { 
       
        driverRunHook.preDriverRun(hookContext); 
        //运行HiveDriverRunHook相关类的的preDriverRun方法 
       
        } 
       
        } 
        catch 
        (Exception e) { 
       
        errorMessage = 
        "FAILED: Hive Internal Error: " 
        + Utilities.getNameMessage(e); 
       
        SQLState = ErrorMsg. findSQLState(e.getMessage()); 
       
        downstreamError = e; 
       
        console.printError( errorMessage + 
        "\n" 
       
        + org.apache.hadoop.util.StringUtils.stringifyException(e)); 
       
        return 
        new 
        CommandProcessorResponse(
        12
        , errorMessage , SQLState ); 
       
        } 
       
        // Reset the perf logger 
       
        PerfLogger perfLogger = PerfLogger.getPerfLogger( 
        true
        ); 
       
        perfLogger.PerfLogBegin( CLASS_NAME, PerfLogger.DRIVER_RUN); 
       
        perfLogger.PerfLogBegin( CLASS_NAME, PerfLogger.TIME_TO_SUBMIT); 
       
        int 
        ret; 
       
        boolean 
        requireLock = 
        false
        ; 
       
        boolean 
        ckLock = 
        false
        ; 
       
        try 
        { 
       
        ckLock = checkConcurrency();  
        //检测是否支持并发，并获取并发实现的类，比如常用的 org.apache.hadoop.hive.ql.lockmgr.DummyTxnManager 
       
        createTxnManager(); 
       
        } 
        catch 
        (SemanticException e) { 
       
        errorMessage = 
        "FAILED: Error in semantic analysis: " 
        + e.getMessage(); 
       
        SQLState = ErrorMsg. findSQLState(e.getMessage()); 
       
        downstreamError = e; 
       
        console.printError( errorMessage, 
        "\n" 
       
        + org.apache.hadoop.util.StringUtils.stringifyException(e)); 
       
        ret = 
        10
        ; 
       
        return 
        new 
        CommandProcessorResponse(ret, errorMessage , SQLState ); 
       
        } 
       
        ret = recordValidTxns(); 
       
        if 
        (ret != 
        0
        ) 
        return 
        new 
        CommandProcessorResponse(ret, errorMessage, SQLState); 
       
        if 
        (!alreadyCompiled) { 
       
        ret = compileInternal(command);  
        //调用compileInternal方法 
       
        if 
        (ret != 
        0
        ) { 
       
        return 
        new 
        CommandProcessorResponse(ret, errorMessage, SQLState); 
       
        } 
       
        } 
       
        // the reason that we set the txn manager for the cxt here is because each 
       
        // query has its own ctx object. The txn mgr is shared across the 
       
        // same instance of Driver, which can run multiple queries. 
       
        ctx.setHiveTxnManager( txnMgr); 
       
        if 
        (ckLock) {  
        //断是否只对mapred job进行锁,参数hive.lock.mapred.only.operation，默认为false 
       
        boolean 
        lockOnlyMapred = HiveConf.getBoolVar( conf, HiveConf.ConfVars.HIVE_LOCK_MAPRED_ONLY); 
       
        if
        (lockOnlyMapred) { 
       
        Queue<Task<? 
        extends 
        Serializable>> taskQueue = 
        new 
        LinkedList<Task<? 
        extends 
        Serializable>>(); 
       
        taskQueue.addAll( plan.getRootTasks()); 
       
        while 
        (taskQueue.peek() != 
        null
        ) { 
       
        Task<? 
        extends 
        Serializable> tsk = taskQueue.remove(); 
       
        requireLock = requireLock || tsk.requireLock(); 
       
        if
        (requireLock) { 
       
        break
        ; 
       
        } 
       
        if 
        (tsk 
        instanceof 
        ConditionalTask) { 
       
        taskQueue.addAll(((ConditionalTask)tsk).getListTasks()); 
       
        } 
       
        if
        (tsk.getChildTasks()!= 
        null
        ) { 
       
        taskQueue.addAll(tsk.getChildTasks()); 
       
        } 
       
        // does not add back up task here, because back up task should be the same 
       
        // type of the original task. 
       
        } 
       
        } 
        else 
        { 
       
        requireLock = 
        true
        ; 
       
        } 
       
        } 
       
        if 
        (requireLock) { 
        //获取锁 
       
        ret = acquireReadWriteLocks(); 
       
        if 
        (ret != 
        0
        ) { 
       
        try 
        { 
       
        releaseLocks( ctx.getHiveLocks()); 
       
        } 
        catch 
        (LockException e) { 
       
        // Not much to do here 
       
        } 
       
        return 
        new 
        CommandProcessorResponse(ret, errorMessage, SQLState); 
       
        } 
       
        } 
       
        ret = execute(); 
        //job运行 
       
        if 
        (ret != 
        0
        ) { 
       
        //if needRequireLock is false, the release here will do nothing because there is no lock 
       
        try 
        { 
       
        releaseLocks( ctx.getHiveLocks()); 
       
        } 
        catch 
        (LockException e) { 
       
        // Nothing to do here 
       
        } 
       
        return 
        new 
        CommandProcessorResponse(ret, errorMessage , SQLState ); 
       
        } 
       
        //if needRequireLock is false, the release here will do nothing because there is no lock 
       
        try 
        { 
       
        releaseLocks( ctx.getHiveLocks()); 
       
        } 
        catch 
        (LockException e) { 
       
        errorMessage = 
        "FAILED: Hive Internal Error: " 
        + Utilities.getNameMessage(e); 
       
        SQLState = ErrorMsg. findSQLState(e.getMessage()); 
       
        downstreamError = e; 
       
        console.printError( errorMessage + 
        "\n" 
       
        + org.apache.hadoop.util.StringUtils.stringifyException(e)); 
       
        return 
        new 
        CommandProcessorResponse(
        12
        , errorMessage , SQLState ); 
       
        } 
       
        perfLogger.PerfLogEnd( CLASS_NAME, PerfLogger.DRIVER_RUN); 
       
        perfLogger.close(LOG, plan); 
       
        // Take all the driver run hooks and post-execute them. 
       
        try 
        { 
       
        for 
        (HiveDriverRunHook driverRunHook : driverRunHooks) {  
        //运行HiveDriverRunHook相关类的的postDriverRun方法 
       
        driverRunHook.postDriverRun(hookContext); 
       
        } 
       
        } 
        catch 
        (Exception e) { 
       
        errorMessage = 
        "FAILED: Hive Internal Error: " 
        + Utilities.getNameMessage(e); 
       
        SQLState = ErrorMsg. findSQLState(e.getMessage()); 
       
        downstreamError = e; 
       
        console.printError( errorMessage + 
        "\n" 
       
        + org.apache.hadoop.util.StringUtils.stringifyException(e)); 
       
        return 
        new 
        CommandProcessorResponse(
        12
        , errorMessage , SQLState ); 
       
        } 
       
        return 
        new 
        CommandProcessorResponse(ret); 
       
        }

8）

 
        再来看下compileInternal方法
       
        private 
        static 
        final 
        Object compileMonitor = 
        new 
        Object(); 
       
        private 
        int 
        compileInternal(String command) { 
       
        int 
        ret; 
       
        synchronized 
        ( compileMonitor) { 
       
        ret = compile(command);  
        //调用compile方法 
       
        } 
       
        if 
        (ret != 
        0
        ) { 
       
        try 
        { 
       
        releaseLocks( ctx.getHiveLocks()); 
       
        } 
        catch 
        (LockException e) { 
       
        LOG.warn(
        "Exception in releasing locks. " 
       
        + org.apache.hadoop.util.StringUtils.stringifyException(e)); 
       
        } 
       
        } 
       
        return 
        ret; 
       
        }

调用了compile方法，compile方法分析命令，生成Task,关于compile的具体实现后面详细讲解

9.execute方法,提交task并等待task运行完毕，并打印task运行的信息，比如消耗的时间等

（这里信息也比较多，后面单独讲解

本文转自菜菜光 51CTO博客，原文链接：http://blog.51cto.com/caiguangguang/1571890，如需转载请自行联系原作者

微信关注我们

原文链接：https://yq.aliyun.com/articles/434386

转载内容版权归作者及来源网站所有！

低调大师中文资讯倾力打造互联网数据资讯、行业资源、电子商务、移动互联网、网络营销平台。持续更新报道IT业界、互联网、市场资讯、驱动更新,是最及时权威的产业资讯及硬件资讯报道平台。

hive on tez踩坑记2-hive0.14 on tez

在测试hive0.14.0 on tez时遇到的问题比较多：1.在使用cdh5.2.0+hive0.14.0+tez-0.5.0测试时，首先遇到下面的问题 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 java.lang.NoSuchMethodError:org.apache.tez.dag.api.client.Progress.getFailedTaskAttemptCount()I atorg.apache.hadoop.hive.ql.exec.tez.TezJobMonitor.printStatusInPlace(TezJobMonitor.java: 613 ) atorg.apache.hadoop.hive.ql.exec.tez.TezJobMonitor.monitorExecution(TezJobMonitor.java: 311 ) atorg.apache.hadoop.hive.ql.exec.tez.TezTask.execute(TezTask.java: 167 ) at...

2017-11-15

792

hive0.13 rows loaded为空问题源码分析及fix

升级hive0.13之后发现job运行完成后Rows loaded的信息没有了。 rows loaded的信息在hive0.11中由HiveHistory类的printRowCount输出。HiveHistory类的主要用途是记录job运行的信息，包括task的counter等。默认的目录在/tmp/$user中。 hive0.11在SessionState的start方法中会初始化HiveHistory的对象 1 2 3 if (startSs.hiveHist== null ){ startSs.hiveHist= new HiveHistory(startSs); } 而在hive0.13中HiveHistory是一个抽象类，其具体的实现在HiveHistoryImpl类中，其中初始化HiveHistoryImpl对象时增加了一层判断,判断hive.session.history.enabled的设置（默认为false），导致不会实例化HiveHistoryImpl类 1 2 3 4 5 6 7 8 if (startSs.hiveHist== null ){ if (...

2017-11-15

537

发表评论

资源下载

更多资源

优质分享App

近一个月的开发和优化，本站点的第一个app全新上线。该app采用极致压缩，本体才4.36MB。系统里面做了大量数据访问、缓存优化。方便用户在手机上查看文章。后续会推出HarmonyOS的适配版本。

Nacos

Nacos /nɑ:kəʊs/ 是 Dynamic Naming and Configuration Service 的首字母简称，一个易于构建 AI Agent 应用的动态服务发现、配置管理和AI智能体管理平台。Nacos 致力于帮助您发现、配置和管理微服务及AI智能体应用。Nacos 提供了一组简单易用的特性集，帮助您快速实现动态服务发现、服务配置、服务元数据、流量管理。Nacos 帮助您更敏捷和容易地构建、交付和管理微服务平台。

Spring

Spring框架（Spring Framework）是由Rod Johnson于2002年提出的开源Java企业级应用框架，旨在通过使用JavaBean替代传统EJB实现方式降低企业级编程开发的复杂性。该框架基于简单性、可测试性和松耦合性设计理念，提供核心容器、应用上下文、数据访问集成等模块，支持整合Hibernate、Struts等第三方框架，其适用范围不仅限于服务器端开发，绝大多数Java应用均可从中受益。

Rocky Linux

Rocky Linux（中文名：洛基）是由Gregory Kurtzer于2020年12月发起的企业级Linux发行版，作为CentOS稳定版停止维护后与RHEL（Red Hat Enterprise Linux）完全兼容的开源替代方案，由社区拥有并管理，支持x86_64、aarch64等架构。其通过重新编译RHEL源代码提供长期稳定性，采用模块化包装和SELinux安全架构，默认包含GNOME桌面环境及XFS文件系统，支持十年生命周期更新。