如何用Cloud Function触发按需调度查询？GCS文件触发场景

核心实现代码 #

以下是对接你已配置的GCS触发事件，用于触发BigQuery加载查询的Python代码：

import os
from google.cloud import bigquery

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

def trigger_bigquery_load(event, context):
    # 可选：获取触发事件中的单个文件路径（如需针对新增的单个文件加载）
    # file_uri = f"gs://{event['bucket']}/{event['name']}"

    # 你的LOAD DATA查询语句
    query = """
        LOAD DATA INTO spreadsheep-20220603.Case_Studies.loading_test
        FROM FILES (
            format='CSV',
            uris=['gs://triggered_upload/*.csv']
        )
        -- 可选：根据CSV实际格式添加配置，比如跳过表头
        -- OPTIONS(skip_leading_rows=1)
    """

    try:
        # 提交BigQuery查询作业
        job = client.query(query)
        # 等待作业完成（按需选择是否同步等待，异步场景可移除该行）
        job.result()
        print(f"BigQuery加载作业执行完成，作业ID：{job.job_id}")
    except Exception as e:
        print(f"BigQuery作业执行失败：{str(e)}")
        raise e

关键配置与注意事项 #

• 依赖配置：部署Cloud Function时，需在requirements.txt中添加依赖：

  google-cloud-bigquery>=3.0.0
  

• 权限设置：确保Cloud Function使用的服务账号拥有以下权限：
  • bigquery.jobs.create：允许提交BigQuery作业
  • bigquery.tables.updateData：允许向目标表追加数据
  • storage.objects.get：允许读取GCS桶内的CSV文件
• 重复处理规避：若要防止同一文件被重复加载，可在函数中记录已处理的文件名（比如存入BigQuery元数据表或Firestore），触发时先校验状态
• CSV格式适配：根据你的CSV实际情况，在LOAD DATA语句中添加OPTIONS配置，比如指定分隔符、跳过空行等

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