PostgreSQL批量插入时如何跳过失败行保证其余行正常插入成功

无额外依赖的内置实现方案 #

• COPY ... ON ERROR 服务端批量导入（优先推荐）
  PostgreSQL 12及以上版本支持，是所有方案中性能最高的选择，适合从CSV/文本文件直接导入的场景
  支持直接跳过插入错误的行，同时可将错误信息和原始行写入指定的日志表，无需客户端额外处理逻辑：

  -- 一次性初始化错误日志表
  CREATE TABLE IF NOT EXISTS etl_import_errors (
      error_time TIMESTAMPTZ DEFAULT NOW(),
      error_message TEXT,
      raw_record TEXT
  );
  
  -- 执行批量导入，自动跳过错误行，错误记录存入日志表
  COPY target_table(column1, column2, column3)
  FROM '/path/to/your/data_file.csv'
  WITH (FORMAT csv, HEADER, DELIMITER ',')
  ON ERROR LOG INTO etl_import_errors
  CONTINUE;
  

  如果是从程序内存中批量写入，可以调用COPY FROM STDIN语法，同样支持ON ERROR参数。

• 子批次拆分+事务保存点（兼容低版本PostgreSQL）
  对于12以下的低版本PostgreSQL，可将全量批次拆分为100~1000行的小子批次，优先批量提交正常子批次，仅对报错的子批次逐行校验找错，性能远高于全量逐行插入：

  # Python伪代码示例，其他语言逻辑通用
  import psycopg2
  BATCH_SIZE = 500
  all_rows = [...] # 待导入全量数据
  error_rows = []
  conn = psycopg2.connect("your_db_connection_string")
  cursor = conn.cursor()
  
  for i in range(0, len(all_rows), BATCH_SIZE):
      sub_batch = all_rows[i:i+BATCH_SIZE]
      try:
          cursor.execute("BEGIN")
          cursor.executemany("INSERT INTO target_table VALUES (%s, %s, %s)", sub_batch)
          cursor.execute("COMMIT")
      except Exception as e:
          cursor.execute("ROLLBACK")
          # 仅出错的子批次逐行插入，定位错误行
          for row in sub_batch:
              try:
                  cursor.execute("BEGIN")
                  cursor.execute("INSERT INTO target_table VALUES (%s, %s, %s)", row)
                  cursor.execute("COMMIT")
              except Exception as row_err:
                  cursor.execute("ROLLBACK")
                  error_rows.append({"row": row, "error": str(row_err)})
  

• INSERT ... ON CONFLICT 处理唯一键冲突
  如果仅需要处理重复键类异常，其他错误提前在客户端校验完成，可以直接用该语法跳过重复行，或自定义冲突时的更新逻辑：

  INSERT INTO target_table (id, name, mobile)
  VALUES (1, '张三', '130xxxxxxx'), (2, '李四', '131xxxxxxx')
  ON CONFLICT (id) DO NOTHING; -- 重复id直接跳过
  

常用ETL工具内置支持方案 #

如果你的流水线已经在使用成熟ETL工具，可以直接开启工具自带的容错能力，无需自定义开发：

• 数据集成类工具（Airbyte、Fivetran、DataX）：配置写入目标端时开启脏数据容错，设置最大允许错误数，工具会自动将错误行写入专门的死信表/日志存储，正常数据正常落库。
• Spark大数据流水线：使用PostgreSQL连接器写入时，配置option("errorLimit", "1000")参数允许指定数量的错误行，也可配合foreachBatch算子自定义错误行处理逻辑。
• dbt数据转换场景：可通过前置dbt test校验提前过滤问题数据，也可自定义插入宏实现错误行跳过。

性能优化建议 #

• 导入前临时关闭目标表的非必要索引、触发器，导入完成后再重建，可提升30%以上写入速度
• 客户端提前做轻量前置校验：非空字段判空、字符串长度校验、枚举值范围校验等，提前过滤明显错误行，降低服务端出错概率
• 批量导入期间临时调整PostgreSQL参数：调大maintenance_work_mem、wal_buffers，关闭wal_log_hints，可进一步提升写入吞吐量

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