Apache Beam WriteToBigQuery写入失败，无有效错误日志求助

使用Apache Beam的WriteToBigQuery类写入BigQuery时失败，相关代码、Schema格式及错误信息如下：

代码实现 #

def write_to_bigquery(args):
    dictionary_rows = "Create Linked Dictionary" >> beam.Map(
        _translate_matched_series_to_dict
    )
    return dictionary_rows | "Write to BigQuery" >> beam.io.WriteToBigQuery(
        args.output_table,
        schema=_get_table_schema(),
        write_disposition=beam.io.BigQueryDisposition.WRITE_TRUNCATE,
        create_disposition=beam.io.BigQueryDisposition.CREATE_IF_NEEDED,
        custom_gcs_temp_location=f"gs://{args.temp_storage_bucket}",
    )

Schema格式 #

[{"name": field.name, "type": type_str, "mode": "NULLABLE"}, ...更多字段, ]

错误信息 #

File "/usr/local/lib/python3.8/site-packages/apache_beam/io/gcp/bigquery_tools.py", line 637, in wait_for_bq_job
    raise RuntimeError(
RuntimeError: BigQuery job beam_bq_job_LOAD_alminoralllinkagetest20230501_LOAD_STEP_818_e472c3092b9a929012fe506ee87f1d12_01b64c3e2a604526a0fcfd0f18ec05f5 failed. Error Result: <ErrorProto
 location: 'gs://some_url_A'
 message: 'Error while reading data, error message: JSON table encountered too many errors, giving up. Rows: 6830; errors: 1. Please look into the errors[] collection for more details. File: 'gs://some_url_A'
 reason: 'invalid'> [while running 'Create Linked Dictionary|Write to BigQuery/Write to BigQuery/BigQueryBatchFileLoads/TriggerLoadJobsWithoutTempTables/ParDo(TriggerLoadJobs)-ptransform-70']

当前困境：错误指向的GCS文件仅展示待插入内容，无具体错误日志，无法定位问题根源。

1. 直接查看BigQuery作业详情 #

在BigQuery控制台搜索错误信息中的作业ID（beam_bq_job_LOAD_xxx），进入作业详情页后切换到错误标签，这里会显示完整的错误明细，包括具体出错的行号、字段不匹配类型等核心信息——Beam抛出的错误仅做了上层封装，BigQuery控制台会保留最原始的错误上下文。

2. 启用Beam详细日志 #

通过调整日志级别，捕获BigQuery返回的errors[]集合内容：

• 若使用DirectRunner，在代码开头添加：

  import logging
  logging.basicConfig(level=logging.DEBUG)
  

• 若使用DataflowRunner，提交作业时添加参数：

  --worker-logging-level=DEBUG
  

详细日志会输出BigQuery加载过程中的具体错误细节，帮你定位问题行。

3. 前置数据校验 #

在写入BigQuery前添加校验步骤，提前拦截不符合Schema的数据：

def _validate_row(row, schema):
    schema_field_map = {field['name']: field for field in schema}
    for field_name, field_value in row.items():
        if field_name not in schema_field_map:
            raise ValueError(f"未知字段: {field_name}")
        target_type = schema_field_map[field_name]['type']
        # 根据BigQuery类型做基础校验，可按需扩展
        if target_type == 'INT64' and not isinstance(field_value, (int, float)):
            raise ValueError(f"字段{field_name}值{field_value}不符合INT64类型")
        elif target_type == 'DATE' and not isinstance(field_value, str):
            raise ValueError(f"字段{field_name}值{field_value}需为日期格式字符串")
    return row

# 在WriteToBigQuery前插入校验步骤
dictionary_rows = "Create Linked Dictionary" >> beam.Map(_translate_matched_series_to_dict)
validated_rows = dictionary_rows | "Validate Rows" >> beam.Map(_validate_row, schema=_get_table_schema())
return validated_rows | "Write to BigQuery" >> beam.io.WriteToBigQuery(...)

4. 检查GCS临时文件格式 #

BigQuery要求JSON加载文件为JSON Lines格式（每行一个独立JSON对象），可临时将数据写入GCS验证格式：

dictionary_rows | "Write Temp JSON" >> beam.io.WriteToText(
    f"gs://{args.temp_storage_bucket}/temp_data.json",
    file_name_suffix='.jsonl'
)

下载文件后检查每行是否为合法JSON对象，排查是否存在引号不闭合、多余逗号等格式问题。

内容的提问来源于stack exchange，提问作者Alceu.wv