Cloud Functions教程：GCS Bucket传文件时向BigQuery插入数据

刚接触GCP Cloud Functions，想在现有代码中添加功能，实现当特定文件上传至GCS Bucket时，向BigQuery插入数据。不确定是否需要先编写检查文件是否存在的函数，目前已写出部分插入数据的Cloud Function代码及文件检查代码如下：

插入BigQuery的代码 #

from google.cloud import bigquery

def conn_to_bigquery(request):
    client = bigquery.Client()
    query = """
    -- 这里填写你的插入SQL语句
    """
    query_job = client.query(query)

文件检查代码 #

name = 'file_i_want_to_check.txt'   
storage_client = storage.Client()
bucket_name = 'my_bucket_name'
bucket = storage_client.bucket(bucket_name)
stats = storage.Blob(bucket=bucket, name=name).exists(storage_client)

• 无需额外编写文件存在检查函数：
  Cloud Functions的GCS触发事件（如google.storage.object.finalize）只会在文件成功上传并完成写入后触发，事件触发时目标文件必然已存在，不需要额外调用exists()方法检查。

• 整合代码实现需求：
  需创建由GCS上传事件触发的函数，而非HTTP触发的conn_to_bigquery，示例代码如下：

from google.cloud import bigquery, storage

def gcs_to_bigquery(event, context):
    # 从触发事件中获取上传文件的信息
    file_name = event['name']
    bucket_name = event['bucket']
    
    # 仅处理指定文件名的文件
    target_file = 'file_i_want_to_check.txt'
    if file_name != target_file:
        return f"跳过非目标文件: {file_name}"
    
    # 连接BigQuery并执行插入操作
    bq_client = bigquery.Client()
    # 替换为符合你表结构的插入SQL，可根据实际需求从GCS文件读取数据后插入
    insert_query = """
    INSERT INTO `your-project.your-dataset.your-table` (column1, column2)
    VALUES ('value1', 'value2')
    """
    query_job = bq_client.query(insert_query)
    query_job.result()  # 等待插入操作完成
    
    return f"成功处理文件 {file_name}，已插入数据到BigQuery"

• 关键注意事项：
  • 确保Cloud Functions的服务账号拥有GCS对象读取权限和BigQuery数据写入权限；
  • 插入SQL需根据你的表结构和数据来源调整，若要从上传的GCS文件中读取内容插入，可结合storage.Client读取文件内容；
  • 配置Cloud Functions触发器为目标GCS Bucket的「最终化/创建」事件。

内容的提问来源于stack exchange，提问作者Timothy-Ryan25