Python百万级数据拼接API返回慢，求执行提速解决方案

环境信息 #

• Python 3.9.2
• MariaDB 10.5
• Debian

原始代码 #

sql = (
    "SELECT team.teamid, team.level, score.id AS scorekey, score.points, score.timestamp, score.`update`, score.lid, score.data, score.userid "
    "FROM score INNER JOIN music ON team.id = score.teamid "
    "WHERE music.type = :type"
)
# Add version, songid, and chart to the WHERE clause if specified
where_clauses = []
params = {
    "type": type.value,
}
if class is not None:
    where_clauses.append("team.class = :class")
    params["class"] = version
if where_clauses:
    sql += " AND " + " AND ".join(where_clauses)

# Execute the query
cursor = self.execute(sql, params)
return [
    (
        UserID(result["userid"]),
        Score(
            result["scorekey"],
            result["teamid"],
            result["level"],
            result["points"],
            result["timestamp"],
            result["update"],
            result["lid"],
            result["counts"],  # 注：原SQL未查询counts字段，属于代码笔误
            self.deserialize(result["data"]),
        ),
    )
    for result in cursor.fetchall()
]

耗时分析 #

总耗时：8.27s

• SQL执行：3.03s
• 结果处理：5.24s
  数据量：11万条（目标为近1000万条）

优化方案 #

1. 用namedtuple替换自定义数据类 #

如果UserID和Score只是简单的数据容器（无复杂业务逻辑），用collections.namedtuple替代普通类，能大幅降低实例化的内存开销与时间成本：

from collections import namedtuple

# 提前定义轻量数据容器
UserID = namedtuple('UserID', ['userid'])
Score = namedtuple('Score', ['scorekey', 'teamid', 'level', 'points', 'timestamp', 'update', 'lid', 'counts', 'data'])

如果类中存在业务方法，建议将方法剥离为独立函数，保持数据容器的轻量化。

2. 核心优化deserialize方法 #

self.deserialize(result["data"])是单条数据处理的最大耗时点，优先针对性优化：

• 若data为JSON格式，用ujson替代标准库json（解析速度提升2-5倍）；
• 将反序列化的固定逻辑（如格式校验规则）提前提取到循环外，避免重复执行；
• 若data结构固定，用dataclasses配合ujson的object_hook做结构化解析，比手动字段映射更快。

3. 改用服务器端游标+分批拉取 #

不要用fetchall()一次性加载所有数据，尤其是千万级数据会导致内存溢出与处理延迟。改用MariaDB的SSCursor服务器端游标，分批fetchmany()处理：

from mariadb.cursors import SSCursor

cursor = self.execute(sql, params, cursorclass=SSCursor)
batch_size = 10000  # 根据内存情况调整批次大小
result_list = []
deserialize = self.deserialize
create_userid = UserID
create_score = Score

while True:
    batch = cursor.fetchmany(batch_size)
    if not batch:
        break
    # 用元组索引替代字典键访问，速度提升明显
    for r in batch:
        result_list.append(
            (
                create_userid(r[8]),
                create_score(
                    r[2], r[0], r[1], r[3],
                    r[4], r[5], r[6], None,  # 对应原代码中缺失的counts字段，若需此字段请补充SQL查询
                    deserialize(r[7])
                )
            )
        )
return result_list

4. 减少循环内的属性查找开销 #

将循环内频繁调用的方法、类提前绑定到局部变量，避免每次循环都做实例属性/全局变量查找：

# 提前绑定到局部变量，加速查找
deserialize = self.deserialize
create_userid = UserID
create_score = Score

return [
    (
        create_userid(r[8]),
        create_score(r[2], r[0], r[1], r[3], r[4], r[5], r[6], None, deserialize(r[7]))
    )
    for r in cursor
]

5. 修复counts字段的笔误 #

原SQL未查询counts字段，但代码中尝试读取result["counts"]，这会导致每次循环都触发字典键不存在的检查逻辑。若此字段是必需的，补充到SELECT语句中；若不需要，直接删除该参数，避免不必要的性能损耗。

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