使用SQLAlchemy分块读取数据解决查询遍历耗时过长问题

1 首要耗时原因：N+1查询问题 #

你当前代码最大的性能瓶颈来自关联属性的懒加载触发的N+1次SQL请求：

• 首次查询仅拉取所有Entities符合条件的主表数据
• 循环中访问entity.addresses、entity.relationships_parent两个一对多关联属性时，默认懒加载会为每个实体单独触发2次SQL查询，若存在10万条实体，会额外产生20万次网络请求，耗时会呈线性增长

解决方案：预加载关联数据 #

使用selectinload一次性批量加载所有关联数据，仅需额外2次SQL请求即可拉取全部关联的地址、关系数据，适合一对多关联场景，不会导致主表结果膨胀：

from sqlalchemy.orm import selectinload

# 预加载关联属性，避免循环中重复查询
entities = self.session.query(Entities)\
    .filter(Entities.parent_id == 0)\
    .options(
        selectinload(Entities.addresses),
        selectinload(Entities.relationships_parent)
    )

2 分块读取解决全量拉取延迟问题 #

如果实体总数据量超过10万条，全量拉取会导致单次网络传输包过大、等待时间长，可通过以下两种方式实现分块读取：

方案1：内置yield_per分块（简单易用，适合中等数据量） #

直接通过yield_per指定每次从数据库拉取的行数，处理完一批再自动拉下一批，无需手动维护分页参数：

from sqlalchemy.orm import selectinload

BATCH_SIZE = 1000 # 可根据实际场景调整，推荐500-2000
entities = self.session.query(Entities)\
    .filter(Entities.parent_id == 0)\
    .options(
        selectinload(Entities.addresses),
        selectinload(Entities.relationships_parent)
    )\
    .yield_per(BATCH_SIZE)

# 后续循环处理逻辑保持不变即可
index_data = {}
for entity in entities:
    data = entity.__dict__
    # 可手动删除SQLAlchemy内置冗余字段，减少后续存储开销
    data.pop("_sa_instance_state", None)
    data['addresses'] = []
    for address in entity.addresses:
        addr_dict = address.__dict__
        addr_dict.pop("_sa_instance_state", None)
        data['addresses'].append(addr_dict)
    data['relationships'] = []
    for rel in entity.relationships_parent:
        rel_dict = rel.__dict__
        rel_dict.pop("_sa_instance_state", None)
        data['relationships'].append(rel_dict)
    index_data[data['id']] = data

方案2：Keyset分页（高性能，适合百万级以上超大数据量） #

如果数据量极大，yield_per或普通limit+offset分页在偏移量过高时性能会大幅下降，可使用基于主键的Keyset分页，每次查询仅拉取上次处理完的最大ID之后的数据，性能稳定无衰减：

from sqlalchemy.orm import selectinload

BATCH_SIZE = 1000
index_data = {}
last_id = 0

while True:
    batch = self.session.query(Entities)\
        .filter(Entities.parent_id == 0)\
        .filter(Entities.id > last_id)\
        .options(
            selectinload(Entities.addresses),
            selectinload(Entities.relationships_parent)
        )\
        .order_by(Entities.id)\
        .limit(BATCH_SIZE)\
        .all()
    if not batch:
        break
    for entity in batch:
        # 原有处理逻辑保持不变
        data = entity.__dict__
        data.pop("_sa_instance_state", None)
        data['addresses'] = [addr.__dict__ for addr in entity.addresses]
        for addr in data['addresses']:
            addr.pop("_sa_instance_state", None)
        data['relationships'] = [rel.__dict__ for rel in entity.relationships_parent]
        for rel in data['relationships']:
            rel.pop("_sa_instance_state", None)
        index_data[data['id']] = data
    # 更新最后处理的ID
    last_id = batch[-1].id

3 其他可选优化点 #

• 若实体存在大文本、二进制等不需要的字段，可通过.options(load_only(Entities.id, Entities.name, ...))指定仅拉取需要的字段，减少数据传输量
• 不需要保留ORM对象特性的场景，可直接查询字典结构，避免ORM实例化的开销：self.session.query(Entities.id, Entities.xxx, ...).filter(...)，直接返回元组或字典，性能更高

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