如何按文件夹名称自动将GCS数据加载至BigQuery并创建对应表

问题背景 #

此前为降低成本，将BigQuery中的GA4数据迁移至Cloud Storage（GCS），现需导回部分数据。GA4数据原按event_YYYYMMDD格式每日生成表，GCS中存储路径为gs://project-name-big-query-analytics_propertyId/events_/events_YYYYMMDD，需要高效自动创建表并完成数据导回。已有Python脚本，但不清楚如何在GCP环境中落地使用。

一、脚本核心调整 #

先修正原脚本中的关键问题，确保适配GA4数据特性与GCP存储规则：

1. 前缀路径修正：GCS的prefix无需前置斜杠，将folder_prefix改为events_/events_以匹配目标路径
2. GA4 Schema适配：替换占位字段为GA4实际字段（如event_name、event_timestamp等），嵌套字段需按GA4结构定义
3. 文件名处理优化：改用更可靠的方式提取表名，避免前缀长度计算错误

调整后的脚本示例：

import os
from google.cloud import bigquery
from google.cloud import storage

def load_data_to_bigquery(event, context):
    # 配置信息
    project_id = "app-xxxx"
    dataset_id = "analytics_xxxx"
    bucket_name = "app-xxx-big-query-analytics_xxxx"
    folder_prefix = "events_/events_"

    # 初始化客户端
    bq_client = bigquery.Client(project=project_id)
    storage_client = storage.Client(project=project_id)

    # 列出指定前缀下的所有文件（跳过文件夹）
    blobs = storage_client.list_blobs(bucket_name, prefix=folder_prefix)
    for blob in blobs:
        if blob.name.endswith('/'):
            continue
        
        # 提取表名：从文件名中移除后缀，如events_20240101.json → events_20240101
        table_name = os.path.splitext(os.path.basename(blob.name))[0]

        # 创建表（不存在则新建）
        table_ref = bq_client.dataset(dataset_id).table(table_name)
        try:
            bq_client.get_table(table_ref)
        except Exception:
            # 按GA4标准Schema定义，可根据实际需求补充字段
            table = bigquery.Table(table_ref)
            table.schema = [
                bigquery.SchemaField("event_name", "STRING"),
                bigquery.SchemaField("event_timestamp", "INTEGER"),
                bigquery.SchemaField("user_pseudo_id", "STRING"),
                bigquery.SchemaField("event_params", "RECORD", mode="REPEATED", fields=[
                    bigquery.SchemaField("key", "STRING"),
                    bigquery.SchemaField("value", "RECORD", fields=[
                        bigquery.SchemaField("string_value", "STRING"),
                        bigquery.SchemaField("int_value", "INTEGER"),
                        bigquery.SchemaField("float_value", "FLOAT")
                    ])
                ])
            ]
            bq_client.create_table(table)

        # 加载数据到BigQuery
        job_config = bigquery.LoadJobConfig()
        job_config.source_format = bigquery.SourceFormat.NEWLINE_DELIMITED_JSON
        job_config.write_disposition = bigquery.WriteDisposition.WRITE_APPEND
        # 可选：开启自动推断Schema（仅在字段不确定时使用）
        # job_config.autodetect = True

        uri = f"gs://{bucket_name}/{blob.name}"
        load_job = bq_client.load_table_from_uri(uri, table_ref, job_config=job_config)
        load_job.result()  # 等待任务完成

        print(f"数据已加载至表 {table_name}")

if __name__ == "__main__":
    load_data_to_bigquery(None, None)  # 本地测试入口

二、GCP环境运行方案 #

方式1：本地测试运行 #

1. 安装依赖：

   pip install google-cloud-bigquery google-cloud-storage
   

2. 配置认证：
   • 在GCP控制台创建服务账号，下载JSON格式密钥文件
   • 设置环境变量指定密钥路径：

     export GOOGLE_APPLICATION_CREDENTIALS="/path/to/your/service-account-key.json"
     

3. 执行脚本：

   python your_script_name.py
   

方式2：部署为Cloud Function（自动触发） #

适合需监控GCS新增文件并自动导回的场景：

1. 准备文件：
   • 将调整后的脚本保存为main.py
   • 创建requirements.txt：

     google-cloud-bigquery>=3.10.0
     google-cloud-storage>=2.10.0
     

2. 部署函数：
   • 进入GCP控制台 → Cloud Functions → 创建函数
   • 触发器类型选择Cloud Storage，事件类型为最终创建，绑定目标存储桶
   • 运行时选择Python 3.10+，上传main.py和requirements.txt，入口点填写load_data_to_bigquery
   • 配置权限：确保函数服务账号拥有BigQuery数据编辑者和Cloud Storage对象查看者权限
3. 触发测试：上传GA4数据文件到GCS指定路径，函数会自动触发数据加载

三、关键注意事项 #

• Schema一致性：GA4包含大量嵌套字段，建议从原BigQuery GA4表导出准确Schema，避免数据加载失败
• 成本控制：可批量处理文件而非单文件触发，或设置定时任务集中加载，降低BigQuery费用
• 错误处理：可添加异常捕获逻辑，将错误日志写入Cloud Logging，方便排查问题

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