如何在BigQuery表加载完成后自动触发调度器

下面是几种无需人工干预、在表加载完成后自动触发查询的可行方案：

方案一：事件驱动（BigQuery + Pub/Sub + Cloud Functions） #

这是最直接的触发式方案，通过捕获表加载完成的事件来启动查询：

1. 创建Pub/Sub主题
   先建立一个接收BigQuery事件的主题：

   gcloud pubsub topics create bq-table-load-events
   

2. 配置BigQuery事件触发器
   为目标表创建事件触发器，当表完成数据加载（包括批量导入、流式写入完成等场景）时，将事件消息发送到上面的Pub/Sub主题：

   gcloud eventarc triggers create bq-load-trigger \
     --location=us-central1 \
     --destination-topic=bq-table-load-events \
     --event-filters="type=google.cloud.bigquery.v2.tableDataUpdated" \
     --event-filters="project_id=你的项目ID" \
     --event-filters="dataset_id=你的数据集ID" \
     --event-filters="table_id=你的目标表ID"
   

3. 编写Cloud Functions执行查询
   创建一个订阅上述Pub/Sub主题的Cloud Function，函数逻辑中调用BigQuery API执行你的目标查询。示例Python代码：

   from google.cloud import bigquery
   
   def run_bigquery_query(event, context):
       client = bigquery.Client()
       query = """
           -- 这里替换成你需要执行的查询语句
           INSERT INTO `目标数据集.目标表`
           SELECT * FROM `源数据集.源加载表` WHERE ...
       """
       job = client.query(query)
       job.result()  # 等待查询执行完成
       print("查询已成功触发并执行")
   

   部署函数时关联到之前的Pub/Sub主题即可。

方案二：工作流编排（Cloud Workflows） #

如果你的数据加载本身就是自动化作业（比如定时导出到GCS再加载到BigQuery），可以把加载和查询放到同一个工作流里，实现顺序执行：

1. 定义Workflow流程
   创建一个YAML格式的工作流配置，先执行数据加载作业，等待完成后再触发查询：

   main:
     steps:
       - load_data_to_bq:
           call: googleapis.bigquery.v2.jobs.insert
           args:
             projectId: 你的项目ID
             body:
               configuration:
                 load:
                   sourceUris: ["gs://你的存储桶/数据文件.csv"]
                   destinationTable:
                     projectId: 你的项目ID
                     datasetId: 你的数据集ID
                     tableId: 目标加载表
                   skipLeadingRows: 1
                   sourceFormat: "CSV"
       - wait_for_load:
           call: googleapis.bigquery.v2.jobs.get
           args:
             projectId: 你的项目ID
             jobId: ${load_data_to_bq.body.jobReference.jobId}
           retry:
             predicate: ${wait_for_load.body.status.state == "RUNNING"}
             max_retries: 30
             backoff:
               initial_delay: 10
               multiplier: 2
       - run_query:
           call: googleapis.bigquery.v2.jobs.insert
           args:
             projectId: 你的项目ID
             body:
               configuration:
                 query:
                   query: "INSERT INTO `目标数据集.结果表` SELECT * FROM `目标加载表` WHERE ..."
                   useLegacySql: false
   

2. 部署并触发工作流
   用gcloud命令部署工作流，之后可以通过定时触发器或者API调用启动工作流，实现加载完成后自动执行查询。

方案三：加载作业内置查询触发（针对批量加载） #

如果你的数据是通过bq load命令或BigQuery加载API导入的，可以在加载作业完成后直接链式执行查询，不需要额外事件系统：

比如用bash脚本组合加载和查询命令：

# 执行数据加载
bq load --source_format=CSV `你的项目ID.数据集.目标表` gs://存储桶/数据.csv 字段定义

# 检查加载是否成功，成功则执行查询
if [ $? -eq 0 ]; then
  bq query --use_legacy_sql=false "INSERT INTO `结果表` SELECT * FROM `目标表` WHERE ..."
fi

将这个脚本放到Cloud Scheduler中定时执行，就能实现加载完成后自动跑查询。

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