如何基于DataFrame识别数据变更并通过SQLAlchemy执行INSERT/UPDATE

前提说明 #

首先确保你的DataFrame有唯一主键列（如user_id），这是匹配新旧数据、区分新增/修改的核心依据。

步骤1：识别新增行 #

本地DataFrame中存在但数据库中没有的行，就是需要插入的新增数据：

# 方法1：用isin快速筛选
new_rows = df[~df['user_id'].isin(df2['user_id'])]

# 方法2：用merge的indicator标记（适合需要查看匹配状态的场景）
merge_result = pd.merge(df[['user_id']], df2[['user_id']], on='user_id', how='left', indicator=True)
new_rows = df[merge_result['_merge'] == 'left_only']

步骤2：识别修改行 #

筛选出两边都存在但内容有差异的行：

# 先获取两边共有的主键ID
common_ids = df[df['user_id'].isin(df2['user_id'])]['user_id']

# 取出共同行并按主键索引对齐
df_local = df[df['user_id'].isin(common_ids)].set_index('user_id')
df_db = df2[df2['user_id'].isin(common_ids)].set_index('user_id')

# 对比所有非主键列，找出有差异的行
changed_rows = df_local[~df_local.eq(df_db).all(axis=1)].reset_index()

# （可选）处理浮点列精度问题
import numpy as np
for col in df_local.columns:
    if np.issubdtype(df_local[col].dtype, np.floating):
        # 用np.isclose替代等于判断，避免精度误差
        mask = np.isclose(df_local[col], df_db[col])
    else:
        mask = df_local[col] == df_db[col]
    changed_mask = ~mask.all(axis=1)
changed_rows = df_local[changed_mask].reset_index()

步骤3：用SQLAlchemy执行批量操作 #

初始化连接与表对象

from sqlalchemy import create_engine, Table, MetaData

# 构造SQL Server连接引擎
engine = create_engine('mssql+pyodbc://用户名:密码@服务器/数据库?driver=ODBC+Driver+17+for+SQL+Server')
metadata = MetaData()
# 反射数据库中的目标表（假设表名为target_table）
target_table = Table('target_table', metadata, autoload_with=engine)

批量插入新增行

if not new_rows.empty:
    with engine.begin() as conn:
        # 把DataFrame转成字典列表，批量插入
        conn.execute(target_table.insert(), new_rows.to_dict('records'))

批量更新修改行

if not changed_rows.empty:
    with engine.begin() as conn:
        # 遍历修改行，按主键匹配执行更新
        for _, row in changed_rows.iterrows():
            update_stmt = (
                target_table.update()
                .where(target_table.c.user_id == row['user_id'])
                .values(**row.drop('user_id').to_dict())
            )
            conn.execute(update_stmt)

（进阶）用SQL Server MERGE语句高效批量处理

如果数据量较大，推荐用SQL原生的MERGE语句，比逐行更新效率更高：

# 1. 先把本地更新后的DataFrame写入临时表
df.to_sql('#temp_update', engine, if_exists='replace', index=False)

# 2. 构造MERGE语句
merge_sql = """
MERGE INTO target_table AS target
USING (SELECT * FROM #temp_update) AS source
ON target.user_id = source.user_id
WHEN MATCHED THEN
    UPDATE SET
        col1 = source.col1,
        col2 = source.col2,
        col3 = source.col3  -- 列出所有需要同步的列
WHEN NOT MATCHED THEN
    INSERT (user_id, col1, col2, col3)
    VALUES (source.user_id, source.col1, source.col2, source.col3);
"""

# 3. 执行MERGE
with engine.begin() as conn:
    conn.execute(merge_sql)

注意事项 #

• 确保数据库主键列有唯一约束，避免重复插入。
• 处理空值：pd.eq会将NaN视为不相等，若需忽略空值差异，可先填充默认值再对比。
• 性能优化：数据量超10万行时，优先用MERGE语句或批量写入临时表再同步。

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