BigQuery中bq load加载CSV时如何忽略数据类型不匹配的行？

一、直接通过bq load命令调整参数解决 #

1. 忽略指定数量的错误行 #

在bq load命令中添加--max_bad_records=N参数（N为允许的错误行数，比如设为100），当错误行数量不超过N时，加载任务会继续执行，错误行被自动忽略。

修改后的命令示例：

bq load --schema=***.json --skip_leading_rows=1 --replace --source_format=CSV --max_bad_records=100 prod*** gs://consolidated_report.csv

2. 自定义日期格式解析 #

针对日期格式不匹配的问题，无需修改源文件，直接在JSON schema中为日期字段指定对应格式。比如针对disputeDt字段（CSV日期格式为MM/DD/YYYY），修改schema文件：

[
  ...,
  {
    "name": "disputeDt",
    "type": "DATE",
    "format": "%m/%d/%Y"
  },
  ...
]

BigQuery会自动按照指定格式解析日期，避免格式报错。

二、临时表中转清洗（精准控制错误处理） #

如果错误行较多，需要精准过滤或留存错误数据用于排查，可先将CSV导入全字符串类型的临时表，再用SQL清洗后写入目标表：

1. 创建临时表（所有字段设为STRING类型）：

bq load --schema=temp_schema.json --skip_leading_rows=1 --replace --source_format=CSV temp_dataset.temp_table gs://consolidated_report.csv

temp_schema.json中所有字段的type设为STRING。

2. SQL清洗并插入目标表：

INSERT INTO prod***
SELECT
  SAFE_CAST(int_field AS INT64) AS int_field, -- 安全转换，失败返回NULL
  SAFE.PARSE_DATE("%m/%d/%Y", disputeDt) AS disputeDt, -- 安全解析日期，失败返回NULL
  -- 其他字段按需转换
FROM temp_dataset.temp_table
WHERE
  SAFE_CAST(int_field AS INT64) IS NOT NULL -- 过滤int转换失败的行
  AND SAFE.PARSE_DATE("%m/%d/%Y", disputeDt) IS NOT NULL -- 过滤日期解析失败的行

若需留存错误行，可同步插入错误记录表：

-- 插入正确数据到目标表
INSERT INTO prod***
SELECT ... FROM temp_dataset.temp_table WHERE ...

-- 插入错误数据到错误表
INSERT INTO error_dataset.error_table
SELECT ... FROM temp_dataset.temp_table WHERE ...

三、用Airflow原生Operator替代bash命令 #

Airflow的GCSToBigQueryOperator支持更多错误处理参数，适配ETL场景更灵活：

from airflow.providers.google.cloud.transfers.gcs_to_bigquery import GCSToBigQueryOperator

load_task = GCSToBigQueryOperator(
    task_id='load_csv_to_bq',
    bucket='your_bucket',
    source_objects=['consolidated_report.csv'],
    destination_project_dataset_table='prod***',
    schema_fields=[
        {"name": "disputeDt", "type": "DATE", "mode": "NULLABLE", "format": "%m/%d/%Y"},
        # 其他字段定义
    ],
    skip_leading_rows=1,
    write_disposition='WRITE_TRUNCATE', # 对应bq load的--replace参数
    source_format='CSV',
    max_bad_records=100, # 允许的错误行数
    ignore_unknown_values=True, # 忽略CSV中不存在于schema的字段
    allow_jagged_rows=True, # 允许行末尾缺少字段
    dag=dag
)

该Operator可直接处理日期格式，同时通过max_bad_records忽略错误行，无需额外编写Python清洗逻辑。

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