使用pyodbc批量插入含空值的Pandas DataFrame到SQL Server时多记录批次失败的问题排查与解决求助

大家好，我现在遇到一个头疼的问题：用pyodbc把大体积的Pandas DataFrame批量插入SQL Server时，只要批次大小大于1就会出现部分批次失败的情况，但单条插入完全正常。我已经定位到问题出在不同类型列的空值处理上，但不知道具体该怎么调整，想请教下各位大佬！

问题背景 #

我的场景是数据量太大必须分批插入（比如batch_size=3），单条插入（batch_size=1）时所有数据都能成功写入，但只要批次里有2条以上数据，部分批次就会报错。排查后发现，报错批次里必然包含带空值的记录：

• 字符串列的None（比如entity_name、entity_address里的空值）
• 浮点类型列的np.nan（比如entity_height、entity_weight）
• 从float转成支持空值的Int64类型列的<NA>（比如entity_age_code，原数据是带np.nan的float，需要转成整数类型的空值）

我尝试把所有np.nan转成None后再转成列表传给executemany，但还是解决不了多批次失败的问题。

可重现的测试数据与环境 #

下面是能复现问题的完整测试代码，包含测试数据集、类型转换逻辑和插入逻辑：

import pandas as pd
import numpy as np
import pyodbc
import sys
import logging

# 初始化日志
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

# 自定义类型转换函数：处理支持空值的类型转换（比如Int64）
def change_dtype(col, dtype):
    try:
        return col.astype(dtype)
    except ValueError:
        # 处理空值转类型的情况
        return col.apply(lambda x: None if pd.isna(x) else x).astype(dtype)

# 测试用数据集
data = {
    "unique_id": [
        "String_3", "String_5", "String_10", "String_9", "String_4",
        "String_7", "String_2", "String_6", "String_1", "String_8"
    ],
    "entity_name": [
        "Alice", None, "Eve", "Alice", "Bob", "Alice", "Alice", None,
        "Charlie", "Eve"
    ],
    "entity_address": [
        "456 Elm St", "789 Oak St", "123 Main St", "456 Elm St", None,
        "456 Elm St", "123 Main St", None, "789 Oak St", "789 Oak St"
    ],
    "entity_age_code": [
        763.0, 349.0, np.nan, np.nan, 888.0, 999.0, 711.0, 574.0, 963.0, 300.0
    ],
    "entity_height": [
        93.357616, 48.408745, 79.978718, 94.953377, 11.094891, np.nan,
        33.282917, np.nan, 82.714043, np.nan
    ],
    "entity_weight": [
        19.158688, np.nan, 73.853124, 54.005774, 70.846664, 70.996657,
        np.nan, 26.325328, 11.360588, 51.324372
    ]
}

df = pd.DataFrame(data)
# 把float类型的entity_age_code转成支持空值的Int64类型
df['entity_age_code'] = change_dtype(df['entity_age_code'], 'Int64')

出错的批量插入核心代码 #

下面是插入SQL Server的核心逻辑，重点是我处理空值的部分和fast_executemany的设置：

# 数据库连接配置
entity_key = ['unique_id']
entity_table_name = 'DscEntityTable'
batch_size = 3
conn_str = "xxxxxxx"  # 替换为你的实际连接字符串
conn = pyodbc.connect(conn_str)
cursor = conn.cursor()

# 推断SQL Server数据类型的函数
def infer_sql_server_dtype(df):
    dtype_mapping = {
        'object': 'VARCHAR(255)',
        'Int64': 'INT',
        'float64': 'FLOAT',
        # 可根据实际数据类型补充更多映射
    }
    dtype = {}
    for col in df.columns:
        col_dtype = str(df[col].dtype)
        dtype[col] = dtype_mapping.get(col_dtype, 'VARCHAR(255)')
    return dtype

# 检查目标表是否存在
cursor.execute(f"SELECT OBJECT_ID('{entity_table_name}', 'U')")
table_exists = cursor.fetchone()[0] is not None

