Postgres批量数据迁移事务失败，如何优化单条处理效率？

一、先定位批量失败的具体记录，避免全批次逐条处理 #

批量插入失败往往是个别记录违反约束（主键重复、字段非空、外键不存在等）导致的，优先筛选出异常记录，剩余正常记录继续批量处理，能大幅减少逐条执行的次数：

1. 预校验异常记录
   针对每张目标表的约束，提前查询当前批次中不符合要求的记录，直接标记为错误，剩下的记录正常执行批量迁移：

-- 示例：检查stage2.A表的主键冲突、col1非空约束
SELECT id 
FROM stage1.dump
WHERE is_processed = False AND batch_id = 1
AND (
    id IN (SELECT id FROM stage2.A)  -- 主键冲突
    OR col1 IS NULL  -- 非空约束
);

将查询出的异常ID插入stage2.errors，再用剩余记录执行原批量插入逻辑，大部分记录仍能保持批量处理的效率。

2. 用ON CONFLICT捕获单表异常，避免批次回滚
   利用PostgreSQL的ON CONFLICT语法，让批量插入跳过有问题的记录，无需回滚整个批次：

INSERT INTO stage2.A (col1, col2)
SELECT col1, col2 FROM stage1.dump
WHERE is_processed = False AND batch_id = 1
ON CONFLICT (id) DO NOTHING;  -- 主键冲突则跳过

-- 事后记录失败的ID
INSERT INTO stage2.errors (failed_id)
SELECT id FROM stage1.dump
WHERE is_processed = False AND batch_id = 1
AND id NOT IN (SELECT id FROM stage2.A);

这种方式能让批量操作继续完成，只跳过异常记录，避免全批次回滚后逐条处理的低效。

二、优化失败批次的逐条处理逻辑 #

如果必须处理失败批次，也可通过合并操作减少数据库交互次数：

1. 单条记录的多表迁移合并为单次事务
   在Python中，将单条记录插入多张表的操作放在同一个事务中执行，而非分10次单独执行：

import psycopg2

conn = psycopg2.connect("dbname=your_db user=your_user")
cur = conn.cursor()

# 获取失败批次的所有记录
cur.execute("SELECT id, col1, col2, col3, col4 FROM stage1.dump WHERE batch_id = 1 AND is_processed = False")
failed_records = cur.fetchall()

for record in failed_records:
    try:
        conn.autocommit = False
        # 批量执行该记录的所有表插入
        cur.execute("INSERT INTO stage2.A (col1, col2) VALUES (%s, %s)", (record[1], record[2]))
        cur.execute("INSERT INTO stage2.B (col3, col4) VALUES (%s, %s)", (record[3], record[4]))
        # ... 其他8张表的插入语句
        conn.commit()
        # 标记该记录为已处理
        cur.execute("UPDATE stage1.dump SET is_processed = True WHERE id = %s", (record[0],))
        conn.commit()
    except Exception as e:
        conn.rollback()
        cur.execute("INSERT INTO stage2.errors (failed_id) VALUES (%s)", (record[0],))
        conn.commit()

这样每条记录只需要一次事务往返，而非10次独立查询。

2. 失败批次拆分为更小的子批次重试
   将失败的5000条记录拆分为更小的子批次（比如500条一组），尝试再次批量迁移，只有子批次失败时再逐条处理，减少逐条执行的总次数。

三、用COPY命令提升单条/小批量记录的导入效率 #

PostgreSQL的COPY命令比普通INSERT效率高数倍，即使处理单条或小批量记录，也能大幅提升速度：

import csv
from io import StringIO

# 处理单条记录生成CSV流
record = (1, "val1", "val2")
csv_data = StringIO()
writer = csv.writer(csv_data)
writer.writerow([record[1], record[2]])
csv_data.seek(0)

# 用COPY导入stage2.A
cur.copy_expert("COPY stage2.A (col1, col2) FROM STDIN WITH CSV", csv_data)

四、改进批次划分与状态标记逻辑 #

建议改用id范围划分批次（比如每次取is_processed = False且id > last_processed_id的前5000条），而非固定batch_id，这样即使批次失败，也能精准追踪未处理记录。同时，批量迁移成功后，批量标记is_processed = True，避免逐条更新的开销。

内容的提问来源于stack exchange，提问作者Not yet decided