hive0.13 rows loaded为空问题源码分析及fix-低调大师

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

2017-11-14 538

升级hive0.13之后发现job运行完成后Rows loaded的信息没有了。

rows loaded的信息在hive0.11中由HiveHistory类的printRowCount输出。HiveHistory类的主要用途是记录job运行的信息，包括task的counter等。默认的目录在/tmp/$user中。

hive0.11在SessionState 的start方法中会初始化HiveHistory的对象

 
        if 
        (startSs. hiveHist == 
        null
        ) { 
       
        startSs. hiveHist = 
        new 
        HiveHistory(startSs); 
       
        }

而在hive0.13中HiveHistory是一个抽象类，其具体的实现在HiveHistoryImpl类中，其中初始化HiveHistoryImpl对象时增加了一层判断,判断hive.session.history.enabled的设置（默认为false），导致不会实例化HiveHistoryImpl类

 
        if
        (startSs.hiveHist == 
        null
        ){ 
       
        if 
        (startSs.getConf().getBoolVar(HiveConf.ConfVars.HIVE_SESSION_HISTORY_ENABLED)) { 
       
        startSs.hiveHist = 
        new 
        HiveHistoryImpl (startSs); 
       
        }
        else 
        { 
       
        //Hive history is disabled, create a no-op proxy 
       
        startSs.hiveHist = HiveHistoryProxyHandler .getNoOpHiveHistoryProxy(); 
       
        } 
       
        }

在fix这个配置之后，仍然没有发现rows loaded的信息，通过分析源码

printRowCount方法的实现如下：

 
        public 
        void 
        printRowCount(String queryId) { 
       
        QueryInfo ji = queryInfoMap.get(queryId); 
       
        if 
        (ji == 
        null
        ) {   
        // 如果ji为空，则直接返回 
       
        return
        ; 
       
        } 
       
        for 
        (String tab : ji. rowCountMap.keySet()) { 
       
        console.printInfo(ji. rowCountMap.get(tab) + 
        " Rows loaded to " 
        + tab); 
        // 从hashmap中获取数据 
       
        } 
       
        }

在hive0.13中，这里获取的ji对象是空值。

近一步发现，是由于counter中没有TABLE_ID_(\\d+)_ROWCOUNT，导致不能匹配ROW_COUNT_PATTERN的正则。就不能正常获取的row count的值。

其中获取tasker count的rows loaded信息的getRowCountTableName方法内容如下:

 
        private 
        static 
        final 
        String ROW_COUNT_PATTERN = 
        "TABLE_ID_(\\d+)_ROWCOUNT"
        ; 
       
        private 
        static 
        final 
        Pattern rowCountPattern = Pattern.compile(ROW_COUNT_PATTERN); 
       
        ......
       
        String getRowCountTableName(String name) { 
       
        if 
        (idToTableMap == 
        null
        ) { 
       
        return 
        null
        ; 
       
        } 
       
        Matcher m = rowCountPattern.matcher(name); 
       
        if 
        (m.find()) {  
        // //没有和TABLE_ID_xxxx match的counter导致，即counter没有打印出TABLE_ID_(\\d+)_ROWCOUNT导致。。 
       
        String tuple = m.group(
        1
        ); 
       
        return 
        idToTableMap.get(tuple); 
       
        } 
       
        return 
        null
        ; 
       
        }

