使用pandas_gbq同步MySQL至BigQuery遇Arrow转换错误求助

问题分析 #

你的代码在执行pandas_gbq.to_gbq()时触发pyarrow.lib.ArrowNotImplementedError: NumPyConverter doesn't implement <list<item: int64>> conversion错误，核心原因是：DataFrame中存在表面标注为object类型，但实际存储了Python列表（list<int64>）的列。虽然df.info()显示所有列都是常规类型，但object类型可以容纳任意Python对象，包括列表，而pyarrow无法直接将这种列表类型转换为BigQuery支持的格式。

另外你提供的table_schema存在两个问题：

1. 列名存在多余空格（比如" Version"、"code_of_country "），和DataFrame的列名不匹配
2. 部分列类型定义错误（比如Wheel原是float64却定义为STRING，month原是int64却定义为STRING）

解决方案步骤 #

1. 定位包含列表的列 #

先遍历所有object类型列，找出实际存储列表的列：

# 在读取DataFrame后添加这段代码
list_cols = []
for col in df.select_dtypes(include=['object']).columns:
    # 抽样检查列中是否有列表类型值
    if any(isinstance(x, list) for x in df[col].sample(100)):
        list_cols.append(col)
        print(f"检测到列表类型列: {col}")

2. 处理列表列 #

根据业务需求选择以下一种方式处理：

方式一：将列表转为字符串（最简单，适合不需要结构化查询的场景）

for col in list_cols:
    df[col] = df[col].apply(lambda x: ','.join(map(str, x)) if isinstance(x, list) else x)

方式二：转为BigQuery支持的数组类型（保留结构化信息）

如果需要在BigQuery中以数组类型存储，需将列值统一为列表，并修改schema对应类型为ARRAY<int64>：

# 统一列值为列表，空值转为空列表
for col in list_cols:
    df[col] = df[col].apply(lambda x: x if isinstance(x, list) else [x] if pd.notna(x) else [])

# 同时在table_schema中对应列的type改为"ARRAY<int64>"，mode改为"REPEATED"

方式三：直接删除不需要的列表列

如果该列无业务价值，可直接删除：

df = df.drop(columns=list_cols)

3. 修正table_schema #

确保列名和DataFrame完全一致，类型匹配实际数据：

table_schema=[
    {"name": "SKU", "type": "STRING", "mode": "NULLABLE"},
    {"name": "text", "type": "STRING", "mode": "NULLABLE"},
    {"name": "document_date1", "type": "DATE", "mode": "NULLABLE"},
    {"name": "Sum_quantity1", "type": "FLOAT64", "mode": "NULLABLE"},
    {"name": "Landon", "type": "STRING", "mode": "NULLABLE"},
    {"name": "Category", "type": "STRING", "mode": "NULLABLE"},
    {"name": "Wheel", "type": "FLOAT64", "mode": "NULLABLE"},
    {"name": "Frame", "type": "STRING", "mode": "NULLABLE"},
    {"name": "Version", "type": "STRING", "mode": "NULLABLE"},
    {"name": "Color", "type": "STRING", "mode": "NULLABLE"},
    {"name": "YYYY_MM", "type": "STRING", "mode": "NULLABLE"},
    {"name": "month", "type": "INT64", "mode": "NULLABLE"},
    {"name": "code_of_country", "type": "STRING", "mode": "NULLABLE"},
]

完整修改后代码 #

import pandas as pd
from sqlalchemy import create_engine
import pandas_gbq

def query_mysql(request):
    engine = create_engine(
        "mysql+pymysql://bd:pass@host/user"
    )
    df = pd.read_sql("SELECT * FROM table", engine)

    # 检测并处理列表类型列
    list_cols = []
    for col in df.select_dtypes(include=['object']).columns:
        if any(isinstance(x, list) for x in df[col].sample(100)):
            list_cols.append(col)
            print(f"处理列表列: {col}")
            # 这里用转字符串的方式，可根据需求替换为其他处理逻辑
            df[col] = df[col].apply(lambda x: ','.join(map(str, x)) if isinstance(x, list) else x)
    
    # 修正后的表结构
    table_schema=[
        {"name": "SKU", "type": "STRING", "mode": "NULLABLE"},
        {"name": "text", "type": "STRING", "mode": "NULLABLE"},
        {"name": "document_date1", "type": "DATE", "mode": "NULLABLE"},
        {"name": "Sum_quantity1", "type": "FLOAT64", "mode": "NULLABLE"},
        {"name": "Landon", "type": "STRING", "mode": "NULLABLE"},
        {"name": "Category", "type": "STRING", "mode": "NULLABLE"},
        {"name": "Wheel", "type": "FLOAT64", "mode": "NULLABLE"},
        {"name": "Frame", "type": "STRING", "mode": "NULLABLE"},
        {"name": "Version", "type": "STRING", "mode": "NULLABLE"},
        {"name": "Color", "type": "STRING", "mode": "NULLABLE"},
        {"name": "YYYY_MM", "type": "STRING", "mode": "NULLABLE"},
        {"name": "month", "type": "INT64", "mode": "NULLABLE"},
        {"name": "code_of_country", "type": "STRING", "mode": "NULLABLE"},
    ]

    pandas_gbq.to_gbq(
        df,
        destination,
        project_id,
        if_exists="replace",
        table_schema=table_schema
    )

内容的提问来源于stack exchange，提问作者Simon Breton