求编写Python/BQ CLI脚本：批量校验BigQuery表多环境存在性并输出结果

这个脚本可遍历输入文件中的表名，检查三个指定BigQuery项目的所有数据集，识别staging表，并按要求输出格式化结果。

依赖安装 #

首先安装Google Cloud BigQuery Python客户端：

pip install google-cloud-bigquery

注意：需提前完成Google Cloud认证，比如通过gcloud auth application-default login或配置服务账号密钥，确保拥有三个项目的BigQuery数据查看权限。

完整脚本 #

from google.cloud import bigquery
from typing import Dict, Tuple

# 配置三个环境的项目名
PROJECTS = {
    "prod": "my-project-prod",
    "test": "my-project-test",
    "dev": "my-project-dev"
}

def check_table_exists(client: bigquery.Client, project_id: str, base_table_name: str) -> Tuple[bool, bool]:
    """
    检查指定项目中是否存在目标表，以及是否存在对应的staging表（表名+_stg在_stg结尾的dataset中）
    返回：(原表是否存在任意数据集, 是否存在staging表)
    """
    base_exists = False
    stg_exists = False
    
    # 获取项目下所有数据集
    datasets = client.list_datasets(project_id)
    
    for dataset in datasets:
        dataset_id = dataset.dataset_id
        # 检查原表是否存在当前数据集
        table_ref = client.dataset(dataset_id, project=project_id).table(base_table_name)
        try:
            client.get_table(table_ref)
            base_exists = True
        except:
            pass
        
        # 检查staging表（仅当数据集以_stg结尾时）
        if dataset_id.endswith("_stg"):
            stg_table_name = f"{base_table_name}_stg"
            stg_table_ref = client.dataset(dataset_id, project=project_id).table(stg_table_name)
            try:
                client.get_table(stg_table_ref)
                stg_exists = True
            except:
                pass
    
    return base_exists, stg_exists

def determine_action(prod_exists: bool, stg_exists: bool, dev_test_exists: bool) -> str:
    """根据规则确定action列的值"""
    if stg_exists:
        return "DELETE"
    if prod_exists:
        return "DELETE"
    if dev_test_exists:
        return "DROP"
    return "NONE"  # 表在所有环境都不存在的情况

def main(input_file: str, output_file: str):
    client = bigquery.Client()
    
    # 读取输入文件中的表名
    with open(input_file, "r") as f:
        table_names = [line.strip() for line in f if line.strip()]
    
    # 准备输出内容
    output_lines = [
        "tablename | exists_in_prod | exists_in_test | exists_in_dev | is_stg_table | action"
    ]
    
    for table_name in table_names:
        # 检查各个环境的表存在情况
        prod_base, prod_stg = check_table_exists(client, PROJECTS["prod"], table_name)
        test_base, test_stg = check_table_exists(client, PROJECTS["test"], table_name)
        dev_base, dev_stg = check_table_exists(client, PROJECTS["dev"], table_name)
        
        # 合并stg存在情况（任意环境的stg表存在就算）
        is_stg_table = any([prod_stg, test_stg, dev_stg])
        # 各环境的存在标记（Y/N）
        exists_prod = "Y" if prod_base else "N"
        exists_test = "Y" if test_base else "N"
        exists_dev = "Y" if dev_base else "N"
        
        # 判断dev/test是否存在（只要其中一个存在就算）
        dev_test_exists = test_base or dev_base
        # 确定action
        action = determine_action(prod_base, is_stg_table, dev_test_exists)
        
        # 格式化行，保证列对齐
        formatted_line = (
            f"{table_name:<8} |       {exists_prod}        |       {exists_test}        |      {exists_dev}        |     {'Y' if is_stg_table else 'N'}        | {action}"
        )
        output_lines.append(formatted_line)
    
    # 写入输出文件
    with open(output_file, "w") as f:
        f.write("\n".join(output_lines))
    
    print(f"结果已成功写入到 {output_file}")

if __name__ == "__main__":
    # 替换成你的输入文件和输出文件路径
    INPUT_FILE = "tables.txt"
    OUTPUT_FILE = "table_check_results.txt"
    main(INPUT_FILE, OUTPUT_FILE)

使用说明 #

1. 修改脚本中的INPUT_FILE和OUTPUT_FILE为实际的文件路径
2. 运行脚本：

python bq_table_check.py

关键逻辑说明 #

• 表存在性检查：遍历每个项目下的所有数据集，分别检查原表和对应staging表（仅在_stg结尾的数据集）的存在性
• is_stg_table列：只要任意环境的_stg数据集中存在原表名_stg，就标记为Y
• action列优先级：
  1. 存在staging表 → DELETE
  2. Prod环境存在原表 → DELETE
  3. 仅Dev/Test环境存在原表 → DROP
  4. 所有环境都不存在 → NONE

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