如何高效合并SQLAlchemy ORM中同value_c值的两行数据？

我有一个含约40列的SQLAlchemy模型Model，每次需处理约50万行新数据且每次处理会创建新表。使用session.bulk_insert_mappings插入数据后，存在value_c值重复的行（最多2条）：一行有value_a值、value_b为空，另一行有value_b值、value_a为空。需要将这些同value_c的行合并为一行以去除重复，此前用session.merge逐行处理效率过低，现寻求分插入与更新阶段的高效实现方式。

模型定义：

class Model(Base):
    __tablename__ = "table"
    id = Column(Integer, primary_key=True)
    value_a = Column(String)
    value_b = Column(String)
    value_c = Column(String)
    # 其余约37列...

插入后数据示例：

合并后预期结果：

阶段1：批量插入数据 #

保持使用session.bulk_insert_mappings完成初始批量插入，这是SQLAlchemy中处理大量数据最高效的方式之一，避免ORM实例化的开销。

# 假设new_data是包含50万条数据的字典列表
session.bulk_insert_mappings(Model, new_data)
session.commit()

阶段2：批量合并重复行 #

由于重复行最多2条且仅value_a/value_b互补为空，直接用SQLAlchemy核心层执行批量SQL操作，比ORM逐行操作效率高几个数量级。

方案A：更新补全数据后删除重复行 #

1. 补全数据：通过聚合子查询获取每个value_c对应的非空值，批量更新到对应行

from sqlalchemy import update, func, select

# 子查询：按value_c分组，取每个分组的非空value_a和value_b（MAX忽略NULL）
subquery = select(
    Model.value_c,
    func.max(Model.value_a).label('merged_a'),
    func.max(Model.value_b).label('merged_b')
).group_by(Model.value_c).subquery()

# 批量更新主表数据
update_stmt = update(Model).values(
    value_a=subquery.c.merged_a,
    value_b=subquery.c.merged_b
).where(Model.value_c == subquery.c.value_c)

session.execute(update_stmt)
session.commit()

2. 删除重复行：每个value_c仅保留一行（这里选最小id，可根据需求调整）

# 子查询：获取每个value_c需要保留的id
keep_ids_subquery = select(func.min(Model.id)).group_by(Model.value_c).subquery()

# 删除不在保留列表中的行
delete_stmt = Model.__table__.delete().where(Model.id.not_in(keep_ids_subquery))

session.execute(delete_stmt)
session.commit()

方案B：插入合并数据到临时表，替换原表（适配每次创建新表的场景） #

因为每次处理都会创建新表，可直接生成合并后的数据到临时表，再替换原表，避免原表的更新/删除操作：

from sqlalchemy import func, select

# 创建与原表结构一致的临时表
temp_table = Model.__table__.tometadata(Base.metadata, name='temp_table')
temp_table.create(session.bind)

# 插入合并后的数据到临时表：按value_c分组，聚合补全所有列的非空值
insert_temp_stmt = temp_table.insert().from_select(
    # 指定所有列（除自增id）
    [col for col in Model.__table__.columns if col.name != 'id'],
    select(
        func.max(Model.value_a).label('value_a'),
        func.max(Model.value_b).label('value_b'),
        Model.value_c,
        # 其余列用MAX聚合取非空值，若需其他逻辑可替换聚合函数
        *[func.max(col).label(col.name) for col in Model.__table__.columns if col.name not in ['id', 'value_a', 'value_b', 'value_c']]
    ).group_by(Model.value_c)
)

session.execute(insert_temp_stmt)
session.commit()

# 删除原表，将临时表重命名为原表名
Model.__table__.drop(session.bind)
temp_table.rename(Model.__tablename__)
session.commit()

核心优化点 #

• 全程使用SQLAlchemy核心层操作，避免ORM实例化带来的性能损耗
• 利用数据库聚合函数MAX自动忽略NULL值，完美适配互补为空的重复行场景
• 所有操作仅需2-3条SQL语句，处理50万级数据的速度远快于逐行merge

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