Hive表统计信息采集及应用_数据分析

背景

在大数据平台建设过程中，数据治理是不可或缺的一环，优秀的数据治理可以保证数据质量，数据安全等，并能够更有效的利用资源。
以hive为基础构建的数据仓库，在进行数据治理方面，hive表的统计信息是很重要的基础数据，以此数据可以了解hive表的概况，来支持hive表的合理建设和管理。

采集hive表统计信息

方案

hive表分析功能介绍

hive提供了分析表和分区的功能，可以将分析后的统计信息存入元数据中，该功能可以自动执行分析表或手动执行分析表。
自动执行分析主要针对新创建的表，可以通过配置启用，配置说明详见 configuration properties。
hive默认启用表级别的统计信息收集，在 dml （除了load data语句）操作期间，自动收集并更新统计信息。
默认不启用列级别的统计信息收集。
手动执行分析主要针对已存在的表，可以通过执行sql指令完成，示例如下：

analyze table table1 partition(ds='2008-04-09', hr=11) compute statistics;

hive表分析功能详情请参考：statsdev。

hive表统计信息采集方案

在hive应用中，部分表是使用hive dml方式操作数据，这类表可以自动收集统计信息，还有一部分表是非hive dml方式操作数据的，比如外部表，这类表需要手动执行分析表，才能收集的统计信息。
当hive表收集到统计信息后，目前想到两种方式采集统计信息：

方式一：通过hive元数据采集统计信息
方式二：通过hive describe 指令获取统计信息，解析数据并存储

当时考虑，如果使用方式一，需要直接和元数据库交互，可能会对元数据库造成不良影响。而使用方式二，则不会直接和元数据库交互，也许更安全。
最终决定使用方式二进行统计信息采集。
采集指标说明：

指标	describe统计数据	说明
num_files	numfiles	文件数量
num_partitions	numpartitions	分区数量
num_rows	numrows	行数
raw_data_size	rawdatasize	原始数据大小，单位b。parquet格式表的此数据不准，hive4.0修复
total_size	totalsize	hdfs上的总文件大小，单位b，

实现

主要步骤

指定需要手动执行分析的表，并执行分析
获取hive database列表
获取hive每个database的table列表
获取每个表的 describe 指令结果，解析出统计信息

核心代码示例

表结构：

-- 存储 describe 指令结果
create table dg.dg_table_desc
(
  db_name string comment 'database name',
  tbl_name string comment 'table name',
  desc_date string comment '描述日期',
  desc_row_number int comment '清洗后的describe结果数据行号',
  desc_col_name string comment 'describe table col_name',
  desc_data_type string comment 'describe table data_type',
  desc_comment string comment 'describe table comment'
)
comment '表描述信息'
stored as parquet
tblproperties ('parquet.compression' = 'snappy');

-- 存储统计信息
create table dg.dg_table_stats
(
  db_name string comment 'database name',
  tbl_name string comment 'table name',
  stat_date string comment '统计日期',
  num_files bigint comment 'numfiles，表文件数量',
  num_partitions bigint comment 'numpartitions，表分区数量',
  num_rows bigint comment 'numrows，表行数',
  raw_data_size bigint comment 'rawdatasize，原始数据大小，单位b。parquet格式表的此数据不准，hive4.0修复',
  total_size bigint comment 'totalsize，总文件大小，单位b'
)
comment '表统计信息'
stored as parquet
tblproperties ('parquet.compression' = 'snappy');

使用pyhive连接hive：

# 获取spark runtime conf
principal = spark.conf.get('spark.yarn.principal', default=none)
keytab = spark.conf.get('spark.yarn.keytab', default=none)
queue = spark.conf.get('spark.yarn.queue', default=none)

with krbcontext(using_keytab=true, principal=principal, keytab_file=keytab):
    hive_conf = {}
    if queue:
        hive_conf['mapreduce.job.queuename'] = queue
    hive_conn = hive.connect(
        host=args.hiveserver2_host,
        port=args.hiveserver2_port,
        auth='kerberos',
        configuration=hive_conf,
        kerberos_service_name='hive'
    )
    cursor = hive_conn.cursor()

    # 操作hive
    pass
    
    # 关闭连接
    cursor.close()
    hive_conn.close()

手动执行分析表：

cursor.execute(f'analyze table {db_tbl} partition({partition}) compute statistics')

获取hive database列表：

cursor.execute('show databases')
dbs = []
for db_i in cursor.fetchall()
	dbs.append(db_i[0])

获取hive每个database的table列表：

cursor.execute(f'use {db}')
cursor.execute('show tables')
tbls = []
for tbl_i in cursor.fetchall():
    tbls.append(tbl_i[0])

获取每个表的 describe 指令结果

def transform_desc_row(desc_row):
    row = []
    for i in desc_row:
        if i is none:
            row.append(none)
        else:
            i = i.strip()
            row.append(none if i == '' else i)
    return row

cursor.execute(f'describe formatted {db_tbl}')
rn = 1
for i in cursor.fetchall():
    col_name, data_type, comment = transform_desc_row(i)
    if any([col_name, data_type, comment]):
        data.append({
            'db_name': db,
            'tbl_name': tbl,
            'desc_date': desc_date,
            'desc_row_number': rn,
            'desc_col_name': col_name,
            'desc_data_type': data_type,
            'desc_comment': comment
        })
        rn += 1

解析统计信息：

with new_desc as
  (select db_name,
          tbl_name,
          desc_date,
          desc_data_type,
          desc_comment
   from ${dg}.dg_table_desc
   where desc_date >= '${start_dt}'
     and desc_date <= '${end_dt}'
     and desc_data_type in ('numfiles',
                            'numpartitions',
                            'numrows',
                            'rawdatasize',
                            'totalsize'))

insert overwrite table ${dg}.dg_table_stats
select db_name,
       tbl_name,
       stat_date,
       num_files,
       num_partitions,
       num_rows,
       raw_data_size,
       total_size
from ${dg}.dg_table_stats
where stat_date < '${start_dt}'
union all
select db_name,
       tbl_name,
       desc_date,
       max(if(desc_data_type = 'numfiles', cast(desc_comment as bigint), 0)) as num_files,
       max(if(desc_data_type = 'numpartitions', cast(desc_comment as bigint), 0)) as num_partitions,
       max(if(desc_data_type = 'numrows', cast(desc_comment as bigint), 0)) as num_rows,
       max(if(desc_data_type = 'rawdatasize', cast(desc_comment as bigint), 0)) as raw_data_size,
       max(if(desc_data_type = 'totalsize', cast(desc_comment as bigint), 0)) as total_size
from new_desc
group by db_name,
         tbl_name,
         desc_date;