使用Pandas的to_sql插入数据时，如何仅跳过损坏条目？

核心思路 #

先通过预处理过滤明显不兼容的行，再采用「大分块→小批量→逐行试错」的分层插入策略，既保证插入效率，又能精准跳过损坏条目并记录错误信息。

步骤与代码实现 #

1. 依赖导入与基础配置

import pandas as pd
from sqlalchemy import create_engine
import psycopg2
from psycopg2 import sql
import logging

# 配置错误日志，记录跳过的行信息
logging.basicConfig(
    filename='insert_errors.log',
    level=logging.ERROR,
    format='%(asctime)s - %(message)s'
)
logger = logging.getLogger(__name__)

# 初始化数据库连接（替换为你的数据库信息）
engine = create_engine('postgresql://user:password@host:port/dbname')

2. 分块读取+预处理过滤

先对CSV分块读取，针对目标表的数值型列做类型校验，提前过滤转换失败的行：

# 配置分块大小，可根据内存调整
chunk_size = 1000
# 目标表中需要double precision类型的列（替换为你的实际列名）
numeric_cols = ['price', 'quantity']

for chunk in pd.read_csv('data.csv', chunksize=chunk_size):
    # 对指定列做数值转换，无法转换的标记为NaN
    for col in numeric_cols:
        chunk[col] = pd.to_numeric(chunk[col], errors='coerce')
    
    # 过滤掉包含NaN的行（这些行必然无法插入）
    valid_chunk = chunk.dropna(subset=numeric_cols)
    if valid_chunk.empty:
        continue

3. 小批量插入+逐行容错

将预处理后的块拆成更小的批次，优先批量插入；若批次失败，再逐行尝试插入，捕获类型错误并跳过：

# 小批量大小，平衡效率与容错
    small_batch_size = 100
    for start in range(0, len(valid_chunk), small_batch_size):
        batch = valid_chunk.iloc[start:start+small_batch_size]
        try:
            # 尝试批量插入
            batch.to_sql(
                name='target_table',
                con=engine,
                if_exists='append',
                index=False,
                method='multi'  # 启用批量插入优化
            )
        except Exception:
            # 批量失败，逐行处理
            for idx, row in batch.iterrows():
                try:
                    with engine.connect() as conn:
                        # 构造原生插入语句，适配列名与值
                        cols = sql.SQL(', ').join(map(sql.Identifier, row.index))
                        vals = sql.SQL(', ').join(sql.Placeholder() * len(row))
                        insert_stmt = sql.SQL("INSERT INTO target_table ({cols}) VALUES ({vals})").format(cols=cols, vals=vals)
                        conn.execute(insert_stmt, tuple(row.values))
                        conn.commit()
                except psycopg2.errors.InvalidTextRepresentation as e:
                    # 捕获类型不兼容错误，记录并跳过
                    logger.error(f"跳过行{idx}：{str(e)}，数据：{row.to_dict()}")
                    continue
                except Exception as e:
                    # 其他未知错误，同样记录
                    logger.error(f"行{idx}插入失败：{str(e)}，数据：{row.to_dict()}")
                    continue

关键注意事项 #

• 提前匹配表结构：必须明确目标表的列类型，针对性做预处理，避免无效试错。
• 日志排查：insert_errors.log会记录所有跳过的行及错误原因，方便后续修复数据。
• 参数调优：chunk_size和small_batch_size需根据内存大小、数据库性能调整，避免内存溢出或连接过载。
• 性能平衡：逐行插入效率较低，仅在小批量失败时启用，优先保证大部分数据的批量插入效率。

内容的提问来源于stack exchange，提问作者Bruno Henrique Peixoto