优化1亿行SQLite数据库的导入与查询效率问题求助

针对你提到的两个核心问题，以下是具体的优化实现方案：

一、数据导入优化（解决千万级数据导入耗时过长） #

原导入脚本的核心问题是逐行执行INSERT语句，每一次插入都要和数据库进行IO交互，累计开销巨大。优化方案如下：

• 开启批量插入，使用executemany替代循环内的execute
• 增大事务提交的批次（比如每1万条提交一次），减少事务提交次数
• 提前处理文件内容，减少循环内的重复字符串操作
• 关闭SQLite的自动提交，手动控制事务

修改后的导入脚本：

import sqlite3

def batch_insert(file_path, batch_size=10000):
    conn = sqlite3.connect('Main_Database.db')
    conn.execute('PRAGMA synchronous = OFF')  # 关闭同步，提升写入速度（需确保数据安全）
    conn.execute('PRAGMA journal_mode = WAL')  # 启用WAL模式，优化并发写入
    c = conn.cursor()
    
    batch_data = []
    with open(file_path, 'r') as f:
        for line in f:
            line = line.strip()
            if not line:
                continue
            try:
                l = line.split(':')
                if len(l) != 2:
                    continue
                batch_data.append((l[0], l[1]))
                # 达到批次大小就提交
                if len(batch_data) >= batch_size:
                    c.executemany("INSERT OR IGNORE INTO user_data VALUES (?,?)", batch_data)
                    conn.commit()
                    batch_data = []
            except Exception as e:
                print(f"处理行出错: {line}, 错误: {e}")
                continue
    # 提交剩余数据
    if batch_data:
        c.executemany("INSERT OR IGNORE INTO user_data VALUES (?,?)", batch_data)
        conn.commit()
    
    conn.close()

if __name__ == "__main__":
    batch_insert('my_list_2.txt')

二、查询优化（解决全表加载耗时过长） #

原查询脚本的核心问题是先把全表数据加载到内存再过滤，1亿行数据完全无法在内存中处理，而且完全浪费了SQLite的查询优化能力。优化方案如下：

• 将三种查询逻辑直接用SQL语句实现，让数据库引擎完成过滤
• 为user_data表的第一列（邮箱列）创建索引，加速查询
• 不需要加载全表，直接从数据库获取过滤后的结果
• 流式处理查询结果，减少内存占用

第一步：先为邮箱列创建索引（只需执行一次） #

import sqlite3

conn = sqlite3.connect('Main_Database.db')
c = conn.cursor()
# 创建唯一索引（如果邮箱是唯一的），普通索引用CREATE INDEX
c.execute("CREATE UNIQUE INDEX IF NOT EXISTS idx_user_data_email ON user_data (column1)")
conn.commit()
conn.close()

注意：将column1替换为你实际的邮箱列名

修改后的查询脚本 #

import sqlite3
from datetime import datetime, date
import csv

def export_results(results, method, query_params):
    with open('saved_emails.csv', 'a', encoding='UTF8', newline='') as f:
        writer = csv.writer(f)
        now = datetime.now()
        current_time = now.strftime("%H:%M:%S")
        today = date.today()
        d1 = today.strftime("%d/%m/%Y")
        writer.writerow([f"********* report {d1} {current_time} **************"])
        writer.writerow([f"method used for this query is: {method}"])
        writer.writerow([f"query parameters: {query_params}"])
        writer.writerow(["emails obtained are: "])
        for row in results:
            writer.writerow([row[0]])

def query_database():
    conn = sqlite3.connect('Main_Database.db')
    c = conn.cursor()
    
    # 选择查询方式
    method = 'method 3'
    results = []
    
    if method == 'method 1':
        exact_match_mail = 'abc123'
        c.execute("SELECT * FROM user_data WHERE column1 = ?", (exact_match_mail,))
        results = c.fetchall()
        for row in results:
            print('找到精确匹配的邮箱: ', row[0])
        export_results(results, method, f"精确匹配: {exact_match_mail}")
    
    elif method == 'method 2':
        word_match_mail = 'hot'
        # 拆分邮箱为@和_分隔的部分，用LIKE匹配
        c.execute("""
            SELECT * FROM user_data 
            WHERE column1 LIKE ? OR column1 LIKE ? OR column1 LIKE ?
        """, (f"%_{word_match_mail}_%", f"%_{word_match_mail}@%", f"{word_match_mail}_%"))
        results = c.fetchall()
        for row in results:
            print('找到包含指定单词的邮箱: ', row[0])
        export_results(results, method, f"包含单词: {word_match_mail}")
    
    elif method == 'method 3':
        word1 = 'hot'
        word2 = 'chocolate'
        # 匹配同时包含两个单词的邮箱
        c.execute("""
            SELECT * FROM user_data 
            WHERE (column1 LIKE ? OR column1 LIKE ? OR column1 LIKE ?)
            AND (column1 LIKE ? OR column1 LIKE ? OR column1 LIKE ?)
        """, (
            f"%_{word1}_%", f"%_{word1}@%", f"{word1}_%",
            f"%_{word2}_%", f"%_{word2}@%", f"{word2}_%"
        ))
        results = c.fetchall()
        for row in results:
            print('找到同时包含两个单词的邮箱: ', row[0])
        export_results(results, method, f"包含单词: {word1} 和 {word2}")
    
    conn.close()

if __name__ == "__main__":
    query_database()
    input('按任意键退出...')

额外优化建议 #

• 对于更复杂的邮箱拆分匹配，可以考虑使用SQLite的REGEXP函数（需要启用扩展），或者预先将邮箱拆分后的部分存储到单独的列中，进一步加速查询
• 若数据量持续增长到1亿行，考虑使用更适合大数据的数据库（如PostgreSQL），但SQLite通过WAL模式和索引优化也能支撑千万级到亿级数据的查询

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