hive升级至0.11 partition schemal问题-低调大师

hive升级至0.11 partition schemal问题

2017-11-13 504

线上由hive0.10升级到hive0.11之后，部分etl job运行出现问题，具体表现为：

在分区表中，当改变table的表的字段类型之后，旧的partition数据可能会出现NULL的情况。

这其实是hive0.11的一个新的特性，每个partition可以有自己的schemal信息。

分析hive0.10和hive0.11 mysql中的元数据信息，发现在table的schemal更改后，两种版本下新的partition都会继承table的schemal，而老的partition不会改变。

在hive0.11这个特性的描述中：

Rows in partitions are now read using partition schema and than made to comply with table schema, instead of being read directly using table schema.

也就是说，hive0.11的partition都是直接读自己的column_v2表中的信息，而hive0.10是读table的信息，这个特性是默认开启的，这就造成在升级到hive0.11之后部分job运行结果有问题。

解决方法：

通过更新mysql中hive元数据的sds表，把partition的cd_id信息都改成对应的table的cd_id的值即可。

实际操作中需要做3个表的join相关操作，在mysql中计算会比较慢，因此把数据load到gp中进行计算，通过python生成update语句做更新(因为涉及到元数据的更新，暂时还是用了手动的方式来更新，另外做了一个监控，当有数据不一致时会有报警邮件产生)：

python代码如下：

 
         import 
         psycopg2 
        
         import 
         time,sys 
        
         class 
         gpUtil: 
        
         def 
         __init__(
         self
         ,host,user,passwd,db): 
        
         self
         .host 
         = 
         host 
        
         self
         .user 
         = 
         user 
        
         self
         .passwd 
         = 
         passwd 
        
         self
         .db 
         = 
         db 
        
         def 
         db_connect(
         self
         ): 
        
         self
         .conn 
         = 
         psycopg2.connect(host
         =
         self
         .host,user
         =
         self
         .user,password
         =
         self
         .passwd,database
         =
         self
         .db) 
        
         self
         .cursor 
         = 
         self
         .conn.cursor() 
        
         def 
         db_fetch_all(
         self
         ,sql): 
        
         self
         .cursor.execute(sql) 
        
         try
         : 
        
         result 
         = 
         self
         .cursor.fetchall() 
        
         except
         : 
        
         result 
         = 
         "Wrong" 
        
         return 
         result 
        
         def 
         db_close(
         self
         ): 
        
         self
         .cursor.close() 
        
         self
         .conn.close() 
        
         def 
         get_list(sql): 
        
         conn 
         = 
         gpUtil(
         'xxxx'
         ,
         'xxxxx'
         ,
         'xxxxx'
         ,
         'xxxxxx'
         ) 
        
         conn.db_connect() 
        
         host_list 
         = 
         conn.db_fetch_all(sql) 
        
         return 
         host_list 
        
         conn.db_close() 
        
         def 
         get_item_list(): 
        
         tble_list 
         = 
         [] 
        
         d 
         = 
         0 
        
         sql 
         = 
         """ 
        
         SELECT TBL_NAME,a.TBL_ID,TABLE_CD_ID,PART_NAME,PART_ID,PARTITION_CD_ID,PARITION_SD_ID FROM                                        
        
         (select PART_ID,PART_NAME,TBL_ID,CD_ID AS PARTITION_CD_ID,b.SD_ID  AS PARITION_SD_ID from vipbi.partitions a,vipbi.sds b where a.SD_ID=b.SD_ID)            
        
         a join (select TBL_ID,TBL_NAME,CD_ID AS TABLE_CD_ID from vipbi.tbls a,vipbi.sds b where a.SD_ID=b.SD_ID) b on a.TBL_ID=b.TBL_ID and a.PARTITION_CD_ID<>b.TABLE_CD_ID
        
         """ 
        
         re 
         = 
         get_list(sql) 
        
         for 
         line 
         in 
         re: 
        
         line 
         = 
         list
         (line) 
        
         table_name 
         = 
         line[
         0
         ] 
        
         table_id 
         = 
         line[
         1
         ] 
        
         table_cd_id 
         = 
         line[
         2
         ] 
        
         part_name 
         = 
         line[
         3
         ] 
        
         part_id 
         = 
         line[
         4
         ] 
        
         part_cd_id 
         = 
         line[
         5
         ] 
        
         part_sd_id 
         =  
         line[
         6
         ] 
        
         tble_list.append([table_name,table_id,table_cd_id,part_name,part_id,part_cd_id,part_sd_id]) 
        
         return 
         tble_list 
        
         def 
         change_partition(tble_list): 
        
         if 
         len
         (tble_list) 
         =
         = 
         0
         : 
        
         print 
         "no need to update" 
        
         sys.exit(
         1
         ) 
        
         now 
         = 
         time.time() 
        
         change_content 
         = 
         '' 
        
         rollback_content 
         =  
         '' 
        
         change_sql 
         = 
         open
         (
         "/home/hdfs/bin/hadoop_tools/sql/meta_change" 
         + 
         str
         (
         int
         (now)) 
         + 
         ".sql"
         ,
         'w+'
         ) 
        
         rollback_sql 
         = 
         open
         (
         "/home/hdfs/bin/hadoop_tools/sql/meta_rollback"  
         + 
         str
         (
         int
         (now)) 
         + 
         ".sql"
         ,
         'w+'
         ) 
        
         for 
         table_item 
         in 
         tble_list: 
        
         table_name 
         = 
         table_item[
         0
         ] 
        
         table_id 
         = 
         table_item[
         1
         ] 
        
         table_cd_id 
         = 
         table_item[
         2
         ] 
        
         part_name 
         = 
         table_item[
         3
         ] 
        
         part_id 
         = 
         table_item[
         4
         ] 
        
         part_cd_id 
         = 
         table_item[
         5
         ] 
        
         part_sd_id 
         = 
         table_item[
         6
         ] 
        
         change_content 
         +
         = 
         """-- table para<name:%s,id:%s,cd_id:%s>; partition para<name:%s,id:%s,cd_id:%s,sd_id:%s>\n""" 
         % 
         (table_name,table_id,table_cd_id,part_name,part_id,part_cd_id,part_sd_id) 
        
         rollback_content 
         +
         = 
         """-- table para<name:%s,id:%s,cd_id:%s>; partition para<name:%s,id:%s,cd_id:%s,sd_id:%s>\n""" 
         % 
         (table_name,table_id,table_cd_id,part_name,part_id,part_cd_id,part_sd_id) 
        
         change_content 
         +
         = 
         """ update sds set CD_ID='%s' where SD_ID='%s';\n"""  
         % 
         (table_cd_id,part_sd_id) 
        
         rollback_content 
         +
         = 
         """ update sds set CD_ID='%s' where SD_ID='%s';\n"""  
         % 
         (part_cd_id,part_sd_id) 
        
         change_sql.write(change_content) 
        
         rollback_sql.write(rollback_content) 
        
         change_sql.close() 
        
         rollback_sql.close() 
        
         if 
         __name__ 
         =
         = 
         '__main__'
         : 
        
         #change_content = '' 
        
         start_time 
         = 
         time.time() 
        
         tble_list 
         = 
         [] 
        
         tble_list 
         = 
         get_item_list() 
        
         change_partition(tble_list) 
        
         stop_time 
         = 
         time.time() 
        
         elapse_time 
         = 
         stop_time 
         - 
         start_time 
        
         print 
         "elapse_time is  %s" 
         % 
         (
         str
         (elapse_time))

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

