Python实现从GCS按日期子目录创建对应BigQuery表

核心实现代码 #

以下代码会遍历GCS指定路径下的日期分区目录，自动提取日期生成对应BQ表名，并加载csv.gz文件，同时优化了Schema自动检测的稳定性：

from google.cloud import storage, bigquery
import re

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

# 配置参数
GCS_BUCKET_NAME = "your-bucket-name"
GCS_BASE_PATH = "data/"
BQ_DATASET_ID = "your-dataset-id"

# 遍历GCS中的日期分区目录
bucket = storage_client.get_bucket(GCS_BUCKET_NAME)
blobs = bucket.list_blobs(prefix=GCS_BASE_PATH)

# 去重获取所有日期分区目录（避免重复处理同一目录下的多个文件）
date_dirs = set()
for blob in blobs:
    # 匹配date=YYYY-MM-DD格式的目录
    match = re.search(r'date=(\d{4}-\d{2}-\d{2})', blob.name)
    if match and blob.name.endswith('/'):
        date_dirs.add(match.group(1))

# 处理每个日期分区
for date_str in date_dirs:
    # 转换日期格式：2023-04-21 → 20230421
    formatted_date = date_str.replace('-', '')
    table_id = f"{BQ_DATASET_ID}.sessions_{formatted_date}"
    
    # 构建GCS文件路径
    gcs_uri = f"gs://{GCS_BUCKET_NAME}/{GCS_BASE_PATH}date={date_str}/*.csv.gz"
    
    # 配置加载作业
    job_config = bigquery.LoadJobConfig(
        source_format=bigquery.SourceFormat.CSV,
        skip_leading_rows=1,  # 假设csv有表头
        autodetect=True,
        compression=bigquery.Compression.GZIP,
        max_bad_records=5,  # 允许少量错误记录，避免直接失败
    )
    
    try:
        # 启动加载作业
        load_job = bq_client.load_table_from_uri(
            gcs_uri, table_id, job_config=job_config
        )
        load_job.result()  # 等待作业完成
        
        # 验证结果
        destination_table = bq_client.get_table(table_id)
        print(f"成功创建表 {table_id}，包含 {destination_table.num_rows} 行数据")
    except Exception as e:
        print(f"处理日期 {date_str} 时出错: {str(e)}")
        # 可选：添加自动检测失败后的 fallback 逻辑

解决自动Schema检测错误的方案 #

自动Schema检测偶尔失败通常是因为采样数据不足以推断正确字段类型，或部分字段格式不一致，可以通过以下方式优化：

• 提升采样数据有效性：确保对应日期目录下至少有一个数据量足够的文件（比如不要只有几行测试数据），让BQ能基于足够样本推断字段类型。
• 手动定义基础Schema：如果某些字段类型固定，先定义基础Schema，让BQ自动补充其他字段：

  # 示例：手动定义核心字段，其余自动检测
  base_schema = [
      bigquery.SchemaField("user_id", "STRING"),
      bigquery.SchemaField("session_duration", "INTEGER"),
  ]
  job_config.schema = base_schema
  job_config.autodetect = True
  

• 强化数据容错：合理调整max_bad_records参数，允许少量异常行存在，避免因个别格式错误导致整个加载作业失败。

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