使用Peewee批量更新SQLite百万级数据遇性能问题求优化

问题背景 #

我有一个包含2000万行数据的SQLite3数据库表，需要更新所有行的部分列值。使用Python的peewee模块操作时遇到严重性能问题：初始每秒仅能处理约1000行，半小时后处理速度降至每秒250行，完成全部更新需耗时超24小时。

我尝试了两种批量更新方案，但都未解决问题：

方案1：使用islice遍历游标（内存溢出） #

import math, itertools
from tqdm import tqdm
from cool_project.database import db, MyTable

def update_row(row):
    row.column_a = computation(row.column_d)
    row.column_b = computation(row.column_d)
    row.column_c = computation(row.column_d)

fields = (MyTable.column_a
          MyTable.column_b
          MyTable.column_c)

rows = MyTable.select()
total_rows = rows.count()
page_size = 1000
total_pages = math.ceil(total_rows / page_size)
# Start #
with db.atomic():
    for page_num in tqdm(range(total_pages)):
        page = list(itertools.islice(rows, page_size))
        for row in page: update_row(row)
        MyTable.bulk_update(page, fields=fields)

该方案失败，原因是会将全量查询结果加载至内存，导致内存溢出。

方案2：使用paginate分页（性能依旧极差） #

import math
from tqdm import tqdm
from cool_project.database import db, MyTable

def update_row(row):
    row.column_a = computation(row.column_d)
    row.column_b = computation(row.column_d)
    row.column_c = computation(row.column_d)

fields = (MyTable.column_a
          MyTable.column_b
          MyTable.column_c)

rows = MyTable.select()
total_rows = rows.count()
page_size = 1000
total_pages = math.ceil(total_rows / page_size)
# Start #
with db.atomic():
    for page_num in tqdm(range(1, total_pages+1)):
        # Get a batch #
        page = MyTable.select().paginate(page_num, page_size)
        # Update #
        for row in page: update_row(row)
        # Commit #
        MyTable.bulk_update(page, fields=fields)

性能依旧很差，且处理速度随时间明显下降，想知道是否遗漏了关键优化点？

核心问题与优化方案 #

1. 先解决SQLite分页的本质性能坑 #

SQLite的OFFSET分页（peewee的paginate底层依赖此逻辑）在数据量超大时会急剧变慢——每翻一页，数据库都需要从头扫描到OFFSET指定的位置，2000万行的场景下，后续页面的扫描成本呈指数级上升，这就是处理速度越来越慢的核心原因。

2. 替代方案：用主键/唯一有序列做范围扫描分页 #

放弃页码+OFFSET的方式，改用主键（或唯一有序列）的范围查询分批取数据，比如按id从小到大，每次取id > last_id的前N行：

from tqdm import tqdm
from peewee import fn
from cool_project.database import db, MyTable

def process_batch(batch):
    updated_rows = []
    for row in batch:
        row.column_a = computation(row.column_d)
        row.column_b = computation(row.column_d)
        row.column_c = computation(row.column_d)
        updated_rows.append(row)
    MyTable.bulk_update(updated_rows, fields=[MyTable.column_a, MyTable.column_b, MyTable.column_c])

# 调大批次大小，SQLite对大批次更新支持良好，可根据内存调整
batch_size = 10000
last_id = 0

# 获取最大id作为终止判断条件
max_id = MyTable.select(fn.MAX(MyTable.id)).scalar()

with db.atomic():
    with tqdm(total=max_id) as pbar:
        while last_id < max_id:
            # 用范围查询取数，避免OFFSET的性能损耗
            batch = list(
                MyTable.select()
                .where(MyTable.id > last_id)
                .order_by(MyTable.id)
                .limit(batch_size)
                .iterator()  # 禁用对象缓存，减少内存占用
            )
            if not batch:
                break
            process_batch(batch)
            # 更新last_id为当前批次最后一行的id
            last_id = batch[-1].id
            pbar.update(len(batch))

3. 必须做的额外优化 #

• 调大batch_size：原方案用1000太小，SQLite批量更新的开销主要在事务和IO，批次越大，平均每行的开销越低。可先尝试1万，再根据内存占用调整至5万甚至10万（只要内存足够）。
• 优化SQLite连接参数：给SQLite设置关键参数提升写入性能：

  # 初始化数据库连接时配置
  db.init('your_database.db', pragmas={
      'journal_mode': 'WAL',  # 写前日志，大幅提升写入性能
      'synchronous': 'NORMAL',  # 降低同步级别（若可接受极小的崩溃风险）
      'cache_size': -2000000,  # 分配2GB内存缓存（负数单位为KB，根据自身内存调整）
      'temp_store': 'MEMORY',  # 临时表存入内存，减少磁盘IO
  })
  

• 将计算逻辑移至数据库端（如果可行）：如果computation函数是简单的数值计算或字符串处理，直接用SQL表达式更新，无需将数据拉到Python层处理，这是最快的方式：

  # 示例：假设computation是column_d乘以2
  MyTable.update(
      column_a=MyTable.column_d * 2,
      column_b=MyTable.column_d * 2,
      column_c=MyTable.column_d * 2
  ).execute()
  

• 优化computation函数：如果该函数是纯Python实现且耗时占比高，可考虑用C扩展（如Cython）或NumPy加速，减少单条数据的处理时间。

4. 原方案失败的原因 #

• 方案1的islice：MyTable.select()返回的查询对象，用islice遍历仍会逐步将全量数据加载到内存，游标持续保持导致内存占用不断升高，最终溢出。
• 方案2的paginate：底层依赖LIMIT ... OFFSET ...，当页码过大时，OFFSET需要扫描数百万行才能定位到目标数据，性能随页码增加急剧下降。

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