使用Dataflow Apache Beam向BigQuery写入数据时bcdate字段报错

核心问题分析 #

1. 数据格式不匹配：transform_pandas返回的是包含单个字典的列表（如[{'peakid': ...}]），但Beam的WriteToBigQuery要求每个元素是单个字典对象，列表结构会导致BigQuery解析JSON时出错。
2. 字段类型冲突：
   • heightm字段以字符串形式（如'6055'）输出，但BigQuery schema定义为INTEGER，字符串无法直接转换为整数。
   • bcdate字段若BigQuery schema为DATETIME类型，当前仅输出%Y-%m-%d（无时间部分）不符合要求，DATETIME需要YYYY-MM-DD HH:MM:SS格式；若实际存储的是日期数据，应将字段类型改为DATE。
3. 空值处理错误：使用字符串'null'填充空列，后续类型转换时会引发混乱，BigQuery仅识别JSON中的null为合法NULL值。

修复代码 #

1. 修改转换函数transform_pandas

def transform_pandas(data):
    import pandas as pd
    import json

    df = pd.DataFrame([data])
    
    # 用Python原生None填充空列，而非字符串'null'
    columns = ['peakid', 'route1', 'bcdate', 'pkname', 'heightm']
    df = df.reindex(columns, fill_value=None, axis=1)
    
    # 处理bcdate：根据BigQuery字段类型选择格式
    # 若为DATE类型，保留'%Y-%m-%d'；若为DATETIME，改为'%Y-%m-%d 00:00:00'
    df['bcdate'] = pd.to_datetime(df['bcdate'], errors='coerce').dt.strftime('%Y-%m-%d')
    
    # 将heightm转为整数，空值保留为None
    df['heightm'] = pd.to_numeric(df['heightm'], errors='coerce').astype('Int64')
    
    # 返回单个字典对象，而非列表
    return json.loads(df.to_json(orient='records'))[0]

2. 明确BigQuery Schema并修正写入逻辑

output = (
        (input_p1, input_p2)
        | 'Join' >> beam.CoGroupByKey()
        | 'Final Dict' >> beam.Map(lambda el: to_final_dict(el[1]))
        | 'Transformation' >> beam.Map(transform_pandas)
        | beam.Map(print)
        | 'Write To BigQuery' >> beam.io.gcp.bigquery.WriteToBigQuery(
           table='project:dataset.expeditions',
           # 根据实际字段类型调整：bcdate选DATE或DATETIME
           schema='peakid:STRING,route1:STRING,bcdate:DATE,pkname:STRING,heightm:INTEGER',
           method='FILE_LOADS',
           custom_gcs_temp_location='gs://bucket/folder/temp',
           create_disposition=beam.io.BigQueryDisposition.CREATE_IF_NEEDED,
           write_disposition=beam.io.BigQueryDisposition.WRITE_TRUNCATE)    
    )

关键修复点说明 #

• 返回单个字典：通过[0]提取JSON数组中的单个对象，符合Beam写入BigQuery的数据格式要求。
• 统一空值处理：用None替代字符串'null'，确保JSON输出的null能被BigQuery正确识别为字段NULL值。
• 类型对齐：将heightm转换为支持空值的整数类型，与BigQuery的INTEGER字段匹配；调整bcdate格式以匹配BigQuery的DATE/DATETIME类型要求。
• 明确Schema：关闭自动推断，手动指定字段类型，避免BigQuery错误推断导致的加载失败。

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