使用SQLAlchemy实现MariaDB批量Upsert的高效方案

针对你遇到的批量刷新商品数据需求，MariaDB原生支持的INSERT ... ON DUPLICATE KEY UPDATE是最高效的解决方案，结合SQLAlchemy可以通过两种方式实现，彻底解决merge()速度慢、add_all()主键冲突的问题：

方案一：ORM批量映射插入（推荐，代码简洁） #

SQLAlchemy 1.4及以上版本支持bulk_insert_mappings配合on_duplicate_key_update参数，直接基于你的Item模型实现批量Upsert：

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from your_model_module import Item, Base

# 初始化数据库连接
engine = create_engine("mariadb+mariadbconnector://用户名:密码@主机:端口/数据库名")
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

def batch_upsert_items(items_data):
    """
    items_data为字典列表，每个字典包含item_id、price、updated_at等字段
    """
    db = SessionLocal()
    try:
        db.bulk_insert_mappings(
            Item,
            items_data,
            # 指定主键冲突时需要更新的字段
            on_duplicate_key_update={
                "price": Item.price,
                "updated_at": Item.updated_at,
                # 其他需要同步的字段可在此添加
            }
        )
        db.commit()
    except Exception as e:
        db.rollback()
        raise e
    finally:
        db.close()

# 示例调用
# items_list = [{"item_id": 1, "price": 99, "updated_at": 1718000000}, ...]
# batch_upsert_items(items_list)

这个方案依托ORM封装，底层会生成高效的批量INSERT语句，配合ON DUPLICATE KEY UPDATE逻辑，性能和原生SQL几乎一致，3万条数据的刷新速度会比merge()快一个数量级。

方案二：原生SQL构造（灵活适配复杂逻辑） #

如果需要更复杂的更新规则（比如更新时做字段计算），可以直接构造原生SQL执行：

def batch_upsert_items_raw(items_data):
    db = SessionLocal()
    try:
        # 提取字段名和对应值列表
        columns = ["item_id", "price", "updated_at"]
        values = [(item["item_id"], item["price"], item["updated_at"]) for item in items_data]
        
        # 构造批量Upsert语句
        insert_stmt = f"""
            INSERT INTO items ({', '.join(columns)})
            VALUES ({', '.join(['%s']*len(columns))})
            ON DUPLICATE KEY UPDATE
                price = VALUES(price),
                updated_at = VALUES(updated_at)
        """
        # 批量执行参数绑定
        db.execute(insert_stmt, values)
        db.commit()
    except Exception as e:
        db.rollback()
        raise e
    finally:
        db.close()

关键注意事项 #

• 确保item_id是主键/唯一键：MariaDB的ON DUPLICATE KEY UPDATE依赖主键或唯一索引判断冲突，你的模型中item_id已设为主键，满足要求。
• 分批次处理超大数据：如果数据量超过10万条，建议分批次（比如每批5000条）执行，避免单次请求数据量过大导致数据库超时。
• 关闭自动提交/刷新：初始化Session时设置autocommit=False, autoflush=False，减少不必要的数据库交互开销。
• 启用连接池：生产环境依赖SQLAlchemy默认的连接池，避免频繁创建销毁数据库连接。

之前merge()速度慢是因为每条数据都要先查询再判断操作，3万条数据会产生6万次数据库请求；add_all()仅做插入操作，遇到主键冲突自然报错。上面两种方案均为批量操作，单次请求处理多条数据，性能会大幅提升。

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