使用GCP Dataflow导入JSON至BigQuery时遇加载错误求助

问题原因 #

你的代码中beam.Map(print) 是核心问题：print函数执行后返回None，导致后续写入BigQuery的数据流全部变为null，最终GCS临时存储的文件内容为空，触发BigQuery加载失败。

修正方案 #

要么直接移除beam.Map(print)操作，要么修改为既打印数据又返回原数据的逻辑，确保数据流能正常传递到后续步骤。

修正后的代码示例 #

import apache_beam as beam
import apache_beam.io.gcp.bigquery as b_query

p1 = beam.Pipeline()
trips_schema = 'trip_id:INTEGER,vendor_id:INTEGER,trip_distance:FLOAT,fare_amount:STRING,store_and_fwd_flag:STRING'

freq = (
        p1
        | beam.Create(["{\"vendor_id\":33,\"trip_id\": 1000474,\"trip_distance\": 2.3999996185302734,\"fare_amount\": 42.13,\"store_and_fwd_flag\": \"Y\"}"])
        # 保留打印同时返回原数据，避免丢失数据流
        | beam.Map(lambda x: (print(x), x)[1])
        | 'Write Record to BigQuery' >> b_query.WriteToBigQuery(
            table='trips2', 
            dataset='my_poc',
            custom_gcs_temp_location='gs://XXXX-stage-xxxx/temp',
            schema=trips_schema, 
            project='xxxxxxxx-xxx-2',
            write_disposition=beam.io.BigQueryDisposition.WRITE_APPEND,
            create_disposition=beam.io.BigQueryDisposition.CREATE_IF_NEEDED
        )
)
p1.run()

补充说明 #

Dataflow的Transform链式操作依赖每个步骤返回有效数据传递到下一个环节。如果某个Transform返回None，后续步骤处理的就是空数据，写入GCS的临时文件自然为空，BigQuery无法解析这种无效数据，就会抛出"JSON table encountered too many errors"的加载失败错误。

内容的提问来源于stack exchange，提问作者danny.lesnik