if table_exists:
    # 过滤已存在的主键记录
    key_tuples = list(df[entity_key].dropna().itertuples(index=False, name=None))
    
    # 分批获取数据库中已有的主键（简化实现，实际可优化分批逻辑）
    def get_existing_keys_in_batches(cursor, table_name, keys, key_list):
        placeholders = ", ".join(["?" for _ in keys])
        query = f"SELECT {', '.join(keys)} FROM {table_name} WHERE ({', '.join(keys)}) IN ({', '.join([f'({placeholders})' for _ in key_list])})"
        cursor.execute(query, [item for tup in key_list for item in tup])
        return cursor.fetchall()
    
    existing_keys = get_existing_keys_in_batches(cursor, entity_table_name, entity_key, key_tuples)
    existing_keys = pd.DataFrame(existing_keys, columns=entity_key)
    
    # 对齐主键列的数据类型
    for col in entity_key:
        dtype_str = str(df[col].dtype)
        existing_keys[col] = change_dtype(existing_keys[col], dtype_str)
    
    # 过滤掉已存在的记录
    df = pd.merge(df, existing_keys, how='outer', indicator=True)
    df = df[df['_merge'] == 'left_only'].drop(columns='_merge')
    
    if df.empty:
        logger.info("No new records to insert.")
        sys.exit("No new records to insert.")
else:
    # 创建新表
    dtype = infer_sql_server_dtype(df)
    columns_sql = ", ".join([
        f"[{col}] {dtype[col]} COLLATE Latin1_General_100_CI_AS_SC_UTF8"
        if dtype[col].startswith("VARCHAR") else f"[{col}] {dtype[col]}"
        for col in df.columns
    ])
    pk_sql = ", ".join(f"[{col}]" for col in entity_key)
    create_sql = f"CREATE TABLE {entity_table_name} ({columns_sql}, PRIMARY KEY ({pk_sql}))"
    logger.debug(f"Create table query: {create_sql}")
    cursor.execute(create_sql)
    conn.commit()
    logger.info(f"Table {entity_table_name} created successfully.")

# 构建插入SQL
columns = list(df.columns)
placeholders = ", ".join(["?"] * len(columns))
insert_sql = f"INSERT INTO {entity_table_name} ({', '.join(columns)}) VALUES ({placeholders})"
logger.debug(f"Insert query template: {insert_sql}")

written_count = 0

# 分批插入逻辑
for batch_num, start in enumerate(range(0, len(df), batch_size)):
    batch = df.iloc[start:start + batch_size]
    cursor.fast_executemany = True
    
    # 我当前的空值处理：把所有pd识别的空值转成None
    values = batch.map(lambda x: None if pd.isna(x) else x).values.tolist()
    logger.debug(f"Batch {batch_num} sample values: {values[:2]}...")
    
    try:
        cursor.executemany(insert_sql, values)
        conn.commit()
        logger.info(f"Successfully wrote batch {batch_num} (rows {start} to {start + len(batch) - 1})")
        written_count += len(batch)
    except Exception as e:
        logger.error(f"Failed to write batch {batch_num}: {str(e)}")
        continue  # 跳过失败批次，继续下一批

logger.info(f"Insert complete! Total written records: {written_count}")
cursor.close()
conn.close()

我尝试过的方案（无效） #

1. 把所有np.nan和<NA>统一转成None：用batch.map(lambda x: None if pd.isna(x) else x)处理后转成列表，但多批次还是失败
2. 调整fast_executemany为False：虽然能解决部分批次失败问题，但插入速度慢到无法接受
3. 单独处理Int64列的空值：把<NA>转成None后再转成int类型，但会丢失空值信息或者报错

想请教的问题 #

1. 为什么单条插入没问题，多批次就失败？是fast_executemany对空值的处理有特殊要求吗？
2. 针对不同类型的空值（字符串列的None、Int64列的<NA>、float列的np.nan），应该怎么统一处理才能让executemany+fast_executemany=True正常工作？
3. 有没有更优雅的方式处理Pandas DataFrame到SQL Server的分批插入空值问题？

谢谢大家！

内容来源于stack exchange