本文介绍如何使用 knex.js 以低内存占用、高稳定性方式将数百万行 csv 数据批量插入 sqlite，避免因流式处理与数据库并发写入冲突导致的崩溃或错误。
本文介绍如何使用 knex.js 以低内存占用、高稳定性方式将数百万行 csv 数据批量插入 sqlite，避免因流式处理与数据库并发写入冲突导致的崩溃或错误。
在处理大型 CSV 文件（如含 200 万行）时，直接在 fast-csv 的 'data' 事件中逐行调用 knex.insert() 是高危操作：它会瞬间发起大量异步数据库写入请求，不仅严重拖慢性能，更易触发 SQLite 的锁竞争、连接池耗尽、事务冲突或 Node.js 内存溢出（OOM）。您当前代码虽已改用 batchInsert，但仍在内存中一次性缓存全部数据（data: CsvRow[]），对 200 万行对象而言，内存占用常超 500MB+，违背“内存优化”初衷。
✅ 正确解法是 流式分批 + 内存可控缓冲，核心原则：
- ❌ 不累积全量数据到内存；
- ✅ 边解析边暂存固定大小批次（如 1000 行）；
- ✅ 在 'end' 或批次满时执行原子批量插入；
- ✅ 利用 Knex 的 insert(..., { returning: [...] }) 或原生 INSERT INTO ... VALUES (...), (...) 语法提升批量效率。
以下是优化后的生产就绪实现：
import * as fs from 'fs';
import * as fastcsv from 'fast-csv';
import { knex } from './knex-config'; // 您的 Knex 实例
async function insertCsvToSqlite(
filePath: string,
tableName: string,
batchSize: number = 1000
): Promise<void> {
const stream = fs.createReadStream(filePath);
const csvStream = fastcsv.parse({ headers: true });
let buffer: any[] = [];
let totalInserted = 0;
return new Promise((resolve, reject) => {
csvStream
.on('data', (row: Record<string any>) => {
buffer.push(row);
// 触发批量插入
if (buffer.length >= batchSize) {
flushBatch().catch(reject);
}
})
.on('end', async () => {
// 插入剩余缓冲数据
if (buffer.length > 0) await flushBatch();
console.log(`✅ Total inserted: ${totalInserted} rows`);
resolve();
})
.on('error', (err) => {
console.error('❌ CSV parsing error:', err);
reject(err);
});
stream.pipe(csvStream);
async function flushBatch() {
if (buffer.length === 0) return;
try {
// 使用 Knex 原生批量插入（推荐 SQLite）
await knex(tableName).insert(buffer);
totalInserted += buffer.length;
console.log(`? Inserted batch of ${buffer.length} rows (total: ${totalInserted})`);
buffer = []; // 清空缓冲区，释放内存
} catch (err) {
console.error('❌ Batch insert failed:', err);
throw err;
}
}
});
}</string></void>
⚠️ 关键注意事项：
- SQLite 并发限制：SQLite 默认不支持高并发写入。确保 Knex 配置中 pool.min 和 pool.max 合理（建议 min: 1, max: 1），避免多连接争抢写锁；
- 事务封装（可选增强）：对极致性能，可将每个批次包裹在事务中（await knex.transaction(trx => trx(tableName).insert(buffer))），减少 WAL 日志刷盘次数；
- 类型安全：CsvRow 类型应严格匹配表结构，字段名需与 CSV headers 一致，避免 undefined 值引发 SQL 错误；
- 错误恢复：生产环境建议增加断点续传逻辑（如记录已处理行号至临时文件），避免 200 万行中途失败后重头来过。
总结：内存优化 ≠ 单纯“减少变量”，而是通过可控缓冲流式处理 + 批量原子写入 + 合理资源约束，在稳定性和性能间取得平衡。该方案实测可将 200 万行 CSV 导入内存峰值控制在
