Hive表统计信息采集及应用

背景

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

采集Hive表统计信息

方案

Hive表分析功能介绍

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

ANALYZETABLE Table1 PARTITION(ds='2008-04-09', hr=11)COMPUTESTATISTICS;

Hive表分析功能详情请参考：StatsDev。

Hive表统计信息采集方案

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

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

当时考虑，如果使用方式一，需要直接和元数据库交互，可能会对元数据库造成不良影响。而使用方式二，则不会直接和元数据库交互，也许更安全。
最终决定使用方式二进行统计信息采集。
采集指标说明：
指标DESCRIBE统计数据说明num_filesnumFiles文件数量num_partitionsnumPartitions分区数量num_rowsnumRows行数raw_data_sizerawDataSize原始数据大小，单位B。Parquet格式表的此数据不准，hive4.0修复total_sizetotalSizeHDFS上的总文件大小，单位B，

实现

主要步骤

1. 指定需要手动执行分析的表，并执行分析
2. 获取Hive database列表
3. 获取Hive每个database的table列表
4. 获取每个表的 DESCRIBE 指令结果，解析出统计信息

核心代码示例

表结构：

-- 存储 DESCRIBE 指令结果CREATETABLE dg.dg_table_desc
(
  db_name STRING COMMENT'database name',
  tbl_name STRING COMMENT'table name',
  desc_date STRING COMMENT'描述日期',
  desc_row_number INTCOMMENT'清洗后的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');-- 存储统计信息CREATETABLE dg.dg_table_stats
(
  db_name STRING COMMENT'database name',
  tbl_name STRING COMMENT'table name',
  stat_date STRING COMMENT'统计日期',
  num_files BIGINTCOMMENT'numFiles，表文件数量',
  num_partitions BIGINTCOMMENT'numPartitions，表分区数量',
  num_rows BIGINTCOMMENT'numRows，表行数',
  raw_data_size BIGINTCOMMENT'rawDataSize，原始数据大小，单位B。Parquet格式表的此数据不准，hive4.0修复',
  total_size BIGINTCOMMENT'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()# 操作hivepass# 关闭连接
    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 指令结果

deftransform_desc_row(desc_row):
    row =[]for i in desc_row:if i isNone:
            row.append(None)else:
            i = i.strip()
            row.append(Noneif i ==''else i)return row

cursor.execute(f'describe formatted {db_tbl}')
rn =1for i in cursor.fetchall():
    col_name, data_type, comment = transform_desc_row(i)ifany([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}'UNIONALLSELECT db_name,
       tbl_name,
       desc_date,max(if(desc_data_type ='numFiles', cast(desc_comment ASBIGINT),0))AS num_files,max(if(desc_data_type ='numPartitions', cast(desc_comment ASBIGINT),0))AS num_partitions,max(if(desc_data_type ='numRows', cast(desc_comment ASBIGINT),0))AS num_rows,max(if(desc_data_type ='rawDataSize', cast(desc_comment ASBIGINT),0))AS raw_data_size,max(if(desc_data_type ='totalSize', cast(desc_comment ASBIGINT),0))AS total_size
FROM new_desc
GROUPBY db_name,
         tbl_name,
         desc_date;

应用

目前想到的应用如下。

小文件分析

通过计算文件平均大小，估算Hive表的小文件情况，以此优化Hive存储策略。
文件平均大小 = total_size / num_files 。

增量分析

通过存量采集数据，可以分析Hive表一定周期内的增量（文件数量增量，行数增量，总文件大小增量等）。
例如日增量分析，通过对比Hive表相邻两日的统计信息数据，可以计算出增量指标。
通过增量分析指标，可以进一步辅助数据生命周期管理的建设。