如何在BigQuery中获取全项目表最后修改时间并导出至GCS

一、批量获取全项目表的最后修改时间 #

1. 跨数据集统一查询（推荐） #

直接利用BigQuery的INFORMATION_SCHEMA实现跨数据集查询，无需逐个遍历，效率更高：

SELECT
  CONCAT(catalog_name, '.', schema_name, '.', table_name) AS full_table_id,
  DATE(last_modified_time) AS last_modified_date
FROM
  `project_id`.`region-europe-west3`.INFORMATION_SCHEMA.TABLES
WHERE
  table_type = 'BASE TABLE' -- 仅查询物理表，排除视图
ORDER BY
  last_modified_date ASC;

2. 动态拼接查询（兼容旧版元数据表） #

如果依赖__TABLES__元数据表，可通过SQL自动生成所有数据集的查询并合并：

EXECUTE IMMEDIATE (
  SELECT STRING_AGG(
    CONCAT(
      'SELECT "', schema_name, '" AS dataset_id, table_id, DATE(TIMESTAMP_MILLIS(last_modified_time)) AS last_modified FROM `project_id.', schema_name, '.__TABLES__`'
    ), ' UNION ALL '
  )
  FROM `project_id`.`region-europe-west3`.INFORMATION_SCHEMA.SCHEMATA
);

二、导出结果到GCS #

1. 控制台手动导出 #

在BigQuery查询结果页，选择导出→Cloud Storage，配置：

• 存储路径：gs://your-bucket/path/tables_last_modified.csv
• 格式：CSV
• 可选：按导出日期添加路径分区，方便归档

2. 命令行自动化导出（适配GitHub Actions） #

用bq命令行工具完成查询+导出，适合脚本化执行：

# 先将查询结果写入临时表
bq query --use_legacy_sql=false \
"SELECT CONCAT(catalog_name, '.', schema_name, '.', table_name) AS full_table_id, DATE(last_modified_time) AS last_modified_date FROM `project_id`.`region-europe-west3`.INFORMATION_SCHEMA.TABLES WHERE table_type = 'BASE TABLE'" \
--destination_table project_id.temp_dataset.temp_table \
--replace

# 导出临时表到GCS，按日期命名文件
bq extract --destination_format CSV \
project_id.temp_dataset.temp_table \
gs://your-bucket/path/tables_last_modified_$(date +%Y%m%d).csv

三、GitHub Actions自动化流程 #

核心步骤 #

• 权限配置：给GitHub Actions关联的服务账号授予BigQuery数据查看权限、GCS写入权限
• 定时触发：通过schedule设置每日运行（例：UTC凌晨2点）
• 执行导出：调用bq命令完成查询和GCS导出
• 过期表检查：用脚本读取CSV，筛选出最后修改时间早于60天的表
• 告警推送：通过邮件、Slack机器人等工具发送告警通知

示例Python检查脚本 #

import pandas as pd
from datetime import datetime, timedelta
from google.cloud import storage

# 从GCS下载CSV文件
storage_client = storage.Client()
bucket = storage_client.get_bucket('your-bucket')
blob = bucket.blob(f'path/tables_last_modified_{datetime.now().strftime("%Y%m%d")}.csv')
blob.download_to_filename('tables_last_modified.csv')

# 筛选过期表
df = pd.read_csv('tables_last_modified.csv')
df['last_modified_date'] = pd.to_datetime(df['last_modified_date'])
expired_threshold = datetime.now() - timedelta(days=60)
expired_tables = df[df['last_modified_date'] < expired_threshold]

# 生成告警内容并推送（示例：打印，实际替换为Slack/邮件逻辑）
if not expired_tables.empty:
    alert_content = f"发现{len(expired_tables)}张超过2个月未修改的表：\n{expired_tables['full_table_id'].to_string(index=False)}"
    print(alert_content)

四、注意事项 #

• 遵循最小权限原则：服务账号仅授予必要的BigQuery和GCS权限
• 定期清理GCS上的历史CSV文件，避免存储冗余
• 可给临时表设置TTL自动过期，减少资源占用

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