如何通过Polars的write_database避免重复记录并更新数据？

Polars的write_database方法默认是追加写入，本身不直接支持"存在则更新、不存在则插入"（Upsert）的逻辑，但可以通过结合数据库原生语法或SQLAlchemy工具来实现，核心思路是先处理数据匹配逻辑，再完成写入/更新操作。

方法一：临时表+数据库原生Upsert语法 #

这是最通用且高效的方案，尤其适合批量数据操作：

1. 将待导入数据写入临时表
   先把Polars DataFrame写入数据库的临时表，用if_exists="replace"确保每次都覆盖临时表内容：

   import polars as pl
   
   # 待更新的目标数据
   update_df = pl.DataFrame({
       "user_id": [1, 2, 3],
       "username": ["alice_updated", "bob", "charlie_new"],
       "email": ["alice@new.com", "bob@old.com", "charlie@new.com"]
   })
   
   # 写入临时表
   update_df.write_database(
       table_name="temp_user_data",
       connection="postgresql://user:password@host:port/db_name",
       if_exists="replace"
   )
   

2. 执行数据库原生Upsert语句
   根据数据库类型，执行对应的合并语句，以user_id为匹配键同步数据到目标表：

   • PostgreSQL（使用ON CONFLICT）：

     from sqlalchemy import create_engine
     
     engine = create_engine("postgresql://user:password@host:port/db_name")
     with engine.connect() as conn:
         conn.execute("""
             INSERT INTO users (user_id, username, email)
             SELECT user_id, username, email FROM temp_user_data
             ON CONFLICT (user_id) DO UPDATE SET
                 username = EXCLUDED.username,
                 email = EXCLUDED.email;
         """)
         conn.commit()
     

   • MySQL（使用ON DUPLICATE KEY UPDATE，需确保user_id是主键或唯一索引）：

     with engine.connect() as conn:
         conn.execute("""
             INSERT INTO users (user_id, username, email)
             SELECT user_id, username, email FROM temp_user_data
             ON DUPLICATE KEY UPDATE
                 username = VALUES(username),
                 email = VALUES(email);
         """)
         conn.commit()
     

   • SQL Server（使用MERGE语句）：

     with engine.connect() as conn:
         conn.execute("""
             MERGE INTO users AS target
             USING temp_user_data AS source
             ON target.user_id = source.user_id
             WHEN MATCHED THEN
                 UPDATE SET username = source.username, email = source.email
             WHEN NOT MATCHED THEN
                 INSERT (user_id, username, email) VALUES (source.user_id, source.username, source.email);
         """)
         conn.commit()
     

方法二：用SQLAlchemy直接实现Upsert #

如果数据量不大，可直接用SQLAlchemy的Core API构造Upsert语句，无需临时表：

from sqlalchemy import Table, Column, Integer, String, MetaData
import polars as pl

# 定义目标表结构（需与数据库实际表一致）
metadata = MetaData()
users_table = Table(
    "users", metadata,
    Column("user_id", Integer, primary_key=True),
    Column("username", String(50)),
    Column("email", String(100))
)

# 待更新数据
update_df = pl.DataFrame({
    "user_id": [1, 2, 3],
    "username": ["alice_updated", "bob", "charlie_new"],
    "email": ["alice@new.com", "bob@old.com", "charlie@new.com"]
})

# 转换为字典列表
records = update_df.to_dicts()

# 执行Upsert
engine = create_engine("postgresql://user:password@host:port/db_name")
with engine.connect() as conn:
    insert_stmt = users_table.insert().values(records)
    # 添加冲突更新逻辑
    upsert_stmt = insert_stmt.on_conflict_do_update(
        index_elements=["user_id"],  # 匹配键
        set_={
            "username": insert_stmt.excluded.username,
            "email": insert_stmt.excluded.email
        }
    )
    conn.execute(upsert_stmt)
    conn.commit()

关键注意事项 #

• 匹配键必须加唯一约束：无论是主键还是唯一索引，数据库需要通过这个约束识别重复记录，否则Upsert逻辑无法生效。
• 性能优先选择临时表：批量数据场景下，先写临时表再合并的性能远高于逐条Upsert。
• 字段一致性检查：确保待导入数据的字段名称、类型与目标表完全匹配，避免因不兼容导致报错。

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