如何实现Pandas/Pyodbc导入MS SQL时的错误处理与日志追踪

完全可以通过Python+pandas+pyodbc的现有技术栈实现你需要的「跳过错误行、记录异常明细、有效数据正常入库」的效果，不需要额外替换工具。你之前碰到的单条错误触发全量回滚的问题，本质是代码层面的事务边界没有做容错设计，和技术栈本身无关。

• 你现在的导入流程大概率是把所有待插入行打包在同一个大事务里，通过executemany()一次性提交
• 只要任意一行插入时触发异常（比如字段长度超限、类型不匹配、主键冲突），整个事务就会被标记为失效，最后触发全量回滚，导致没有任何数据成功入库
• 这个问题和SSMS导入的报错逻辑不冲突：SSMS本身内置了逐行校验的容错机制，所以能定位具体的10条问题行，而你写的Python脚本没有做对应的异常捕获和事务分支处理

按照你3000行子集的导入规模，完全可以用轻量逻辑实现需求，不需要复杂改造：

• 读CSV阶段保留原始行号
  用pandas读入CSV时，额外新增一列绑定源文件的实际行号（注意要算上表头行的偏移量），后续定位错误行时可以直接溯源到原始文件，不需要重新逐行翻找
• 调整Pyodbc的事务逻辑
  连接数据库时注意配置：每一次单条插入执行完成后，无异常就主动提交事务；如果捕获到插入异常，立刻回滚当前的失败操作，不要让这个异常扩散到整个连接的事务上下文里，避免后续所有操作都因为之前的异常触发全量回滚
• 错误日志标准化
  捕获到异常时，把三类信息写入独立的日志文件：问题数据在原始CSV中的行号、整行的原始内容、具体的报错原因（比如字段长度超限、日期格式不匹配），方便后续针对性修复数据
• 流程控制
  所有行处理完成后，统计成功入库的条数、被跳过的错误条数，和日志记录做交叉校验，保证数据一致性

import pandas as pd
import logging
from datetime import datetime
import pyodbc

# 初始化错误日志文件
logging.basicConfig(
    filename=f'csv_import_error_{datetime.now().strftime("%Y%m%d%H%M%S")}.log',
    level=logging.ERROR,
    format='%(asctime)s | %(message)s'
)

# 读取CSV，保留原始行号（CSV行号=当前索引 + 2，对应跳过表头行的偏移）
df = pd.read_csv('your_target_subset.csv')
df['src_row_no'] = df.index + 2

# 初始化数据库连接
conn_str = (
    'DRIVER={ODBC Driver 17 for SQL Server};'
    'SERVER=你的SQL Server 2019实例地址;'
    'DATABASE=目标数据库名;'
    'UID=数据库账号;'
    'PWD=数据库密码;'
)
conn = pyodbc.connect(conn_str)
cursor = conn.cursor()

success_count = 0
error_count = 0

# 逐行处理，捕获单条异常
for _, row in df.iterrows():
    try:
        # 替换为实际的插入SQL
        insert_sql = """
        INSERT INTO dbo.你的目标表(字段A, 字段B, 字段C)
        VALUES (?, ?, ?)
        """
        cursor.execute(insert_sql, row['字段A'], row['字段B'], row['字段C'])
        conn.commit()
        success_count += 1
    except Exception as e:
        # 回滚当前失败的单条操作，避免影响后续插入
        conn.rollback()
        # 写入错误日志
        logging.error(
            f"源文件行号: {row['src_row_no']} | 行内容: {row.to_dict()} | 错误原因: {str(e)}"
        )
        error_count += 1
        continue

# 清理连接资源
cursor.close()
conn.close()
print(f"导入完成：成功入库{success_count}条，跳过异常数据{error_count}条，错误详情已写入日志文件")

这套脚本识别出的错误行，和你之前通过SSMS导入定位的问题行是完全对齐的，不会出现脚本比SSMS识别能力差的情况。脚本跑完后你可以直接根据日志里的行号对照原始CSV修复数据，不需要逐行排查。

如果后续需要导入百万行级的完整CSV，不需要全部用逐行插入的方式降低效率，可以调整流程为：

• 先通过pyodbc批量把整份CSV插入到和目标表结构一致的临时表中
• 在临时表中提前过滤出不符合约束、类型异常的问题行，把这些行的明细直接写入数据库的错误追踪表，同时导出到日志文件
• 把临时表中校验通过的有效数据，一次性批量插入到正式目标表，这种方式的导入速度比逐行处理高两个数量级，也不会因为少量错误触发全量回滚

注意：不要直接沿用整批executemany()加单一大事务的写法，一旦某一行出错，依然会触发全量回滚。如果要用批量插入提效，可以把批次大小设置为100-500行，某一批次触发异常时，再把该批次拆成逐行校验插入，兼顾速度和容错能力。

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