如何用pandas.to_sql()实现PostgreSQL的UPSERT操作

pandas 的 .to_sql() 方法本身没有直接支持 UPSERT 的配置参数，但可以结合 PostgreSQL 原生语法实现需求，以下是两种实用方案：

方案一：临时表 + PostgreSQL 原生UPSERT（推荐，适合大数据量） #

通过临时表批量导入数据，再利用 PostgreSQL 的 INSERT ... ON CONFLICT ... DO UPDATE 语法完成高效的UPSERT操作，避免逐行处理的性能损耗。

步骤代码： #

1. 将DataFrame写入临时表（临时表可安全覆盖，不影响原业务表）：

my_table.to_sql(
    'temp_table',
    con=engine,
    schema='wrhouse',
    if_exists='replace',
    index=False
)

2. 执行UPSERT同步逻辑：

# 动态生成更新字段（排除主键date）
update_cols = ", ".join([f"{col} = EXCLUDED.{col}" for col in my_table.columns if col != 'date'])

# 构建UPSERT SQL语句
upsert_sql = f"""
INSERT INTO wrhouse."table" ({', '.join(my_table.columns)})
SELECT {', '.join(my_table.columns)} FROM wrhouse.temp_table
ON CONFLICT (date) DO UPDATE SET {update_cols};
"""

# 执行SQL
with engine.begin() as conn:
    conn.execute(upsert_sql)

说明： #

• EXCLUDED 是PostgreSQL关键字，代表冲突时原本要插入的行数据
• 动态生成字段的方式适用于列较多的场景，无需手动编写所有字段
• 临时表会在数据库会话结束后自动销毁，也可手动执行 DROP TABLE wrhouse.temp_table; 清理

方案二：拆分数据为插入/更新组（适合小数据量） #

如果数据量较小，可先查询数据库中已存在的主键，拆分DataFrame后分别执行插入和更新：

# 查询数据库中已存在的date
existing_dates = pd.read_sql("SELECT date FROM wrhouse.\"table\"", con=engine)['date']

# 拆分数据：新行（不存在的date）和待更新行（已存在的date）
new_rows = my_table[~my_table['date'].isin(existing_dates)]
update_rows = my_table[my_table['date'].isin(existing_dates)]

# 插入新行
new_rows.to_sql('table', con=engine, schema='wrhouse', if_exists='append', index=False)

# 逐行更新现有数据
update_cols = ", ".join([f"{col} = %s" for col in my_table.columns if col != 'date'])
for _, row in update_rows.iterrows():
    update_sql = f"""
    UPDATE wrhouse."table"
    SET {update_cols}
    WHERE date = %s;
    """
    params = tuple(row[col] for col in my_table.columns if col != 'date') + (row['date'],)
    engine.execute(update_sql, params)

说明： #

• 该方法需要额外查询数据库，大数据量下会增加开销，且逐行更新效率较低
• 同样可通过动态生成字段简化SQL编写

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