如何用Python DataFrame高效更新PostgreSQL表？

先排查to_sql未生效的原因 #

1. 确认DataFrame是否有数据
   执行print(df.shape)和print(df.head())，确保要插入的DataFrame非空且数据符合预期。
2. 验证数据库连接正确性
   用以下代码确认连接的是目标数据库和表：

   from sqlalchemy import create_engine, text
   
   engine = create_engine("postgresql+psycopg2://user:password@host:port/dbname")
   with engine.connect() as conn:
       # 检查表是否存在
       exists = conn.execute(text("SELECT EXISTS (SELECT FROM information_schema.tables WHERE table_name = 'my_table')")).scalar()
       print("目标表存在：", exists)
       # 查看原表数据量
       count = conn.execute(text("SELECT COUNT(*) FROM my_table")).scalar()
       print("原表行数：", count)
   

3. 检查if_exists参数是否正确
   • 若要追加数据到已有表，必须用if_exists="append"（原代码用replace会删除原表重建，若DataFrame为空则生成空表）；
   • 若要替换整个表（删除原数据），才用if_exists="replace"，需谨慎操作。
4. 确认列名与表结构匹配
   PostgreSQL列名默认区分大小写（除非创建表时用引号包裹列名），确保DataFrame的列名和目标表列名完全一致，否则会出现插入失败或数据不匹配。

优化后的批量插入方案 #

方案1：用SQLAlchemy的to_sql批量追加（推荐，代码简洁） #

from sqlalchemy import create_engine

engine = create_engine("postgresql+psycopg2://user:password@host:port/dbname")

# 批量追加数据到已有表
df.to_sql(
    name="my_table",
    con=engine,
    index=False,
    if_exists="append",
    chunksize=1000,  # 分批次插入，避免内存溢出
    method="multi"   # 启用多值插入，大幅提升速度
)

engine.dispose()

• 无需手动处理NULL值：pandas会自动将None/NaN转换为SQL的NULL；
• method="multi"将多条数据合并为一个INSERT语句执行，比逐行插入快数倍；
• chunksize适合大数据量，避免一次性加载过多数据到内存。

方案2：用psycopg2的execute_values（更快，适合中大数据量） #

import psycopg2
from psycopg2.extras import execute_values

conn = psycopg2.connect("postgres://user:password@host:port/dbname")
cur = conn.cursor()

# 将DataFrame转换为元组列表
data_tuples = [tuple(row) for row in df.to_numpy()]

# 批量插入，page_size控制每批次插入行数
execute_values(
    cur,
    "INSERT INTO my_table VALUES %s",
    data_tuples,
    page_size=1000
)

conn.commit()
conn.close()

方案3：用psycopg2的copy_from（最快，适合超大数据集） #

利用PostgreSQL的COPY协议，速度比普通INSERT快一个数量级：

import psycopg2
from io import StringIO

conn = psycopg2.connect("postgres://user:password@host:port/dbname")
cur = conn.cursor()

# 将DataFrame转为TSV格式的内存缓冲区
csv_buffer = StringIO()
df.to_csv(csv_buffer, sep='\t', header=False, index=False, na_rep='NULL')
csv_buffer.seek(0)  # 重置缓冲区指针到开头

# 执行COPY导入
cur.copy_from(
    file=csv_buffer,
    table='my_table',
    sep='\t',
    null='NULL'
)

conn.commit()
conn.close()

原逐行插入代码的问题 #

1. SQL注入风险：用f-string拼接SQL语句，若字段包含单引号或恶意内容，会导致语法错误或被注入攻击；
2. 效率极低：逐行执行INSERT会频繁与数据库交互，大数据量下速度极慢；
3. NULL值处理错误：手动替换'None'为NULL的逻辑不严谨，若字段本身包含'None'字符串会被错误替换，而pandas原生支持NULL值转换。

内容的提问来源于stack exchange，提问作者Angel Peñaflor