Colab中使用job_config指定Schema写入表至BigQuery的配置问题

现有代码的核心问题 #

• 你定义的schemma2是pandas-gbq库的旧版Schema元组格式，谷歌官方BigQuery Python客户端不识别，直接传会抛出参数类型错误。另外变量名多打了一个m，正确拼写是schema。
• 直接给QueryJobConfig传schema无法实现强制类型校验：WRITE_TRUNCATE模式下如果目标表已存在，默认会沿用表本身的Schema；如果表不存在，没开自动建表的话任务直接失败，还是会出现字段类型被自动转换的问题。
• schema_update_options参数完全不需要配置：这个参数是用来允许写入时自动修改表结构（比如新增字段、放宽字段必填限制）的，和你要固定Schema、禁止字段类型变动的需求完全冲突，开了反而会增加类型漂移的风险。

合法的Schema定义格式 #

官方客户端支持两种可正常识别的Schema格式，优先选第一种，类型校验更严格不容易写错：

1. 原生SchemaField对象格式（推荐）

from google.cloud import bigquery
# 所有字段按这个格式定义即可，要固定Year为STRING就加对应行
FIXED_SCHEMA = [
    bigquery.SchemaField("Region", "STRING", mode="NULLABLE"),
    bigquery.SchemaField("Product", "STRING", mode="NULLABLE"),
    # bigquery.SchemaField("Year", "STRING", mode="NULLABLE"),
]

2. 字典列表格式（就是你写的schema_1的格式，注意type值必须全大写，比如STRING/INT64，不能写小写）

schema_1 = [
    {"name": "Region", "type": "STRING", "mode": "NULLABLE"},
    {"name": "Product", "type": "STRING", "mode": "NULLABLE"}
]

注意：元组列表格式的schemma2不要用，官方客户端不认。

可直接运行的修正代码 #

这个实现完全跳过pandas DataFrame环节，性能符合要求，全量覆写时严格固定Schema，只要字段类型不匹配直接报错终止，不会出现类型偷偷转换导致下游崩溃的问题：

from google.cloud import bigquery
from google.cloud.exceptions import NotFound

# --------------------------
# 所有复用场景只需要修改这里的固定Schema定义即可
FIXED_SCHEMA = [
    bigquery.SchemaField("Region", "STRING", mode="NULLABLE"),
    bigquery.SchemaField("Product", "STRING", mode="NULLABLE"),
]
project_id = 'nproject'
dataset_id = "natality_regression"
table_id = "regression_input"
# --------------------------

client = bigquery.Client(project=project_id)
table_id_full = f"{project_id}.{dataset_id}.{table_id}"

# 写入前预处理表，保证表结构和固定Schema完全一致
try:
    table = client.get_table(table_id_full)
    # 校验已存在的表结构是否符合预期，不符合直接抛错阻止写入
    exist_field_set = {(f.name, f.field_type, f.mode) for f in table.schema}
    expect_field_set = {(f.name, f.field_type, f.mode) for f in FIXED_SCHEMA}
    if exist_field_set != expect_field_set:
        raise RuntimeError(f"目标表结构不符合要求，现有字段：{exist_field_set}，预期字段：{expect_field_set}")
except NotFound:
    # 表不存在时先校验数据集，不存在则创建数据集，再按固定Schema建表
    dataset_ref = client.dataset(dataset_id)
    try:
        client.get_dataset(dataset_ref)
    except NotFound:
        client.create_dataset(dataset_ref)
    new_table = bigquery.Table(table_id_full, schema=FIXED_SCHEMA)
    client.create_table(new_table)

# 配置写入任务
job_config = bigquery.QueryJobConfig(
    destination=table_id_full,
    write_disposition=bigquery.WriteDisposition.WRITE_TRUNCATE,
    autodetect=False, # 关键：关闭自动Schema检测，禁止BigQuery自动转换字段类型
    # 不需要配置schema_update_options，默认即禁止修改表结构
)

query = """
    SELECT * FROM `nproject.SalesData1.Sales1` LIMIT 15
"""

# 执行任务
query_job = client.query(query, job_config=job_config)
query_job.result()
print("写入完成")

关键注意事项 #

• 必须设置autodetect=False：默认配置下BigQuery会自动推断查询结果的Schema，遇到类型不匹配会尝试做隐式转换，关闭后只要查询结果字段类型和目标表对不上，任务直接失败，不会偷偷改字段类型。
• 提前做Schema校验：避免其他协作者手动修改了目标表结构，导致写入的数据类型不符合预期，提前拦截问题比下游崩溃后排查成本低很多。
• 如果你后续要加字段，只需要修改FIXED_SCHEMA的定义，同步更新目标表结构即可，不要开schema_update_options让BigQuery自动改表。

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