而TABLE_ID_(\\d+)_ROWCOUNT是由FileSinkOperator类负责写入的。hive0.11中相关的代码如下：

 
        protected 
        void 
        initializeOp(Configuration hconf) 
        throws 
        HiveException { 
       
        ..........
       
        int 
        id = conf.getDestTableId(); 
       
        if 
        ((id != 
        0
        ) && (id <= TableIdEnum. values().length)) { 
       
        String enumName = 
        "TABLE_ID_" 
        + String.valueOf(id) + 
        "_ROWCOUNT"
        ; 
       
        tabIdEnum = TableIdEnum.valueOf(enumName); 
       
        row_count = 
        new 
        LongWritable(); 
       
        statsMap.put( tabIdEnum, row_count ); 
       
        }

而在hive0.13中这部分代码都被去掉了，找到了原因，fix也比较简单，把这个counter加回去就可了。

patch如下：

 
        diff --git a/ql/src/java/org/apache/hadoop/hive/ql/exec/FileSinkOperator.java b/ql/src/java/org/apache/hadoop/hive/ql/exec/FileSinkOperator.java
       
        index 1dde78e..96860f7 
        100644 
       
        --- a/ql/src/java/org/apache/hadoop/hive/ql/exec/FileSinkOperator.java
       
        +++ b/ql/src/java/org/apache/hadoop/hive/ql/exec/FileSinkOperator.java
       
        @@ -
        68
        ,
        13 
        +
        68
        ,
        16 
        @@ 
       
        import 
        org.apache.hadoop.util.ReflectionUtils; 
       
        import 
        com.google.common.collect.Lists; 
       
        +
        import 
        org.apache.commons.logging.Log; 
       
        +
        import 
        org.apache.commons.logging.LogFactory; 
       
        +
       
        /**
       
        * File Sink operator implementation. 
       
        **/ 
       
        public 
        class 
        FileSinkOperator 
        extends 
        TerminalOperator<FileSinkDesc> 
        implements 
       
        Serializable { 
       
        -
       
        +  
        public 
        static 
        Log LOG = LogFactory.getLog(
        "FileSinkOperator.class"
        ); 
       
        protected 
        transient 
        HashMap<String, FSPaths> valToPaths; 
       
        protected 
        transient 
        int 
        numDynParts; 
       
        protected 
        transient 
        List<String> dpColNames; 
       
        @@ -
        214
        ,
        6 
        +
        217
        ,
        7 
        @@ 
        public 
        Stat getStat() { 
       
        protected 
        transient 
        FileSystem fs; 
       
        protected 
        transient 
        Serializer serializer; 
       
        protected 
        transient 
        LongWritable row_count; 
       
        +  
        protected 
        transient 
        TableIdEnum tabIdEnum = 
        null
        ; 
       
        private 
        transient 
        boolean 
        isNativeTable = 
        true
        ; 
       
        /** 
       
        @@ -
        241
        ,
        6 
        +
        245
        ,
        23 
        @@ 
        public 
        Stat getStat() { 
       
        protected 
        transient 
        JobConf jc; 
       
        Class<? 
        extends 
        Writable> outputClass; 
       
        String taskId; 
       
        +  
        public 
        static 
        enum 
        TableIdEnum { 
       
        +       TABLE_ID_1_ROWCOUNT,
       
        +       TABLE_ID_2_ROWCOUNT,
       
        +       TABLE_ID_3_ROWCOUNT,
       
        +       TABLE_ID_4_ROWCOUNT,
       
        +       TABLE_ID_5_ROWCOUNT,
       
        +       TABLE_ID_6_ROWCOUNT,
       
        +       TABLE_ID_7_ROWCOUNT,
       
        +       TABLE_ID_8_ROWCOUNT,
       
        +       TABLE_ID_9_ROWCOUNT,
       
        +       TABLE_ID_10_ROWCOUNT,
       
        +       TABLE_ID_11_ROWCOUNT,
       
        +       TABLE_ID_12_ROWCOUNT,
       
        +       TABLE_ID_13_ROWCOUNT,
       
        +       TABLE_ID_14_ROWCOUNT,
       
        +       TABLE_ID_15_ROWCOUNT;
       
        +  }
       
        protected 
        boolean 
        filesCreated = 
        false
        ; 
       
        @@ -
        317
        ,
        7 
        +
        338
        ,
        15 
        @@ 
        protected 
        void 
        initializeOp(Configuration hconf) 
        throws 
        HiveException { 
       
        prtner = (HivePartitioner<HiveKey, Object>) ReflectionUtils.newInstance( 
       
        jc.getPartitionerClass(), 
        null
        ); 
       
        } 
       
        -      row_count = 
        new 
        LongWritable(); 
       
        +      
        //row_count = new LongWritable(); 
       
        +         
        int 
        id = conf.getDestTableId(); 
       
        +         
        if 
        ((id != 
        0
        ) && (id <= TableIdEnum.values().length)) { 
       
        +               String enumName = 
        "TABLE_ID_" 
        + String.valueOf(id) + 
        "_ROWCOUNT"
        ;   
       
        +               tabIdEnum = TableIdEnum.valueOf(enumName);
       
        +               row_count = 
        new 
        LongWritable(); 
       
        +               statsMap.put(tabIdEnum, row_count);
       
        +         }
       
        +
       
        if 
        (dpCtx != 
        null
        ) { 
       
        dpSetup(); 
       
        }

打完patch后，重新打包，替换线上的hive-exec-xxx.jar包之后测试，rows loaded的数据又回来了。

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

微信关注我们

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

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

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

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

Driver类是对 1 org.apache.hadoop.hive.ql.processors.CommandProcessor.java 接口的实现，重写了run方法，定义了常见sql的执行方式. 1 public class Driver implements CommandProcessor 具体的方法调用顺序： 1 2 run--->runInternal--->(createTxnManager+recordValidTxns)----->compileInternal---> compile--analyzer(BaseSemanticAnalyzer)--->execute 其中compile和execute是两个比较重要的方法： compile用来完成语法和语义的分析，生成执行计划 execute执行物理计划，即提交相应的mapredjob 通过打印perflog可以看到Driver类的简单地时序图：下面来看下Driver类的几个常用的方法实现： 1)createTxnManager 用来获取目前设置的用于实现lock的类，比如: 1 ...

2017-11-15

623

hive的几个排序

hive中常用的几个排序 order by #全局排序，因为是全局排序，所以job是一个reduce，reduce的个数参数设置对其无效 sort by #一个reduce时为全局排序，多个reduce时，每个reduce各自排序，为了提高全局排序的性能，可以先用sort by做局部排序，然后再做全局排序 distribute by #hash 分组，根据key和reduce个数进行数据分发，默认使用hash的方式 cluster by #distribute by + sort by（注意和hive.enforce.bucketing参数的应用）结合两个例子来看排序的应用： 1）cdnjob的优化思路熟悉map slot的分配规则的同学应该知道，text gz的文件是不支持split的，因此这种情况下最多只有一个map（不管数据量多大），这样在做数据处理时，如果文件很大而且计算逻辑比较复杂（比如cdn的报表，需要做各种聚合计算和ip地址的解析），效率就会比较低下。一个思路就是建一个中间表，对原始表进行distribute by，对中间表进行复杂的计算，这样就可以使用多个m...

2017-11-15

622

发表评论

资源下载

更多资源

Mario

马里奥是站在游戏界顶峰的超人气多面角色。马里奥靠吃蘑菇成长，特征是大鼻子、头戴帽子、身穿背带裤，还留着胡子。与他的双胞胎兄弟路易基一起，长年担任任天堂的招牌角色。

腾讯云软件源

为解决软件依赖安装时官方源访问速度慢的问题，腾讯云为一些软件搭建了缓存服务。您可以通过使用腾讯云软件源站来提升依赖包的安装速度。为了方便用户自由搭建服务架构，目前腾讯云软件源站支持公网访问和内网访问。

Rocky Linux

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

Sublime Text

Sublime Text具有漂亮的用户界面和强大的功能，例如代码缩略图，Python的插件，代码段等。还可自定义键绑定，菜单和工具栏。Sublime Text 的主要功能包括：拼写检查，书签，完整的 Python API ， Goto 功能，即时项目切换，多选择，多窗口等等。Sublime Text 是一个跨平台的编辑器，同时支持Windows、Linux、Mac OS X等操作系统。