如何提升Python向PostgreSQL插入数据的性能？

现有插入方法效率测试 #

测试了三种常见插入方法的效率，结果如下：

• 使用execute：每分钟40次插入
• 使用executemany：每分钟41次插入
• 使用psycopg2.extras.execute_values：每分钟42次插入

当前代码的核心问题 #

你提供的代码虽然用了execute_values，但每次循环只插入单条记录，且每次都重复建立、关闭数据库连接，这完全浪费了execute_values的批量插入优势，同时连接的频繁创建/销毁是最大的性能开销来源。

针对性优化方案 #

1. 复用数据库连接

将数据库连接的创建逻辑移到循环外部，整个批量插入过程复用同一个连接，避免重复的连接建立、销毁开销。

2. 真正实现批量插入

一次性准备好所有待插入的记录集合，调用一次execute_values完成批量插入，最大化利用该方法的性能优势。

3. 减少不必要的IO操作

移除循环中的控制台打印（如print(valor)），这类同步IO操作会显著拖慢插入速度。

4. 分批次处理大数据量

如果待插入数据量极大（如百万级），将数据分成若干批次（建议每批次1000-10000条）插入，避免内存占用过高和数据库锁竞争。

5. 数据库层面优化

• 插入前禁用目标表的索引（插入完成后重建），减少插入时的索引维护开销；
• 调整PostgreSQL配置参数，如增大work_mem、maintenance_work_mem，提升内存缓冲区利用率；
• 确保使用事务批量提交，避免单条记录提交的开销。

优化后的代码示例 #

import datetime
import psycopg2
from psycopg2 import extras
from typing import Any

def save_records_to_postgres(records_to_insert: list) -> None:
    insert_query = """
        INSERT INTO pricing.xxxx (description, code, unit, price, created_date, updated_date)
        VALUES %s 
        RETURNING id
    """
    
    # 准备批量插入的记录格式
    formatted_records = [
        (record[2], record[1], record[3], record[4], record[0], datetime.datetime.now())
        for record in records_to_insert
    ]

    conn = None
    cursor = None
    try:
        # 只建立一次连接
        conn = psycopg2.connect(
            database='xxxx',
            user='xxxx',
            password='xxxxx',
            host='xxxx',
            port='xxxx',
            connect_timeout=10
        )
        print("已连接PostgreSQL数据库")

        cursor = conn.cursor()
        # 一次性插入所有记录
        extras.execute_values(cursor, insert_query, formatted_records)
        
        conn.commit()
        print(f"成功插入 {len(formatted_records)} 条记录")

    except Exception as e:
        if conn:
            conn.rollback()
        print(f"插入失败: {str(e)}")
    finally:
        # 最后统一关闭连接和游标
        if cursor:
            cursor.close()
        if conn:
            conn.close()
            print("PostgreSQL连接已关闭")

# 假设df是你的DataFrame
valores = df.values
# 直接传入所有记录进行批量插入
save_records_to_postgres(valores)

大数据量分批次处理的扩展代码 #

如果数据量过大，可按如下方式分批次：

# 定义每批次大小
BATCH_SIZE = 1000
valores = df.values
total_records = len(valores)

# 分批次插入
for i in range(0, total_records, BATCH_SIZE):
    batch = valores[i:i+BATCH_SIZE]
    save_records_to_postgres(batch)
    print(f"已完成第 {i//BATCH_SIZE + 1} 批次插入")

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