使用SQLAlchemy与pandas插入SQL Server失败，如何定位违规数据？

一、批量拆分逐步排查 #

把大DataFrame拆成小批量插入，报错时快速锁定问题批次，再缩小范围定位单条记录：

import numpy as np
# 拆分为10个批次（可根据数据量调整）
batches = np.array_split(df, 10)
for batch_idx, batch in enumerate(batches, 1):
    try:
        batch.to_sql(
            name="your_target_table",
            con=engine,
            if_exists="append",
            index=False
        )
    except Exception as e:
        print(f"批次 {batch_idx} 插入失败：{str(e).split('\n')[0]}")
        # 保存失败批次以便后续分析
        batch.to_csv(f"failed_batch_{batch_idx}.csv", index=False)
        raise

二、提前校验字符串长度（针对截断错误） #

既然报错是字符串长度超过字段限制，直接提前校验DataFrame中每一列的最大长度，和数据库表字段的允许长度对比：

1. 先获取目标表的字段长度信息：

from sqlalchemy import inspect
inspector = inspect(engine)
# 获取表的列结构
table_columns = inspector.get_columns("your_target_table")
# 筛选出字符串类型的列及其最大长度
str_col_limits = {
    col["name"]: col["type"].length
    for col in table_columns
    if hasattr(col["type"], "length")
}

2. 检查DataFrame中对应列的超长记录：

for col_name, max_len in str_col_limits.items():
    if col_name not in df.columns:
        continue
    # 计算每条记录的字符串长度（注意处理空值）
    col_lengths = df[col_name].astype(str).str.len()
    # 筛选出超长记录
    invalid_rows = df[col_lengths > max_len]
    if not invalid_rows.empty:
        print(f"列「{col_name}」存在 {len(invalid_rows)} 条超长记录：")
        print(invalid_rows)
        invalid_rows.to_csv(f"invalid_records_{col_name}.csv", index=False)

三、自定义错误捕获，屏蔽冗长SQL #

捕获错误时只提取核心信息，不打印完整的插入语句：

import pyodbc

try:
    df.to_sql(
        name="your_target_table",
        con=engine,
        if_exists="append",
        index=False
    )
except pyodbc.ProgrammingError as e:
    # 只保留错误的核心提示，截断后续的超长SQL
    core_error = str(e).split("\n")[0]
    print(f"插入失败：{core_error}")
    # 结合前面的方法定位具体记录
    raise

四、逐条/小批量映射插入（精准定位单条记录） #

把DataFrame转为字典列表，逐个插入并捕获单条错误，适合数据量不大的场景：

from sqlalchemy import Table, MetaData

metadata = MetaData()
target_table = Table("your_target_table", metadata, autoload_with=engine)
records = df.to_dict("records")

for idx, record in enumerate(records, 1):
    try:
        with engine.begin() as conn:
            conn.execute(target_table.insert().values(record))
    except Exception as e:
        print(f"第 {idx} 条记录插入失败：{str(e).split('\n')[0]}")
        print(f"违规记录内容：{record}")
        raise

内容的提问来源于stack exchange，提问作者Travis Thompson