Pandas to_sql写入MySQL已有表时如何避免插入重复行

原生方法是否支持？ #

pandas 自带的to_sql方法没有直接实现插入去重的功能。其内置的if_exists参数仅支持3种模式，都无法在追加数据时自动过滤重复行：

• fail：目标表存在时直接抛出错误，终止写入
• replace：目标表存在时先删除原表、重建新表，再写入全量数据
• append：目标表存在时直接追加所有数据，不做重复校验

可行替代方案 #

方案1：自定义插入语句前缀（代码最简洁，性能最优） #

基于SQLAlchemy的方言扩展，给原生INSERT语句加上MySQL支持的IGNORE关键字，遇到唯一键冲突时自动跳过重复行，不需要建临时表。

前置要求：先给目标表dfx创建判断重复的唯一键约束，比如以col1、col2联合判定重复，需提前执行SQL建索引：

ALTER TABLE dfx ADD UNIQUE KEY uk_dup_check (col1, col2);

实现代码：

import pandas as pd
from sqlalchemy import create_engine
from sqlalchemy.dialects.mysql import insert

# 初始化数据库连接
user = 'user1'
pwd = 'xxxx'
host =  'aa1.us-west-1.rds.amazonaws.com'
port = 3306
database = 'main'
engine = create_engine("mysql+pymysql://{}:{}@{}/{}".format(user,pwd,host,database))

# 自定义INSERT IGNORE插入方法
def insert_ignore(table, conn, keys, data_iter):
    data = [dict(zip(keys, row)) for row in data_iter]
    insert_stmt = insert(table.table).prefix_with('IGNORE').values(data)
    conn.execute(insert_stmt)

# 调用to_sql时传入自定义方法即可
df.to_sql(
    name="dfx",
    con=engine,
    if_exists='append',
    index=False,
    method=insert_ignore
)

如果需要遇到重复行时更新已有字段而非直接跳过，把INSERT IGNORE换成ON DUPLICATE KEY UPDATE逻辑即可。

方案2：临时表中转同步（适合复杂同步逻辑） #

先把全量DataFrame写入独立临时表，再通过MySQL原生语法做去重同步，适合需要做复杂数据转换、批量校验的大数据量场景。
实现代码：

# 第一步：全量写入临时表，临时表会话结束自动删除，不影响正式表数据
df.to_sql(name="tmp_dfx", con=engine, if_exists='replace', index=False)

# 第二步：执行去重同步
with engine.connect() as con:
    # 字段顺序需要和DataFrame、目标表dfx的字段完全对齐
    sync_sql = """
    INSERT IGNORE INTO dfx (col1, col2, col3, other_fields)
    SELECT col1, col2, col3, other_fields FROM tmp_dfx;
    """
    con.execute(sync_sql)
    # 手动清理临时表（不执行也会在连接断开后自动删除）
    con.execute("DROP TABLE IF EXISTS tmp_dfx;")

方案3：Python侧预去重（适合小数据量单线程场景） #

如果不想修改数据库表结构、且数据量较小，可以先读取目标表已有数据，在Python侧和待写入DataFrame做差集，仅插入不存在的新行。
实现代码：

# 读取目标表已有数据
existing_data = pd.read_sql("SELECT col1, col2 FROM dfx", con=engine)
# 指定判定重复的字段
dup_check_cols = ['col1', 'col2']
# 筛选出不存在于目标表的新行
new_data = df.merge(
    existing_data[dup_check_cols],
    on=dup_check_cols,
    how='left',
    indicator=True
).query("_merge == 'left_only'").drop(columns=['_merge'])
# 仅写入新行
new_data.to_sql(name="dfx", con=engine, if_exists='append', index=False)

注意：该方案存在并发写入重复的风险，且性能会随目标表数据量增长快速下降，不推荐生产环境大数据量场景使用。

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