优化超大CSV文件读取及写入SQLite的性能问题

处理10GB级别的CSV导入SQLite确实是个头疼的问题，逐行插入的方式肯定会慢到让人崩溃——你现在1MB要10秒，10GB的话简直不敢想。下面给你几个实用的优化方案，亲测能大幅提升导入速度：

1. 用 executemany 批量插入（最立竿见影的优化） #

你现在逐行调用execute，每次都要和数据库交互一次，开销极大。换成批量插入，把几十上百条数据攒成一个列表一次性提交，能把速度提升几十倍甚至上百倍。

修改后的代码示例：

import sqlite3
import csv

path = 'genders.csv'
user_table = 'Users'
batch_size = 10000  # 可根据内存调整，比如1万条一批

conn = sqlite3.connect('db.sqlite')
cur = conn.cursor()

# 开启WAL模式，大幅提升写入性能
conn.execute('PRAGMA journal_mode=WAL;')
# 降低同步级别（如果对导入过程中的数据安全性要求不高的话）
conn.execute('PRAGMA synchronous=NORMAL;')

cur.execute(f'''DROP TABLE IF EXISTS {user_table}''')
cur.execute(f'''CREATE TABLE {user_table} (
userID INTEGER NOT NULL,
gender INTEGER,
PRIMARY KEY (userID))''')

with open(path) as csvfile:
    datareader = csv.reader(csvfile)
    next(datareader, None)  # 跳过表头
    batch_data = []
    for line in datareader:
        # 转换gender字段
        gender = 1 if line[1] == 'f' else 0
        batch_data.append((int(line[0]), gender))
        # 攒够一批就提交
        if len(batch_data) >= batch_size:
            cur.executemany(
                f'''INSERT OR IGNORE INTO {user_table} (userID, gender) VALUES (?, ?)''',
                batch_data
            )
            conn.commit()
            batch_data = []
    # 处理最后一批剩余数据
    if batch_data:
        cur.executemany(
            f'''INSERT OR IGNORE INTO {user_table} (userID, gender) VALUES (?, ?)''',
            batch_data
        )
        conn.commit()

conn.close()

这里的关键细节：

• 用占位符?代替字符串拼接，既安全又减少SQL解析开销
• WAL模式是SQLite写入性能提升的核心开关
• 批量大小可根据内存灵活调整，内存充足的话设成10万条能进一步减少提交次数

2. 用命令行工具导入（速度最快的方案） #

如果数据预处理逻辑不复杂，直接用SQLite的命令行工具.import导入，速度会比Python代码快很多——因为它直接和SQLite内核交互，没有Python中间层的开销。

步骤如下：

1. 先快速预处理CSV，把gender字符串转成数字：

import csv

input_path = 'genders.csv'
output_path = 'genders_processed.csv'

with open(input_path, 'r') as infile, open(output_path, 'w', newline='') as outfile:
    reader = csv.reader(infile)
    writer = csv.writer(outfile)
    # 写入表头
    writer.writerow(next(reader))
    for line in reader:
        line[1] = '1' if line[1] == 'f' else '0'
        writer.writerow(line)

2. 执行命令行导入：

sqlite3 db.sqlite "PRAGMA journal_mode=WAL; DROP TABLE IF EXISTS Users; CREATE TABLE Users (userID INTEGER NOT NULL, gender INTEGER, PRIMARY KEY (userID)); .import --csv genders_processed.csv Users"

3. 关于pandas的误区：其实你可以用pd.to_sql！ #

你说因为主键设置无法使用pd.to_sql，其实是可以的。只要配合SQLAlchemy实现批量插入，再加上INSERT OR IGNORE的逻辑，完全能高效完成导入。

示例代码：

import pandas as pd
from sqlalchemy import create_engine, insert

path = 'genders.csv'
user_table = 'Users'

# 创建SQLAlchemy引擎，开启WAL模式
engine = create_engine('sqlite:///db.sqlite?journal_mode=WAL')

# 自定义插入逻辑，实现INSERT OR IGNORE
def sqlite_insert_ignore(table, conn, keys, data_iter):
    insert_stmt = insert(table.table).values(list(data_iter))
    ignore_stmt = insert_stmt.on_conflict_do_nothing(index_elements=['userID'])
    conn.execute(ignore_stmt)

# 分块读取CSV，避免内存溢出
chunk_size = 100000
for chunk in pd.read_csv(path, chunksize=chunk_size):
    # 转换gender字段
    chunk['gender'] = chunk['gender'].map({'f':1, 'm':0})
    # 批量导入数据库
    chunk.to_sql(
        user_table,
        engine,
        if_exists='append',
        index=False,
        method=sqlite_insert_ignore,
        dtype={'userID': 'INTEGER', 'gender': 'INTEGER'}
    )

这个方案既利用了pandas高效的CSV解析能力，又实现了主键冲突忽略，速度远胜逐行插入。

额外优化小贴士 #

• 导入前关闭非必要的索引和约束（除主键外），导入完成后再重建，减少写入时的索引维护开销
• 把数据库文件放在SSD上，磁盘性能的提升会直接反映在导入速度上
• 增大SQLite缓存：conn.execute('PRAGMA cache_size=1000000;')（单位是页，默认每页4KB，这里设为4GB缓存）

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