当前位置：首页 > news >正文

使用 PySpark 批量清理 Hive 表历史分区

news 来源：原创 2025/4/19 8:20:16

使用 PySpark 批量清理 Hive 表历史分区

在大数据平台中，Hive 表通常采用分区方式存储数据，以提升查询效率和数据管理的灵活性。随着数据的不断积累，历史分区会越来越多，既占用存储空间，也影响元数据管理性能。因此，定期清理过期的历史分区是数据治理的重要环节。本文将介绍如何利用 PySpark 批量清理 Hive 表的历史分区。

一、背景说明

Hive 分区表常用于按时间（如按天、月）组织数据。随着时间推移，早期的分区数据可能已不再需要，但这些分区依然占用 HDFS 空间和 Hive 元数据。手动删除分区效率低且容易出错，采用 PySpark 可以实现自动化、批量化的分区清理。

二、实现思路

通过 SparkSession 连接 Hive（可以通过hive引擎执行）。
获取目标表的所有分区信息。
根据设定的保留策略（如保留最近 N 天），筛选出需要删除的历史分区。
批量执行分区删除操作。

三、代码实现

以下为 PySpark 批量清理 Hive 表历史分区的示例代码：

#******************************************************************#
##author: david.zhou
##create time: 2025-02-26 13:24:11
#******************************************************************#
#!/usr/bin/env python
# -*- coding: utf-8 -*-from pyspark.sql import SparkSession
from datetime import datetime, timedelta
import logging# 配置日志
logging.basicConfig(level=logging.INFO,format='%(asctime)s - %(levelname)s - %(message)s')# 生成需要检查的历史日期列表（检查1098天的历史分区）
#check_days = 1098
TABLE_PARTITON_CHECK_DAYS = 100# 配置要处理的表和保留天数
TABLES_CONFIG = {'ods.ods_sec_admin_operate_log': 60,'sale.dwd_pc_wx_msg_df': 60
}def is_external_table(spark, table_name):"""判断 Hive 表是否为外部表"""# 查询表的元数据result = spark.sql(f"DESCRIBE FORMATTED {table_name}").collect()# 查找是否包含 "EXTERNAL" 标志for row in result:if 'EXTERNAL' in row[1]:print(f"{table_name} is_external_table result is {row}")return Truereturn Falsedef get_spark_session():return SparkSession.builder \.appName("HivePartitionCleaner") \.enableHiveSupport() \.config("spark.sql.sources.partitionOverwriteMode", "dynamic") \.getOrCreate()def generate_partition_dates(end_date, days):"""生成需要检查的历史分区日期列表"""dates = []current = end_datefor i in range(days):dates.append(current.strftime('%Y%m%d'))current = current - timedelta(days=1)return dates# 检查分区是否存在
def check_partition_exists(spark, table_name, partition_spec):partitions = spark.sql(f"SHOW PARTITIONS {table_name} PARTITION ({partition_spec}) ").collect()if not partitions:return Falseelse:return Truedef drop_partitions(spark, table_name, retention_days):"""按日期循环删除历史分区"""logging.info(f"开始处理表 {table_name}, 保留 {retention_days} 天数据")try:# 计算截止日期cutoff_date = datetime.now() - timedelta(days=retention_days)logging.info(f"开始处理表 {table_name}, 保留 {retention_days} 天数据, 将清理 {cutoff_date} 之前分区数据")# 生成需要检查的历史日期列表（检查1098天的历史分区）#check_days = 1098partition_dates = generate_partition_dates(cutoff_date, TABLE_PARTITON_CHECK_DAYS)dropped_count = 0for dt in partition_dates:try:# 构建删除单个分区的SQL# TODO: 非 pt 分区单独处理drop_sql = f"""ALTER TABLE {table_name} DROP IF EXISTS PARTITION (pt='{dt}')"""# 执行SQL并获取结果logging.info(f"准备删除分区SQL: {drop_sql}")result = spark.sql(drop_sql)if not check_partition_exists(spark, table_name, f"pt='{dt}'"):print(f"分区 pt='{dt}' 删除成功")dropped_count += 1logging.info(f"成功删除分区: {table_name} - pt={dt}")else:print(f"分区 pt='{dt}' 删除失败")except Exception as e:logging.error(f"删除分区失败 {table_name} - pt={dt}: {str(e)}")continuelogging.info(f"表 {table_name} 处理完成: 删除分区数 {dropped_count}")except Exception as e:logging.error(f"处理表 {table_name} 时发生错误: {str(e)}")def main():logging.info("开始批量清理分区任务")spark = get_spark_session()try:for table_name, retention_days in TABLES_CONFIG.items():#is_external_table(spark, table_name)if is_external_table(spark, table_name):print(f"表 {table_name} 是外部表")logging.error(f"{table_name} 是外部表")else:print(f"表 {table_name} 不是外部表")drop_partitions(spark, table_name, retention_days)finally:spark.stop()logging.info("批量清理分区任务完成")if __name__ == "__main__":main()