Python操作sqlite3大数据批量处理可否使用executemany优化？

这个场景完全适配executemany优化方案。你当前性能低的核心原因是逐行执行SQL会触发大量事务提交、磁盘IO操作，executemany可以把批量操作合并为少量IO请求，通常能带来几十到上百倍的性能提升。

• 遍历源数据时不直接执行SQL，而是把需要UPDATE和INSERT的参数分别存入两个独立列表
• 全量数据遍历完成后，统一执行批量UPDATE、批量INSERT操作
• 显式管理SQLite事务，关闭默认自动提交，避免每次操作都触发磁盘同步

原低效逻辑参考 #

import sqlite3
conn = sqlite3.connect('your_database.db')
cur = conn.cursor()

# 获取当前批次时间范围
cur.execute("SELECT start_date, end_date FROM Crawl WHERE task_id = ?", (current_task_id,))
start_date, end_date = cur.fetchone()

# 逐行处理数据
cur.execute("SELECT url, crawl_date, ext_field FROM URL_Crawl WHERE crawl_date BETWEEN ? AND ?", (start_date, end_date))
for row in cur.fetchall():
    url, crawl_date, ext_field = row
    # 自定义字段计算逻辑
    max_date = max(crawl_date, other_calc_value)
    # 逐行判断URL是否存在
    cur.execute("SELECT 1 FROM Lifetime WHERE url = ?", (url,))
    exists = cur.fetchone()
    if exists:
        cur.execute("UPDATE Lifetime SET max_date = ?, ext_field = ? WHERE url = ?", (max_date, ext_field, url))
    else:
        cur.execute("INSERT INTO Lifetime (url, max_date, ext_field) VALUES (?, ?, ?)", (url, max_date, ext_field))

conn.commit()
conn.close()

优化后批量执行逻辑 #

import sqlite3
conn = sqlite3.connect('your_database.db')
cur = conn.cursor()
# 显式开启事务，关闭自动提交
conn.isolation_level = 'DEFERRED'

# 获取当前批次时间范围
cur.execute("SELECT start_date, end_date FROM Crawl WHERE task_id = ?", (current_task_id,))
start_date, end_date = cur.fetchone()

# 批量查询当前批次所有URL的存在性，避免逐行查库
cur.execute("SELECT url FROM URL_Crawl WHERE crawl_date BETWEEN ? AND ?", (start_date, end_date))
batch_urls = [row[0] for row in cur.fetchall()]
placeholders = ', '.join('?' for _ in batch_urls)
exists_urls = set(row[0] for row in cur.execute(f"SELECT url FROM Lifetime WHERE url IN ({placeholders})", batch_urls))

# 遍历源数据，攒批量参数
update_params = []
insert_params = []
cur.execute("SELECT url, crawl_date, ext_field FROM URL_Crawl WHERE crawl_date BETWEEN ? AND ?", (start_date, end_date))
for row in cur.fetchall():
    url, crawl_date, ext_field = row
    # 保留原有字段计算逻辑
    max_date = max(crawl_date, other_calc_value)
    if url in exists_urls:
        # 参数顺序和UPDATE语句占位符顺序严格对应
        update_params.append((max_date, ext_field, url))
    else:
        # 参数顺序和INSERT语句占位符顺序严格对应
        insert_params.append((url, max_date, ext_field))

# 批量执行操作
if update_params:
    cur.executemany("UPDATE Lifetime SET max_date = ?, ext_field = ? WHERE url = ?", update_params)
if insert_params:
    cur.executemany("INSERT INTO Lifetime (url, max_date, ext_field) VALUES (?, ?, ?)", insert_params)

# 统一提交事务
conn.commit()
conn.close()

• 如果单批次数据量超过10万条，可以按1万条为单位拆分参数列表，分批次执行executemany，避免内存占用过高
• 给Lifetime表的url字段加唯一索引，可直接用INSERT OR REPLACE语法简化逻辑：不需要提前判断URL存在性，所有数据统一攒为参数列表，直接执行cur.executemany("INSERT OR REPLACE INTO Lifetime (url, max_date, ext_field) VALUES (?, ?, ?)", all_params)，SQLite会自动完成存在则更新、不存在则插入的操作
• 重处理场景可临时关闭SQLite同步模式进一步提升性能：执行cur.execute("PRAGMA synchronous = OFF")，异常场景下存在丢数据风险，可按需开启

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