如何改进PostgreSQL批量插入方法以处理重复记录IntegrityError

Great question! The copy_from method is hands down the fastest way to bulk load data into PostgreSQL, but its Achilles' heel is handling unique constraint conflicts—it aborts the entire operation on the first IntegrityError. Here are the most practical, optimized solutions to fix this while keeping that sweet bulk load speed:

1. Temporary Table + ON CONFLICT (Recommended Best Practice) #

This is the gold standard for this scenario. It leverages PostgreSQL's native bulk operations and conflict handling to keep things fast and atomic. Here's the step-by-step workflow:

1. Create a temporary table that mirrors the structure of your target table (temp tables are in-memory, fast, and automatically cleaned up after your session).
2. Use copy_from to bulk load all your data into the temporary table—no conflict checks needed here, so it stays blazingly fast.
3. Copy data to the target table with INSERT ... SELECT ... ON CONFLICT, where you define exactly how to handle duplicates (skip them, update existing records, etc.).

Example Code #

import psycopg2
import io
import pandas as pd

# Establish database connection
conn = psycopg2.connect(
    dbname="your_database",
    user="your_user",
    password="your_password",
    host="your_host"
)
cur = conn.cursor()

# Step 1: Create temporary table matching target table structure
cur.execute("""
CREATE TEMP TABLE temp_sales (
    sale_id INT PRIMARY KEY,
    product_name VARCHAR(100),
    sale_amount NUMERIC(10,2)
) ON COMMIT DROP; -- Auto-delete when transaction ends
""")

# Step 2: Bulk load DataFrame into temp table using copy_from
# Convert DataFrame to tab-separated string (matches copy_from's default sep)
data = pd.read_excel("sales_data.xlsx")
output_buffer = io.StringIO()
data.to_csv(output_buffer, sep="\t", header=False, index=False)
output_buffer.seek(0)  # Reset buffer to start

cur.copy_from(
    file=output_buffer,
    table="temp_sales",
    sep="\t",
    columns=("sale_id", "product_name", "sale_amount")
)

# Step 3: Move data to target table, handle duplicates
cur.execute("""
INSERT INTO sales (sale_id, product_name, sale_amount)
SELECT sale_id, product_name, sale_amount FROM temp_sales
ON CONFLICT (sale_id) DO NOTHING; -- Skip duplicate records
-- OR for updates: DO UPDATE SET product_name = EXCLUDED.product_name, sale_amount = EXCLUDED.sale_amount
""")

# Get metrics for success/skipped records
cur.execute("SELECT COUNT(*) FROM temp_sales")
total_loaded = cur.fetchone()[0]
inserted = cur.rowcount
skipped = total_loaded - inserted

print(f"Loaded {total_loaded} records total.")
print(f"Successfully inserted {inserted} new records.")
print(f"Skipped {skipped} duplicate records.")

conn.commit()
cur.close()
conn.close()

2. Pre-Filter Duplicates in Pandas (For Smaller Datasets) #

If your dataset isn't huge (think 100k rows or less), you can pre-filter out existing records before using copy_from. This avoids hitting conflicts entirely, but note:

• It can cause race conditions if other processes insert records between your query and bulk load.
• It uses more memory if you're fetching a large list of existing keys.

Example Code #

# Fetch existing unique keys from target table
cur.execute("SELECT sale_id FROM sales")
existing_ids = {row[0] for row in cur.fetchall()}

# Filter DataFrame to keep only new records
filtered_data = data[~data["sale_id"].isin(existing_ids)]

# Proceed with copy_from as usual
output_buffer = io.StringIO()
filtered_data.to_csv(output_buffer, sep="\t", header=False, index=False)
output_buffer.seek(0)

cur.copy_from(output_buffer, "sales", sep="\t", columns=("sale_id", "product_name", "sale_amount"))

3. Use copy_expert for Custom Bulk Loads #

If you need more control over the COPY process (like handling CSV with custom delimiters, headers, or formatting), use copy_expert instead of copy_from. It lets you run a full PostgreSQL COPY command, which you can still pair with the temporary table + ON CONFLICT workflow.

Example Code #

# Load directly from a CSV file (or Excel converted to CSV)
with open("sales_data.csv", "r") as f:
    cur.copy_expert("""
    COPY temp_sales (sale_id, product_name, sale_amount)
    FROM STDIN
    WITH (FORMAT CSV, HEADER FALSE, DELIMITER ',');
    """, f)

# Then run the same INSERT...ON CONFLICT command as before

Key Notes #

• Always ensure your temporary table's data types match the target table to avoid conversion errors.
• The ON CONFLICT clause requires a unique constraint (primary key or unique index) on the target table for the column(s) you're checking for duplicates.
• Using ON COMMIT DROP on the temporary table ensures it's cleaned up automatically, even if your transaction fails.

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