Knex.js内存优化：百万级CSV数据导入SQLite的高效方案

你的原始代码会把200万行数据全部加载到内存中，这会导致极高的内存占用，而且一次性批量插入也可能触发SQLite的性能瓶颈。之前在data回调中直接调用插入出现错误，本质是流式读取的速度远快于数据库插入速度，导致大量并发插入请求堆积——而SQLite是单写数据库，无法处理并行写入操作，进而引发锁冲突或超时错误。

下面是内存高效且避免并发问题的优化实现：

async function insertData(filePath: string, tableName: string) {
  const BATCH_SIZE = 1000; // 可根据SQLite性能调整，建议500-2000之间
  let batch: CsvRow[] = [];
  let streamPaused = false;

  await new Promise<void>((resolve, reject) => {
    const stream = fs.createReadStream(filePath)
      .pipe(fastcsv.parse({ headers: true }))
      .on("data", async (row: CsvRow) => {
        if (streamPaused) return;
        
        batch.push(row);
        // 攒够批次就开始插入
        if (batch.length >= BATCH_SIZE) {
          stream.pause();
          streamPaused = true;
          try {
            await db.batchInsert(tableName, batch, BATCH_SIZE);
            console.log(`Inserted ${BATCH_SIZE} rows`);
            batch = []; // 清空批次
            streamPaused = false;
            stream.resume(); // 恢复流读取
          } catch (error) {
            console.error("Batch insert failed:", error);
            stream.destroy(); // 终止流
            reject(error);
          }
        }
      })
      .on("end", async () => {
        // 插入剩余不足一个批次的行
        if (batch.length > 0) {
          try {
            await db.batchInsert(tableName, batch, batch.length);
            console.log(`Inserted remaining ${batch.length} rows`);
          } catch (error) {
            console.error("Final batch insert failed:", error);
            reject(error);
            return;
          }
        }
        console.log("All data inserted successfully");
        resolve();
      })
      .on("error", (error) => {
        console.error("Stream error:", error);
        reject(error);
      });
  });
}

关键优化点 #

• 流式分批处理：不再一次性加载所有数据，内存仅保留当前批次的行，内存占用大幅降低
• 串行批次插入：每批次插入完成后再读取下一批，彻底避免SQLite的并发写入冲突
• 流控（pause/resume）：防止流的读取速度超过数据库插入速度，避免内存溢出

额外性能优化建议 #

• 调整批次大小：如果SQLite写入慢，可适当减小批次；内存允许的话，增大批次能减少数据库交互次数，提升整体速度
• 开启SQLite WAL模式：预写日志模式支持更高的并发读写性能，能显著提升批量插入速度：

  await db.raw("PRAGMA journal_mode = WAL;");
  

• 临时关闭约束与索引：插入前关闭外键检查和索引，插入完成后恢复，避免每次插入都更新索引的开销：

  // 插入前执行
  await db.raw("PRAGMA foreign_keys = OFF;");
  // 插入完成后恢复
  await db.raw("PRAGMA foreign_keys = ON;");
  

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