如何批量删除BigQuery表指定日期前的分区？寻求高效SQL方案

问题描述 #

需要删除BigQuery中多数据集、多表内特定日期之前的分区，优先考虑SQL相关实现：

• 尝试过常规DELETE语句：

  DELETE FROM tablename WHERE _partitiontime < '2021-01-01'
  

  但该语句会扫描全量数据，成本极高，仅适合一次性操作，无法集成到定期运行的多表项目中。
• BigQuery不支持通过SQL直接执行DROP PARTITION操作，仅能通过BQ CLI或API单分区删除，串行操作耗时极长。
• 曾尝试从INFORMATION_SCHEMA获取分区ID并循环执行DROP，但BigQuery不支持该语法：

  FOR record IN
    (SELECT partition_id FROM dataset.INFORMATION_SCHEMA.PARTITIONS) 
  DO
    DROP dataset.table$record.partition_id; 
  END FOR;
  

寻求适配该场景的高效替代方案。

高效解决方案 #

方案1：生成批量BQ CLI命令并行执行 #

通过查询INFORMATION_SCHEMA筛选目标分区，直接生成批量删除命令，再批量执行：

1. 执行以下SQL生成删除命令：

SELECT CONCAT(
  'bq rm -f -t `', table_catalog, '.', table_schema, '.', table_name, '$', partition_id, '`'
) AS cli_delete_command
FROM `你的项目ID.目标数据集.INFORMATION_SCHEMA.PARTITIONS`
WHERE partition_id < '20210101' -- 分区ID格式为YYYYMMDD，对应要保留的起始日期
  -- 可添加table_name筛选特定表，去掉则批量处理数据集内所有表
  AND table_name = '目标表名'

2. 将查询结果的cli_delete_command列导出为shell脚本（如delete_partitions.sh），赋予执行权限后运行，可通过并行执行命令大幅缩短耗时。

方案2：Python脚本异步调用BQ API批量删除 #

利用BigQuery Python客户端库，结合INFORMATION_SCHEMA结果，异步批量删除分区，适合集成到定期调度任务：

from google.cloud import bigquery
import asyncio

# 初始化BigQuery客户端
client = bigquery.Client(project="你的项目ID")

async def delete_single_partition(table_full_id, partition_id):
    # 拼接分区的完整ID
    partition_full_id = f"{table_full_id}${partition_id}"
    try:
        # 异步删除分区
        await client.delete_table(partition_full_id, not_found_ok=True)
        print(f"已删除分区: {partition_full_id}")
    except Exception as e:
        print(f"删除分区失败 {partition_full_id}: {str(e)}")

def get_target_partitions():
    # 查询需要删除的分区
    query = """
    SELECT table_catalog, table_schema, table_name, partition_id
    FROM `你的项目ID.目标数据集.INFORMATION_SCHEMA.PARTITIONS`
    WHERE partition_id < '20210101'
    """
    query_job = client.query(query)
    return query_job.result()

async def batch_delete_partitions():
    partitions = get_target_partitions()
    # 创建异步任务列表
    tasks = []
    for row in partitions:
        table_full_id = f"{row.table_catalog}.{row.table_schema}.{row.table_name}"
        tasks.append(delete_single_partition(table_full_id, row.partition_id))
    # 并行执行所有任务
    await asyncio.gather(*tasks)

if __name__ == "__main__":
    asyncio.run(batch_delete_partitions())

方案3：存储过程生成动态命令（间接实现） #

通过存储过程生成批量删除命令，再外部执行，适合标准化多表处理流程：

CREATE OR REPLACE PROCEDURE 目标数据集.generate_partition_delete_commands(
  IN project_id STRING,
  IN dataset_id STRING,
  IN cutoff_date STRING -- 输入格式为'YYYY-MM-DD'
)
BEGIN
  -- 将日期转换为分区ID格式YYYYMMDD
  DECLARE cutoff_partition_id STRING DEFAULT FORMAT_DATE("%Y%m%d", DATE(cutoff_date));
  
  -- 生成批量删除命令
  SELECT CONCAT(
    'bq rm -f -t `', project_id, '.', dataset_id, '.', table_name, '$', partition_id, '`'
  ) AS delete_command
  FROM `project_id.dataset_id.INFORMATION_SCHEMA.PARTITIONS`
  WHERE partition_id < cutoff_partition_id;
END;

调用存储过程后导出结果，即可批量执行删除命令。

内容的提问来源于stack exchange，提问作者Vikrant Bhosale