PostgreSQL批量导入如何跳过错误行并保留有效数据？

嘿，太懂你这种批量插入时被个别外键违规坑得全量回滚的糟心感了！结合你提的「不想提前关联一堆外键表、拒绝低效逐行插入」的需求，我给你两个在PostgreSQL里完全可行的方案，既能保住有效数据的批量插入效率，又能精准抓出失败的记录：

方案一：纯PostgreSQL PL/pgSQL实现批量分块+错误捕获 #

这个方案不用额外工具，完全用PostgreSQL的内置功能实现，核心思路是大批次批量插入，遇到错误就拆分批次到更小粒度（甚至逐行）处理，兼顾效率和错误过滤。

步骤1：先创建存储失败记录的临时表 #

CREATE TEMP TABLE failed_records (
    record_data jsonb, -- 存失败行的完整数据
    error_message text, -- 错误详情
    error_timestamp timestamp DEFAULT now()
);

步骤2：编写PL/pgSQL批量处理逻辑 #

这里我写了一个自适应的处理逻辑：先按1000行批量插入，遇到错误就自动拆成100行的子批次，还失败就逐行处理——这样大部分正常数据还是批量插入，只有有问题的小批次才降级到逐行，平衡效率和容错：

DO $$
DECLARE
    v_batch_size bigint := 1000;
    v_total_rows bigint;
    v_processed bigint := 0;
    v_success bigint := 0;
    v_fail bigint := 0;
    
    -- 处理单批次的内部函数
    PROCEDURE process_batch(p_offset bigint, p_limit bigint)
    AS $$
    DECLARE
        rec record;
    BEGIN
        BEGIN
            -- 尝试批量插入
            INSERT INTO target_table (col1, col2, col3)
            SELECT col1, col2, col3 FROM temp_source_table
            OFFSET p_offset LIMIT p_limit
            ON CONFLICT (unique_key_col) DO UPDATE SET col2 = EXCLUDED.col2; -- 你的冲突更新逻辑
            
            v_success := v_success + p_limit;
        EXCEPTION
            WHEN OTHERS THEN
                -- 批量失败，拆分到更小批次（100行）
                IF p_limit > 100 THEN
                    process_batch(p_offset, p_limit / 10);
                    process_batch(p_offset + p_limit / 10, p_limit - p_limit / 10);
                ELSIF p_limit > 1 THEN
                    -- 拆到100行还失败，逐行处理
                    FOR rec IN SELECT * FROM temp_source_table OFFSET p_offset LIMIT p_limit LOOP
                        BEGIN
                            SAVEPOINT sp_single;
                            INSERT INTO target_table (col1, col2, col3)
                            VALUES (rec.col1, rec.col2, rec.col3)
                            ON CONFLICT (unique_key_col) DO UPDATE SET col2 = EXCLUDED.col2;
                            v_success := v_success + 1;
                        EXCEPTION
                            WHEN OTHERS THEN
                                ROLLBACK TO SAVEPOINT sp_single;
                                INSERT INTO failed_records (record_data, error_message)
                                VALUES (to_jsonb(rec), SQLERRM);
                                v_fail := v_fail + 1;
                        END;
                        RELEASE SAVEPOINT sp_single;
                    END LOOP;
                ELSE
                    -- 最后一行还是失败，直接记录
                    INSERT INTO failed_records (record_data, error_message)
                    SELECT to_jsonb(t), SQLERRM FROM temp_source_table t OFFSET p_offset LIMIT 1;
                    v_fail := v_fail + 1;
                END IF;
        END;
    END $$;
BEGIN
    SELECT COUNT(*) INTO v_total_rows FROM temp_source_table;
    
    WHILE v_processed < v_total_rows LOOP
        process_batch(v_processed, LEAST(v_batch_size, v_total_rows - v_processed));
        v_processed := v_processed + v_batch_size;
    END LOOP;
    
    RAISE NOTICE '批量处理完成：成功导入 % 行，失败 % 行', v_success, v_fail;
END $$;

方案一优缺点 #

• ✅ 纯PostgreSQL实现，无需额外工具
• ✅ 自适应批次大小，尽量保证批量插入效率
• ❌ 需要编写和维护PL/pgSQL代码，逻辑稍复杂

方案二：用pgloader实现高效错误跳过 #

你提到了pgloader，它刚好就是解决这种场景的神器——它的错误处理机制天生支持跳过违规行，效率接近PostgreSQL的COPY命令，还能自动记录失败数据。

步骤1：编写pgloader配置文件 #

创建一个名为batch_load.conf的配置文件，指定源表、目标表、错误处理规则：

LOAD TABLE
    FROM pgsql://your_user:your_password@localhost/your_db?table=temp_source_table
    INTO pgsql://your_user:your_password@localhost/your_db?table=target_table
    WITH
        on error skip, -- 跳过错误行
        log errors to '/var/log/pgloader/failed_records.log', -- 错误日志路径
        batch rows 1000, -- 批量大小
        truncate target -- 如果需要清空目标表（可选）
    SET
        work_mem = '64MB',
        maintenance_work_mem = '256MB';

步骤2：运行pgloader #

直接执行命令启动加载：

pgloader batch_load.conf

方案二优缺点 #

• ✅ 配置简单，无需写复杂代码
• ✅ 效率极高，接近原生COPY的速度
• ✅ 自动记录所有失败行的详细信息到日志
• ❌ 需要额外安装pgloader工具（大部分Linux发行版的包管理器都能直接装）

后续处理 #

不管用哪个方案，你都可以通过failed_records表（或pgloader的错误日志）查看失败的行和错误原因，比如外键引用的记录不存在，补全引用数据后，直接从失败表重新导入即可：

INSERT INTO target_table (col1, col2, col3)
SELECT (record_data->>'col1')::int, (record_data->>'col2')::text, (record_data->>'col3')::date
FROM failed_records
ON CONFLICT (unique_key_col) DO UPDATE SET col2 = EXCLUDED.col2;

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