如何提取Google BigQuery现有项目DDL至SQL文件用于版本控制？

方法一：使用bq命令行工具编写Shell脚本 #

这是最直接的方案，无需额外依赖，适合快速批量提取：

#!/bin/bash
PROJECT_ID="你的项目ID"
REGION="你的BigQuery区域（如us-central1）"
OUTPUT_DIR="./bq_ddl"

# 创建根输出目录
mkdir -p "$OUTPUT_DIR"

# 获取当前项目下的所有数据集列表（过滤表头）
DATASETS=$(bq ls --project_id="$PROJECT_ID" --format=csv | tail -n +2 | cut -d',' -f1)

for DATASET in $DATASETS; do
    # 为当前数据集创建视图、例程子目录
    DATASET_DIR="$OUTPUT_DIR/$DATASET"
    mkdir -p "$DATASET_DIR/views" "$DATASET_DIR/routines"

    # 提取视图完整DDL并写入文件
    bq query --use_legacy_sql=false --format=csv "SELECT table_name, DDL FROM \`$REGION\`.INFORMATION_SCHEMA.VIEWS WHERE table_catalog='$PROJECT_ID' AND table_schema='$DATASET'" | tail -n +2 | while read -r VIEW_NAME DDL; do
        echo "$DDL" > "$DATASET_DIR/views/$VIEW_NAME.sql"
    done

    # 提取例程（存储过程/函数）完整DDL并写入文件
    bq query --use_legacy_sql=false --format=csv "SELECT routine_name, DDL FROM \`$REGION\`.INFORMATION_SCHEMA.ROUTINES WHERE routine_catalog='$PROJECT_ID' AND routine_schema='$DATASET'" | tail -n +2 | while read -r ROUTINE_NAME DDL; do
        echo "$DDL" > "$DATASET_DIR/routines/$ROUTINE_NAME.sql"
    done

    echo "已处理数据集: $DATASET"
done

echo "DDL提取完成，文件保存至 $OUTPUT_DIR"

方法二：使用Python脚本（灵活扩展） #

如果需要更复杂的逻辑（比如过滤特定名称的对象、处理跨区域数据集），可以用google-cloud-bigquery库：

1. 先安装依赖：

pip install google-cloud-bigquery

2. 脚本示例：

from google.cloud import bigquery
import os

PROJECT_ID = "你的项目ID"
REGION = "你的BigQuery区域（如us-central1）"
OUTPUT_DIR = "./bq_ddl"

# 初始化BigQuery客户端
client = bigquery.Client(project=PROJECT_ID)

# 创建根输出目录
os.makedirs(OUTPUT_DIR, exist_ok=True)

# 遍历项目下所有数据集
for dataset in client.list_datasets():
    dataset_id = dataset.dataset_id
    dataset_dir = os.path.join(OUTPUT_DIR, dataset_id)
    # 创建视图、例程子目录
    os.makedirs(os.path.join(dataset_dir, "views"), exist_ok=True)
    os.makedirs(os.path.join(dataset_dir, "routines"), exist_ok=True)

    # 提取视图DDL
    views_query = f"""
        SELECT table_name, DDL
        FROM `{REGION}.INFORMATION_SCHEMA.VIEWS`
        WHERE table_catalog = '{PROJECT_ID}'
          AND table_schema = '{dataset_id}'
    """
    for row in client.query(views_query).result():
        view_path = os.path.join(dataset_dir, "views", f"{row.table_name}.sql")
        with open(view_path, "w", encoding="utf-8") as f:
            f.write(row.DDL)

    # 提取例程DDL
    routines_query = f"""
        SELECT routine_name, DDL
        FROM `{REGION}.INFORMATION_SCHEMA.ROUTINES`
        WHERE routine_catalog = '{PROJECT_ID}'
          AND routine_schema = '{dataset_id}'
    """
    for row in client.query(routines_query).result():
        routine_path = os.path.join(dataset_dir, "routines", f"{row.routine_name}.sql")
        with open(routine_path, "w", encoding="utf-8") as f:
            f.write(row.DDL)

    print(f"已处理数据集: {dataset_id}")

print(f"提取完成，文件保存至 {OUTPUT_DIR}")

注意事项 #

• 确保已完成BigQuery授权：Shell脚本用gcloud auth login，Python脚本可通过gcloud auth application-default login或设置GOOGLE_APPLICATION_CREDENTIALS环境变量
• 替换脚本中的PROJECT_ID和REGION为实际值
• 若存在跨区域数据集，需调整脚本遍历所有目标区域的INFORMATION_SCHEMA
• 提取的DDL包含完整的CREATE VIEW/CREATE PROCEDURE/CREATE FUNCTION语句，可直接用于版本控制

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