Pandas DataFrame写入SQL表时校验重复行的高效方案咨询

核心优化思路是避免全量拉取/替换SQL表数据，所有重复校验逻辑尽量下沉到数据库侧或仅做增量比对，同时保留你需要的新增列适配、DataFrame自身去重能力。

前置准备 #

• 先给SQL目标表的唯一识别字段（可唯一确定一行数据的1个或多个字段）创建联合唯一索引，这是底层避免重复、提升查询效率的基础，以SQL Server为例的建索引语句：

CREATE UNIQUE NONCLUSTERED INDEX idx_自定义索引名 ON dbo.你的表名 (字段1, 字段2, ...字段N)

• 写入前先清理待写入DataFrame自身的重复行：

# 按你的唯一键字段去重，保留第一条即可
df_sql = df_sql.drop_duplicates(subset=['唯一字段1','唯一字段2'], keep='first').reset_index(drop=True)

方案1：数据库原生去重写入（最高效，推荐） #

所有校验逻辑在数据库侧执行，无需拉取主表数据，性能不受主表体量增长影响，同时兼容新增列场景：

1. 先适配新增列：检测待写入df的列是否在SQL表中存在，不存在则自动加列

from sqlalchemy import inspect

# 获取SQL表现有列
inspector = inspect(engine)
existing_cols = [col['name'] for col in inspector.get_columns(table_name, schema='dbo')]
new_cols = [col for col in df_sql.columns if col not in existing_cols]

# 给SQL表新增不存在的列（默认用NVARCHAR(MAX)兼容，可根据实际类型调整）
for col in new_cols:
    with engine.connect() as conn:
        conn.execute(f"ALTER TABLE dbo.{table_name} ADD [{col}] NVARCHAR(MAX)")
        conn.commit()

2. 用临时表+MERGE语句实现存在则跳过、不存在则写入，以SQL Server为例：

# 先把待写入df导入临时表
df_sql.to_sql(f'#{table_name}_temp', con=engine, schema='dbo', if_exists='replace', index=False)

# 执行MERGE写入，自动跳过重复行
merge_sql = f"""
MERGE dbo.{table_name} AS target
USING dbo.[#{table_name}_temp] AS source
ON target.唯一字段1 = source.唯一字段1 AND target.唯一字段2 = source.唯一字段2 -- 替换为你的唯一键匹配逻辑
WHEN NOT MATCHED THEN
    INSERT ({','.join([f'[{col}]' for col in df_sql.columns])})
    VALUES ({','.join([f'source.[{col}]' for col in df_sql.columns])});
"""
with engine.connect() as conn:
    conn.execute(merge_sql)
    conn.commit()

方案2：增量比对写入（改动最小，适合每次写入数据量小的场景） #

无需修改数据库逻辑，仅拉取主表中和待写入数据相关的行做比对，避免全量拉取：

# 先处理df自身重复
df_sql = df_sql.drop_duplicates(subset=['唯一字段1','唯一字段2'], keep='first').reset_index(drop=True)

# 拼接唯一键的查询条件，仅拉取主表中可能和待写入数据重复的行
unique_vals = df_sql[['唯一字段1','唯一字段2']].values.tolist()
# 生成IN查询条件，这里以两个唯一字段为例
conditions = " OR ".join([f"(唯一字段1 = '{val[0]}' AND 唯一字段2 = '{val[1]}')" for val in unique_vals])
exist_data = pd.read_sql(f"SELECT 唯一字段1, 唯一字段2 FROM dbo.{table_name} WHERE {conditions}", engine)

# 过滤掉df中已经存在的行
df_to_write = df_sql.merge(exist_data, on=['唯一字段1','唯一字段2'], how='left', indicator=True)
df_to_write = df_to_write[df_to_write['_merge'] == 'left_only'].drop(columns=['_merge'])

# 直接追加写入即可
if not df_to_write.empty:
    # 可在此处插入前面的新增列适配逻辑，确保新增列存在
    df_to_write.to_sql(table_name, con=engine, schema='dbo', if_exists='append', index=False)

原代码的问题说明 #

你原有的try-except逻辑风险很高：to_sql抛出异常的原因不一定是重复键，可能是数据库连接中断、字段类型不匹配等非重复问题，此时触发全量替换容易导致数据丢失，且全量拉取、替换在表体量超过10万行后性能会骤降，还会长期锁表影响其他业务访问。

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