You need to enable JavaScript to run this app.
优惠活动
大模型
产品
解决方案
定价
更多

如何实现数据库每100条记录取1条的高效查询

方案解答

你的原有实现存在非常明显的性能问题:OFFSET 原理是数据库先扫描并跳过指定数量的行,再返回后续结果,循环中 OFFSET 数值不断累加,当 OFFSET 到9万级别时,单次查询就要扫描9万行,整体查询耗时会随着循环次数增长指数级上升,十万条数据场景下完全不推荐生产使用。

下面是3种更高效的实现方案,按性能优先级排序:

方案1:窗口函数一次性查询(优先推荐)

仅需一次数据库查询,性能最高,兼容 MySQL 8.0+、PostgreSQL、SQL Server 等所有支持窗口函数的主流数据库。核心是用 ROW_NUMBER() 给每行按排序规则编号,直接筛选出符合间隔要求的行。
SQLAlchemy 实现示例:

from sqlalchemy import func

def get_one_paczka_per_n_paczek(db: Session, sesja_id: int, per_n: int):
    # 按主键排序生成行号,可根据业务需求替换为其他排序字段
    subquery = db.query(
        PaczkaDanych,
        func.row_number().over(order_by=PaczkaDanych.id).label('row_num')
    ).subquery()
    # 每per_n条取第一条,调整模运算的余数可以修改采样的起始位置
    result = db.query(subquery).filter(subquery.c.row_num % per_n == 1).all()
    # 转换为PaczkaDanych对象列表
    return [PaczkaDanych(**{k:v for k,v in row._asdict().items() if k != 'row_num'}) for row in result]

方案2:主键锚定分页查询

如果使用不支持窗口函数的低版本数据库,可以用主键代替大偏移量 OFFSET,每次仅固定跳过 per_n-1 行,不会出现偏移量过大的性能问题。
实现示例:

def get_one_paczka_per_n_paczek(db: Session, sesja_id: int, per_n: int):
    lista = []
    last_id = 0
    while True:
        # 基于上一次取到的主键定位,每次仅跳过per_n-1行
        element = db.query(PaczkaDanych)\
            .filter(PaczkaDanych.id > last_id)\
            .order_by(PaczkaDanych.id)\
            .offset(per_n - 1)\
            .first()
        if not element:
            break
        lista.append(element)
        last_id = element.id
    return lista

方案3:近似采样(适合非精准采样场景)

如果业务仅需要近似每N条取1条的抽样效果,不需要严格对齐行号,可以用数据库自带的随机采样函数,性能最高,适合百万级以上大数据量场景。
PostgreSQL 示例:

from sqlalchemy import func

def get_one_paczka_per_n_paczek(db: Session, sesja_id: int, per_n: int):
    # 随机保留1/per_n比例的行,近似实现每N条取1条
    return db.query(PaczkaDanych).filter(func.random() < 1/per_n).all()

内容的提问来源于stack exchange,提问作者Hamilkar Barkas

相关产品推荐
方舟 Agent Plan

超全模态模型 × Harness 升级,最新支持 Deepseek-V4.1-Flash、GLM-5.3 系列、Doubao-Seedream-5.0-pro、Kimi-K3 (部分), 限时 9.9 元起

最近更新时间:2026.10.02 06:39:03