微信关注我们

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

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

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

使用zabbix做聚合监控

zabbix做为越来越受大家欢迎的监控工具，其相对于nagios,cacti之流，最大的一个特点就是数据是存放在关系型数据库中的，这样就可以极大的方便后续的数据查询，处理等，比如我们想知道一台机器全天ioutil 超过80的时间比例，在zabbix的数据库中，一个sql就可以搞定了，而在cacti中就不这么方便了，而且也不用担心数据随着时间的边长而被稀释掉。在做zabbix的数据分析时，用到的比较多的表一般有hosts,items，interface,hisory*,trend*相关表，比如，通过zabbix监控整个hadoop集群的mapred的使用情况，只需要把每台机器的lastvalue进行聚合就好了。。可以简单通过下面这种方式： 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 #!/usr/bin/python #edit by ericni #to get hadoop totaol statistics # ...

2017-11-14

577

java waitfor导致进程挂起一例

线上使用java做hive的任务调度，通过java Runtime 类的exec方法来执行hive的job。最近发现有job卡死的情况。具体表现是调度的脚本挂起，而且还可能导致hive的lock不能正常释放。使用jstack打印java的thread信息：发现如下的lock，最终定位到waitFor函数。 1 2 3 4 5 6 7 8 "main" prio= 10 tid= 0x000000005b24c800 nid= 0x280e in Object.wait() [ 0x00002b3dee8e7000 ] java.lang.Thread.State: WAITING (on object monitor) at java.lang.Object.wait(Native Method) - waiting on < 0x00000000eb6f88b8 > (a java.lang.UNIXProcess) at java.lang.Object.wait(Object.java: 485 ) at java.lang.UNIXProcess.wa...

2017-11-14

637

发表评论

资源下载

更多资源

优质分享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应用均可从中受益。