SQLAlchemy结合PostgreSQL批量插入（on_conflict_do_nothing）性能优化咨询

环境信息 #

• SQLAlchemy版本：2.0.36
• PostgreSQL版本：Docker镜像postgres:15-alpine

背景 #

最初使用psycopg2实现批量插入，代码如下：

sql = """INSERT INTO ego_pose(token, translation, rotation, timestamp)
    VALUES %s ON CONFLICT DO NOTHING;"""
with conn.cursor() as cur:
    sql_args_batch = [
        (
            ego_pose["token"],
            ego_pose["translation"],
            ego_pose["rotation"],
            ego_pose["timestamp"],
        )
        for ego_pose in ego_poses
    ]
    execute_values(cur, sql, sql_args_batch)

迁移到SQLAlchemy后，通过定义EgoPose模型类，使用PostgreSQL专属Insert语句实现插入：

session.execute(Insert(EgoPose).on_conflict_do_nothing(), dataset.ego_pose)  # dataset.ego_pose是待插入的字典列表

问题 #

插入2631083行数据时，SQLAlchemy方案耗时显著更长：

• psycopg2方案耗时：2分19秒
• SQLAlchemy方案耗时：3分16秒

补充信息 #

1. 表创建语句：

CREATE TABLE IF NOT EXISTS ego_pose (
    token UUID NOT NULL DEFAULT uuid_generate_v4(),
    translation NUMERIC(15, 10) [3] NOT NULL,
    rotation NUMERIC(15, 10) [4] NOT NULL,
    timestamp BIGINT NOT NULL,
    PRIMARY KEY (token)
);

2. SQLAlchemy模型定义：

class EgoPose(Base):
    __tablename__ = "ego_pose"
    token: Mapped[str] = mapped_column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4, nullable=False)
    translation: Mapped[List[float]] = mapped_column(ARRAY(Numeric(20, 10)), nullable=False)
    rotation: Mapped[List[float]] = mapped_column(ARRAY(Numeric(20, 10)), nullable=False)
    timestamp: Mapped[int] = mapped_column(BigInteger, nullable=False)

    def __repr__(self) -> str:
        return f"<EgoPose(token={self.token}>"

3. 测试细节：

• 已尝试调整insertmanyvalues_page_size参数，无耗时差异
• dataset.ego_pose可O(1)返回待插入字典列表
• 已排除其他表操作影响，耗时集中在该批量插入步骤

优化方案 #

1. 复用psycopg2原生高效逻辑 #

通过SQLAlchemy获取底层psycopg2连接，直接沿用原批量插入代码，兼顾ORM框架和原生驱动性能：

from sqlalchemy import text
from psycopg2.extras import execute_values

# 从SQLAlchemy会话获取原生连接
conn = session.connection().connection
try:
    sql = """INSERT INTO ego_pose(token, translation, rotation, timestamp)
        VALUES %s ON CONFLICT DO NOTHING;"""
    with conn.cursor() as cur:
        sql_args_batch = [
            (
                ego_pose["token"],
                ego_pose["translation"],
                ego_pose["rotation"],
                ego_pose["timestamp"],
            )
            for ego_pose in dataset.ego_pose
        ]
        execute_values(cur, sql, sql_args_batch)
    session.commit()
finally:
    # 无需手动关闭连接，由SQLAlchemy管理生命周期
    pass

2. 调整SQLAlchemy批量插入参数 #

（1）开启fast_executemany优化

SQLAlchemy 2.0+针对psycopg2提供了批量操作优化选项，开启后可大幅降低执行开销：

insert_stmt = Insert(EgoPose).on_conflict_do_nothing()
session.execute(
    insert_stmt,
    dataset.ego_pose,
    execution_options={"fast_executemany": True}
)
session.commit()

（2）显式指定插入字段

避免SQLAlchemy自动推导字段映射，减少额外计算：

insert_stmt = Insert(EgoPose).values(
    token=EgoPose.token,
    translation=EgoPose.translation,
    rotation=EgoPose.rotation,
    timestamp=EgoPose.timestamp
).on_conflict_do_nothing()

session.execute(insert_stmt, dataset.ego_pose)
session.commit()

3. 对齐数据类型定义 #

模型中translation和rotation的Numeric精度（20,10）与数据库定义（15,10）不匹配，会触发额外类型转换开销，调整模型对齐数据库：

translation: Mapped[List[float]] = mapped_column(ARRAY(Numeric(15, 10)), nullable=False)
rotation: Mapped[List[float]] = mapped_column(ARRAY(Numeric(15, 10)), nullable=False)

4. 临时禁用ORM额外机制 #

批量插入时，关闭自动刷新和ORM事件监听，避免不必要的开销：

# 关闭自动刷新
session.autoflush = False
try:
    session.execute(Insert(EgoPose).on_conflict_do_nothing(), dataset.ego_pose)
    session.commit()
finally:
    session.autoflush = True

若存在自定义before_insert等事件，可临时移除：

from sqlalchemy import event

# 临时移除before_insert事件监听
event.remove(EgoPose, "before_insert", None)
try:
    session.execute(Insert(EgoPose).on_conflict_do_nothing(), dataset.ego_pose)
    session.commit()
finally:
    # 恢复事件监听（若需保留）
    pass

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