如何优雅去除新旧DataFrame重复数据后写入MySQL数据库？

安全阿岳

2026-5-20

嘿，这个场景太常见了！处理多DataFrame写入MySQL时的重复数据，有几个优雅且实用的方案，分场景给你拆解下：

一、内存端去重（适合中小数据量）

如果你的数据规模不大，先在Python内存里完成去重再写入，逻辑直观又好维护：

基于唯一键全局去重
先把所有待写入的DataFrame合并，再根据业务唯一键（比如订单号order_id、用户IDuser_id）去重，同时可以结合时间戳确保保留最新数据：

import pandas as pd

# 假设你收集到的多个DataFrame存在列表中
df_collection = [df_1, df_2, df_3]
combined_df = pd.concat(df_collection)

# 按更新时间降序排序，再按唯一键去重，保留最新的那条数据
deduped_df = combined_df.sort_values(by="update_time", ascending=False)\
                        .drop_duplicates(subset=["order_id"], keep="first")

这里keep="first"对应排序后的第一条（也就是最新数据），如果想保留旧数据可以换成keep="last"。

对比数据库已有数据去重
如果数据库里已经存了历史数据，不想把重复记录再写进去，可以先拉取数据库中的唯一键集合，再过滤新数据：

# 从MySQL读取已有数据的唯一键
existing_keys = pd.read_sql("SELECT order_id FROM your_target_table", con=db_conn)\
                  ["order_id"].tolist()

# 过滤出新DataFrame中不存在于数据库的记录
to_write_df = deduped_df[~deduped_df["order_id"].isin(existing_keys)]

# 写入数据库
to_write_df.to_sql("your_target_table", con=db_conn, if_exists="append", index=False)

注意：如果数据库数据量极大，把所有唯一键拉到内存会占用过多资源，这时候更推荐用下面的数据库端方案。

二、数据库端去重（适合大数据量/高性能需求）

利用MySQL本身的特性处理重复，不用把全量数据拉到内存，效率更高，也是生产环境常用的优雅方式：

INSERT ... ON DUPLICATE KEY UPDATE
首先确保你的目标表已经为唯一键字段设置了PRIMARY KEY或UNIQUE约束，然后用自定义插入逻辑实现“重复则更新，无则插入”：

from sqlalchemy import text

# 把DataFrame转成字典格式的记录列表
records = deduped_df.to_dict("records")

# 构造插入语句，遇到重复键时更新指定字段
insert_stmt = text("""
    INSERT INTO your_target_table (order_id, name, amount, update_time)
    VALUES (:order_id, :name, :amount, :update_time)
    ON DUPLICATE KEY UPDATE
        name = VALUES(name),
        amount = VALUES(amount),
        update_time = VALUES(update_time)
""")

# 批量执行并提交
db_conn.execute(insert_stmt, records)
db_conn.commit()

如果只想跳过重复记录而不更新，可以把语句换成INSERT IGNORE INTO ...，但INSERT IGNORE会忽略所有插入错误，不如ON DUPLICATE KEY UPDATE精准。

临时表批量处理
面对超大数据量时，先把所有DataFrame写入临时表，再通过SQL的JOIN操作完成去重插入/更新，性能拉满：

# 把合并后的DataFrame写入临时表（如果存在则替换）
combined_df.to_sql("temp_target_table", con=db_conn, if_exists="replace", index=False)

# 插入临时表中不存在于主表的数据
insert_sql = text("""
    INSERT INTO your_target_table (order_id, name, amount, update_time)
    SELECT t.order_id, t.name, t.amount, t.update_time
    FROM temp_target_table t
    LEFT JOIN your_target_table y ON t.order_id = y.order_id
    WHERE y.order_id IS NULL
""")

# （可选）更新主表中存在但临时表数据更新的记录
update_sql = text("""
    UPDATE your_target_table y
    JOIN temp_target_table t ON y.order_id = t.order_id
    SET y.name = t.name, y.amount = t.amount, y.update_time = t.update_time
    WHERE t.update_time > y.update_time
""")

# 执行SQL并提交
db_conn.execute(insert_sql)
db_conn.execute(update_sql)
db_conn.commit()

三、从源头减少重复（最优优雅方案）

如果能从数据收集阶段就避免重复，那才是最省心的：

每次收集数据时，只获取增量部分：比如根据上次写入的最大时间戳或最大唯一键，从数据源（日志、API、文件）筛选出新产生的数据，从根源上减少重复。

示例代码：

# 获取数据库中最新的更新时间
latest_update = pd.read_sql("SELECT MAX(update_time) FROM your_target_table", con=db_conn)\
                  .iloc[0, 0]

# 从数据源只拉取最新时间之后的数据
incremental_df = fetch_data_from_source(start_time=latest_update)

# 直接写入，无需额外去重
incremental_df.to_sql("your_target_table", con=db_conn, if_exists="append", index=False)

总结下来，中小数据量选内存去重足够简单，大数据量用数据库端方案更高效，能从源头取增量数据就是最优解，你可以根据自己的业务场景灵活选择～

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