如何对比DataFrame与数据库实现新值插入、变更值更新？

核心思路 #

利用数据库原生的**Upsert（插入或更新）**机制，结合pandas批量操作能力，替代逐条处理，大幅提升效率。核心是基于fruit列的唯一约束，让数据库自动判断是插入新数据还是更新现有数据。

前提准备 #

首先需要给数据库表的fruit列添加唯一约束，确保每个fruit值唯一：

-- MySQL
ALTER TABLE your_table ADD UNIQUE KEY idx_fruit (fruit);

-- PostgreSQL
ALTER TABLE your_table ADD CONSTRAINT unique_fruit UNIQUE (fruit);

方案一：直接使用数据库Upsert（推荐） #

这种方式无需在Python中做数据对比，直接让数据库处理冲突，代码简洁且效率最高。

MySQL 实现 #

from sqlalchemy import create_engine
import pandas as pd

# 初始化数据库连接
engine = create_engine('mysql+pymysql://user:password@host:port/db_name')

# 构造Upsert语句
upsert_sql = """
INSERT INTO your_table (fruit, price)
VALUES (%s, %s)
ON DUPLICATE KEY UPDATE
price = VALUES(price)
"""

# 批量执行
with engine.connect() as conn:
    # 将DataFrame转为可执行的参数列表
    data = df_new.to_records(index=False).tolist()
    conn.execute(upsert_sql, data)
    conn.commit()

PostgreSQL 实现 #

from sqlalchemy import create_engine
import pandas as pd

engine = create_engine('postgresql://user:password@host:port/db_name')

upsert_sql = """
INSERT INTO your_table (fruit, price)
VALUES (%s, %s)
ON CONFLICT (fruit) DO UPDATE SET
price = EXCLUDED.price
"""

with engine.connect() as conn:
    data = df_new.to_records(index=False).tolist()
    conn.execute(upsert_sql, data)
    conn.commit()

方案二：先对比再批量操作 #

如果需要精确控制仅更新price变化的数据，可先读取数据库现有数据做对比，再分别执行更新和插入：

from sqlalchemy import create_engine
import pandas as pd

engine = create_engine('your_database_connection_string')

# 读取数据库现有fruit和price数据
df_db = pd.read_sql("SELECT fruit, price FROM your_table", engine)

# 筛选需要更新的数据：fruit存在但price不同
df_update = pd.merge(df_new, df_db, on='fruit', how='inner')
df_update = df_update[df_update['price_x'] != df_update['price_y']][['fruit', 'price_x']]
df_update.rename(columns={'price_x': 'price'}, inplace=True)

# 筛选需要插入的数据：fruit不存在于数据库
df_insert = df_new[~df_new['fruit'].isin(df_db['fruit'])]

# 批量执行操作
with engine.connect() as conn:
    # 执行更新
    if not df_update.empty:
        update_sql = "UPDATE your_table SET price = %s WHERE fruit = %s"
        conn.execute(update_sql, df_update.to_records(index=False).tolist())
    # 执行插入
    if not df_insert.empty:
        df_insert.to_sql('your_table', engine, if_exists='append', index=False)
    conn.commit()

说明 #

• 方案一适合大部分场景，无需额外对比逻辑，数据库自动处理冲突，效率最高
• 方案二更适合更新比例较低的场景，可减少不必要的数据库更新操作
• 两种方案均为批量操作，避免了逐条处理的性能瓶颈

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