使用psycopg2批量插入PostgreSQL时如何兼顾性能与错误处理？

我正在用Python + psycopg2构建PostgreSQL到PostgreSQL的ETL脚本，原程序采用单条循环插入的方式，对于1000万条数据来说速度实在太慢：

import psycopg2
source_cursor = source_database_connection.conn.cursor()
destination_cursor = destination.conn.cursor()
insert_query = "INSERT INTO mytable (col1, ..., coln) VALUES (%s, ..., %s) ON CONFLICT (key) DO UPDATE SET col1 = value1, ... ;"
for row in source_cursor:
    data = some_business_logic(row)
    try:
        destination_cursor.execute(insert_query, data)
    except:
        # 跳过该行并记录错误

为了提速，我尝试用execute_values()做批量插入，但遇到了新问题：数据错误较多，一旦批次里有一条坏数据，整批都得丢弃；如果在异常里逐条重试，最坏情况每条数据执行两次，性能又回到原点。请问有没有既能加速INSERT，又能优雅处理错误的方法？

这简直是ETL开发中最头疼的两难问题之一——要速度就得批量插，但批量插遇到一条坏数据就得丢整批，退回到单条插又慢到离谱。别慌，我给你几个实用的方案，既能保住批量插入的速度，又能优雅处理那些捣蛋的错误数据：

方案1：分层批量降级处理（首推） #

核心思路是用"大批次优先，出错就拆分"的策略：先按大批次（比如10000条）执行，一旦出错，把这个大批次拆成更小的批次（比如1000条）重试；如果小批次还是出错，再拆成更小的（比如100条），直到最后拆成单条来定位错误数据。这样大部分正确的数据还是走批量，只有包含错误的批次才会逐步降级，完美平衡性能和错误处理。

代码示例：

def batch_insert(cursor, insert_query, data_batch, batch_sizes=[10000, 1000, 100, 1]):
    if not data_batch:
        return
    current_size = batch_sizes[0]
    try:
        if current_size == 1:
            # 单条执行，定位错误数据
            cursor.execute(insert_query, data_batch[0])
            print(f"成功插入单条数据: {data_batch[0]}")
        else:
            # 批量执行，保证速度
            cursor.execute_values(insert_query, data_batch)
            print(f"成功插入{len(data_batch)}条数据")
    except Exception as e:
        print(f"批次大小{current_size}执行失败: {str(e)}")
        if len(batch_sizes) == 1:
            # 已经到最小批次，记录错误并跳过
            print(f"跳过错误数据: {data_batch[0]}")
            return
        # 拆分成更小的批次递归处理
        step = batch_sizes[1]
        for i in range(0, len(data_batch), step):
            sub_batch = data_batch[i:i+step]
            batch_insert(cursor, insert_query, sub_batch, batch_sizes[1:])

# 使用示例
data_buffer = []
base_batch_size = 10000
for idx, row in enumerate(source_cursor, 1):
    processed_data = tuple(some_business_logic(row))
    data_buffer.append(processed_data)
    
    # 达到批次大小就执行
    if idx % base_batch_size == 0:
        batch_insert(destination_cursor, insert_query, data_buffer)
        data_buffer = []

# 处理最后一批剩余数据
if data_buffer:
    batch_insert(destination_cursor, insert_query, data_buffer)

# 最后统一提交
destination.conn.commit()

方案2：结合Savepoint做部分回滚 #

如果不想拆分批次，可以在批量执行前创建一个保存点（Savepoint）：批量执行失败时，回滚到这个保存点，然后对批次内的数据逐条执行，跳过错误数据。这样既不会因为一条错误数据丢失整个批次的正确数据，也只有出错的批次需要逐条处理。

代码示例：

data_buffer = []
base_batch_size = 10000
# 关闭自动提交，减少IO开销
destination.conn.autocommit = False

for idx, row in enumerate(source_cursor, 1):
    processed_data = tuple(some_business_logic(row))
    data_buffer.append(processed_data)
    
    if idx % base_batch_size == 0:
        try:
            # 先尝试批量插入
            destination_cursor.execute_values(insert_query, data_buffer)
            print(f"批量插入{base_batch_size}条数据成功")
        except Exception as e:
            print(f"批量插入失败，开始逐条重试: {str(e)}")
            # 创建保存点，用于隔离错误数据的回滚
            destination_cursor.execute("SAVEPOINT batch_savepoint")
            for item in data_buffer:
                try:
                    destination_cursor.execute(insert_query, item)
                except Exception as item_e:
                    print(f"跳过错误数据{item}: {str(item_e)}")
                    # 回滚到保存点，不影响之前的正确插入
                    destination_cursor.execute("ROLLBACK TO SAVEPOINT batch_savepoint")
            # 释放保存点
            destination_cursor.execute("RELEASE SAVEPOINT batch_savepoint")
        data_buffer = []

# 处理剩余数据
if data_buffer:
    try:
        destination_cursor.execute_values(insert_query, data_buffer)
    except Exception as e:
        print(f"剩余数据批量插入失败，逐条重试: {str(e)}")
        destination_cursor.execute("SAVEPOINT final_batch_savepoint")
        for item in data_buffer:
            try:
                destination_cursor.execute(insert_query, item)
            except Exception as item_e:
                print(f"跳过错误数据{item}: {str(item_e)}")
                destination_cursor.execute("ROLLBACK TO SAVEPOINT final_batch_savepoint")
        destination_cursor.execute("RELEASE SAVEPOINT final_batch_savepoint")

# 统一提交所有变更
destination.conn.commit()

方案3：提前做数据校验（从根源减少错误） #

如果错误是因为数据格式、外键约束、字段长度等可预见的问题，在业务逻辑层提前过滤掉不合格的数据，能从根源上减少批量执行的错误概率，甚至可以做到零错误批量插入。

比如在外键校验时，可以缓存已存在的外键值，避免频繁查询目标数据库：

# 缓存已存在的外键，减少重复查询
valid_foreign_keys = set()

def check_foreign_key_exists(cursor, foreign_key):
    if foreign_key in valid_foreign_keys:
        return True
    # 查询目标数据库是否存在该外键
    cursor.execute("SELECT 1 FROM mytable2 WHERE id = %s", (foreign_key,))
    exists = cursor.fetchone() is not None
    if exists:
        valid_foreign_keys.add(foreign_key)
    return exists

def some_business_logic(row, dest_cursor):
    col1, col2, col3, foreign_key = row
    # 校验外键是否存在
    if not check_foreign_key_exists(dest_cursor, foreign_key):
        print(f"错误：外键{foreign_key}不存在，跳过该行")
        return None
    # 校验字段长度
    if len(col2) > 50:
        print(f"错误：col2值{col2}过长，跳过该行")
        return None
    # 其他格式校验...
    return (col1, col2, col3, foreign_key)

# 收集数据时过滤掉不合格的数据
data_buffer = []
for row in source_cursor:
    processed_data = some_business_logic(row, destination_cursor)
    if processed_data is not None:
        data_buffer.append(tuple(processed_data))
        if len(data_buffer) % 10000 == 0:
            destination_cursor.execute_values(insert_query, data_buffer)
            data_buffer = []

• 增大fetch_size：从源数据库读取数据时，设置source_cursor.itersize = 10000（默认是2000），减少和源数据库的交互次数。
• 用execute_batch替代execute_values：如果你的INSERT语句有复杂的ON CONFLICT逻辑，execute_batch会把批量数据拆分成多个单条INSERT但一次性发送，性能接近execute_values，兼容性更好。
• 批量提交：一定要关闭自动提交，最后统一commit()，避免每批都提交带来的IO开销。

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