Python操作MySQL：批量插入CSV数据时如何追踪失败行并处理异常

1. 安装依赖 #

先安装所需的Python库：

pip install pandas sqlalchemy mysql-connector-python

2. 配置日志系统 #

设置日志规则，确保能记录错误行的原始行号、数据内容、异常详情，方便后续排查：

import logging

# 配置日志文件与格式
logging.basicConfig(
    filename='insert_errors.log',
    level=logging.ERROR,
    format='%(asctime)s - %(levelname)s - 原始行号: %(lineno)s - 详情: %(message)s'
)

3. 读取并预处理CSV #

用pandas读取CSV，完成列重命名、删除冗余列的操作，注意保留原始CSV的行号映射：

import pandas as pd

# 读取目标CSV
df = pd.read_csv('source_data.csv')

# 列重命名：按实际需求替换原列名和新列名
df.rename(columns={'old_column_a': 'db_column_a', 'old_column_b': 'db_column_b'}, inplace=True)

# 删除不需要的列：传入要移除的列名列表
df.drop(columns=['unused_col1', 'unused_col2'], inplace=True)

4. 数据库连接与插入（带错误追踪） #

采用逐行插入+异常捕获的方式精准定位错误行；如果数据量极大，可改成小批次插入，批次失败时再逐行排查该批次，平衡效率与错误定位精度：

from sqlalchemy import create_engine
from sqlalchemy.exc import SQLAlchemyError

# 创建MySQL连接引擎：替换为你的数据库账号、地址和库名
engine = create_engine('mysql+mysqlconnector://user:password@localhost:3306/your_db')

# 获取数据库连接
conn = engine.connect()

try:
    # 遍历DataFrame每一行
    for idx, row in df.iterrows():
        # 映射原始CSV行号：DataFrame索引从0开始，CSV数据行从第2行（表头是第1行）开始，所以+2
        csv_row_num = idx + 2
        try:
            # 单行插入数据库：替换为你的目标表名和字段
            conn.execute(
                "INSERT INTO target_table (db_column_a, db_column_b) VALUES (:db_column_a, :db_column_b)",
                row.to_dict()
            )
        except SQLAlchemyError as e:
            # 记录错误日志：包含行号、数据内容、异常原因
            error_detail = f"数据内容: {row.to_dict()}，异常原因: {str(e)}"
            logging.error(error_detail, extra={'lineno': csv_row_num})
            # 跳过错误行，继续执行后续插入
            continue
    # 提交事务
    conn.commit()
except Exception as e:
    logging.error(f"全局异常: {str(e)}")
    conn.rollback()
finally:
    # 关闭数据库连接
    conn.close()

关键说明 #

• 行号映射：用idx+2对应原始CSV的行号，能直接在源文件中定位错误数据。
• 日志信息：日志包含时间、错误级别、原始行号、错误数据和异常原因，足以排查常见问题（比如数据类型不匹配、主键重复、字段为空等）。
• 批量优化：若数据量超过10万行，可将DataFrame按每1000行拆分批次，先尝试批量插入，当批次抛出异常时，再逐行遍历该批次定位错误行，兼顾插入效率和错误排查精度。

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