大量SQLite数据库my_table按my_index求和的高效实现及通用原则问询

一、最优实现方案 #

方案1：单库增量累加（无中间表，低资源占用） #

直接在汇总数据库中维护目标表，遍历每个源数据库时，通过ATTACH临时挂载源库，用SQLite的冲突处理语法完成增量累加，全程无需将数据加载到Python内存。

代码示例：

import sqlite3
from pathlib import Path

# 初始化汇总数据库与目标表
summary_db = "summary.db"
with sqlite3.connect(summary_db) as conn:
    conn.execute("""
    CREATE TABLE IF NOT EXISTS my_nice_table (
        my_index INTEGER PRIMARY KEY,
        value REAL DEFAULT 0
    )
    """)

# 遍历所有源数据库文件（假设都在source_dbs目录下）
source_db_dir = "./source_dbs"
for idx, db_file in enumerate(Path(source_db_dir).glob("*.db")):
    # 给当前挂载的源库起唯一别名
    db_alias = f"src_{idx}"
    with sqlite3.connect(summary_db) as conn:
        # 挂载源数据库
        conn.execute(f"ATTACH DATABASE '{db_file.resolve()}' AS {db_alias}")
        # 增量累加：存在则更新，不存在则插入
        conn.execute(f"""
        INSERT INTO my_nice_table (my_index, value)
        SELECT my_index, value FROM {db_alias}.my_table
        ON CONFLICT(my_index) DO UPDATE SET value = value + excluded.value
        """)
        # 卸载源数据库
        conn.execute(f"DETACH DATABASE {db_alias}")

方案2：分批次批量处理（更高效率） #

利用SQLite的ATTACH批量挂载多个源库（默认挂载上限为10，因此每次挂载9个），通过UNION ALL合并批量数据后一次性累加，减少重复挂载/卸载的开销。

代码示例：

import sqlite3
from pathlib import Path

summary_db = "summary.db"
source_db_dir = "./source_dbs"
batch_size = 9  # 匹配默认ATTACH数量上限

# 初始化汇总表
with sqlite3.connect(summary_db) as conn:
    conn.execute("""
    CREATE TABLE IF NOT EXISTS my_nice_table (
        my_index INTEGER PRIMARY KEY,
        value REAL DEFAULT 0
    )
    """)

# 获取所有源数据库路径并分批次处理
source_dbs = list(Path(source_db_dir).glob("*.db"))
for batch_start in range(0, len(source_dbs), batch_size):
    batch = source_dbs[batch_start : batch_start + batch_size]
    with sqlite3.connect(summary_db) as conn:
        attach_cmds = []
        union_queries = []
        # 生成挂载命令与合并查询语句
        for idx, db_file in enumerate(batch):
            alias = f"batch_{idx}"
            attach_cmds.append(f"ATTACH DATABASE '{db_file.resolve()}' AS {alias}")
            union_queries.append(f"SELECT my_index, value FROM {alias}.my_table")
        
        # 执行挂载
        for cmd in attach_cmds:
            conn.execute(cmd)
        
        # 批量累加数据
        union_sql = " UNION ALL ".join(union_queries)
        conn.execute(f"""
        INSERT INTO my_nice_table (my_index, value)
        SELECT my_index, SUM(value) FROM ({union_sql}) GROUP BY my_index
        ON CONFLICT(my_index) DO UPDATE SET value = value + excluded.value
        """)
        
        # 卸载所有批量挂载的数据库
        for idx in range(len(batch)):
            conn.execute(f"DETACH DATABASE batch_{idx}")

二、大量同结构数据库处理的通用原则 #

1. 优先数据库层操作：避免将数据加载到Python（如pandas），利用SQLite原生语法处理数据，减少序列化/反序列化与IO开销。
2. 增量更新替代全量合并：直接维护最终目标表，每次仅处理单个/批量源库的增量数据，避免创建庞大的中间临时表。
3. 利用SQLite冲突处理：通过ON CONFLICT实现原子性的插入/更新，无需手动处理重复索引逻辑，安全且高效。
4. 分批次平衡效率与资源：根据工具限制（如ATTACH数量）分批次处理，减少单次操作资源占用，降低重复操作开销。
5. 索引优化：给目标表的关联字段（如my_index）设置主键或唯一索引，大幅提升数据查找与更新效率。
6. 调整性能参数：根据数据安全需求，调整synchronous（如设为NORMAL）、journal_mode（如设为WAL）等参数，提升写入速度。

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