使用pandas_gbq上传DataFrame至BigQuery遇CSV处理错误的解决问询

问题背景 #

要上传的DataFrame列类型如下：

name            object
type            object
population       int32
geometry      geometry
geojson         object
dtype: object

各字段说明：

• name：区域名称字符串
• type：区域类型（如省、市等）字符串
• population：区域人口数值
• geometry：shapely multipolygon类型
• geojson：通过df['geojson'] = df['geometry'].apply(lambda x: json.dumps(shapely.geometry.mapping(x)))生成的GeoJSON格式字符串

上传时触发错误：

pandas_gbq.gbq.GenericGBQException: Reason: 400 Error while reading data, error message: CSV processing encountered too many errors, giving up. Rows: 855402; errors: 8; max bad: 0; error percent: 0

已完成排查：

• 第855402行数据类型与其他行一致：

>>> df.iloc[855402:855403].dtypes
name            object
type            object
population       int32
geometry      geometry
geojson         object
dtype: object

• 该行geometry对象验证有效：

>>> df.iloc[855402]['geometry'].is_valid
True

解决方案 #

1. 清理GeoJSON字符串的特殊字符 #

虽然geometry对象有效，但json.dumps生成的字符串可能包含换行符、制表符等干扰CSV解析的字符。先检查目标行的GeoJSON内容：

print(df.iloc[855402]['geojson'])

若发现特殊字符，重新生成时做清理：

import json
import shapely.geometry

def clean_geojson(geom):
    geojson_str = json.dumps(shapely.geometry.mapping(geom))
    return geojson_str.replace('\n', '').replace('\t', '')

df['geojson'] = df['geometry'].apply(clean_geojson)

2. 允许上传时容错 #

BigQuery默认拒绝任何错误行，可通过参数设置允许一定数量的错误记录：

import pandas_gbq

pandas_gbq.to_gbq(
    df,
    destination_table="your_project.your_dataset.your_table",
    project_id="your_project_id",
    if_exists="replace",
    params={"allow_jagged_rows": True, "max_bad_records": 10}
)

3. 映射GeoJSON到BigQuery地理类型 #

BigQuery支持GEOGRAPHY类型，可直接将geojson列映射为该类型，同时移除原生geometry列避免类型冲突：

# 删除geometry列
df = df.drop(columns=['geometry'])

# 定义表结构
schema = [
    {"name": "name", "type": "STRING"},
    {"name": "type", "type": "STRING"},
    {"name": "population", "type": "INTEGER"},
    {"name": "geojson", "type": "GEOGRAPHY"}
]

# 上传
pandas_gbq.to_gbq(
    df,
    destination_table="your_project.your_dataset.your_table",
    project_id="your_project_id",
    if_exists="replace",
    table_schema=schema
)

4. 改用Parquet格式上传 #

CSV对复杂字符串兼容性差，Parquet格式能完整保留数据类型，避免解析错误：

# 保存为Parquet文件
df.to_parquet("temp_data.parquet")

# 使用BigQuery客户端上传
from google.cloud import bigquery

client = bigquery.Client(project="your_project_id")
job_config = bigquery.LoadJobConfig(
    source_format=bigquery.SourceFormat.PARQUET,
    write_disposition=bigquery.WriteDisposition.WRITE_TRUNCATE
)

with open("temp_data.parquet", "rb") as source_file:
    job = client.load_table_from_file(
        source_file,
        "your_project.your_dataset.your_table",
        job_config=job_config
    )

job.result()  # 等待上传完成

5. 检查其他字段的隐藏问题 #

排查name或type字段是否包含逗号、换行符等CSV分隔符：

row = df.iloc[855402]
print("Name content:", repr(row['name']))
print("Type content:", repr(row['type']))

若发现问题字符，直接替换清理：

df['name'] = df['name'].apply(lambda x: x.replace(',', '').replace('\n', ''))
df['type'] = df['type'].apply(lambda x: x.replace(',', '').replace('\n', ''))

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