如何程序化获取并导出BigQuery定时查询列表至CSV等格式？

一、获取定时查询数据 #

BigQuery提供官方CLI和API接口获取定时查询信息，无需手动复制。你需要具备BigQuery Scheduled Queries Viewer或更高权限（如BigQuery Admin）。

用gcloud CLI快速获取 #

执行以下命令导出所有定时查询的JSON格式数据：

gcloud bigquery scheduled-queries list --format=json

返回的JSON包含所需全部字段：

• 显示名称：displayName
• 查询URL：通过项目ID和查询ID拼接，格式为 https://console.cloud.google.com/bigquery?project=[你的项目ID]&page=scheduledqueries&queryId=[查询ID]（查询ID从name字段最后一段提取，如projects/xxx/locations/xxx/scheduledQueries/[查询ID]）
• 调度时间（UTC）：schedule（Cron表达式格式）
• 下次调度时间：nextRunTime
• 创建者：creator
• 目标数据集/表：destinationTable.datasetId 和 destinationTable.tableId

二、导出到目标格式 #

1. 导出到CSV #

借助jq工具过滤字段并生成CSV（需先安装jq）：

PROJECT_ID=$(gcloud config get-value project)
gcloud bigquery scheduled-queries list --format=json | jq -r '
["显示名称","URL","调度时间(UTC)","下次调度时间","创建者","目标数据集","目标表"],
(.[] | [
  .displayName,
  "https://console.cloud.google.com/bigquery?project='"$PROJECT_ID"'&page=scheduledqueries&queryId=" + (.name | split("/")[-1]),
  .schedule,
  .nextRunTime,
  .creator,
  .destinationTable.datasetId,
  .destinationTable.tableId
]) | @csv' > scheduled_queries.csv

2. 导出到Google Sheets #

用Python脚本结合gspread库实现自动化写入：

import subprocess
import json
import gspread
from oauth2client.service_account import ServiceAccountCredentials

# 获取定时查询数据
cmd_result = subprocess.run(
    ["gcloud", "bigquery", "scheduled-queries", "list", "--format=json"],
    capture_output=True, text=True
)
queries = json.loads(cmd_result.stdout)

# 准备表头和数据行
headers = ["显示名称", "URL", "调度时间(UTC)", "下次调度时间", "创建者", "目标数据集", "目标表"]
data_rows = [headers]
project_id = subprocess.run(
    ["gcloud", "config", "get-value", "project"],
    capture_output=True, text=True
).stdout.strip()

for q in queries:
    query_id = q["name"].split("/")[-1]
    query_url = f"https://console.cloud.google.com/bigquery?project={project_id}&page=scheduledqueries&queryId={query_id}"
    data_rows.append([
        q["displayName"],
        query_url,
        q["schedule"],
        q["nextRunTime"],
        q["creator"],
        q["destinationTable"]["datasetId"],
        q["destinationTable"]["tableId"]
    ])

# 连接并写入Google Sheets
scope = ["https://spreadsheets.google.com/feeds", "https://www.googleapis.com/auth/drive"]
creds = ServiceAccountCredentials.from_json_keyfile_name("你的服务账号密钥文件.json", scope)
client = gspread.authorize(creds)

# 替换成你的Sheet名称和工作表
sheet = client.open("BigQuery定时查询列表").sheet1
sheet.clear()
sheet.update(data_rows)

注意：需创建Google Cloud服务账号，授予Sheets编辑权限，并下载密钥文件。

3. 导出到BigQuery表 #

先将JSON数据保存为文件，再导入到目标表：

# 导出JSON数据
gcloud bigquery scheduled-queries list --format=json > scheduled_queries_raw.json

# 创建目标表并导入数据（替换成你的数据集和表名）
bq load --source_format=NEWLINE_DELIMITED_JSON \
  --autodetect \
  your_dataset.scheduled_queries_list \
  scheduled_queries_raw.json

--autodetect会自动识别字段类型，也可手动指定表结构匹配需求。

三、自动化更新 #

用Cloud Scheduler创建定时任务，定期执行上述脚本/命令（比如每天凌晨执行一次），确保数据实时更新。

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