将大 CSV 流式导入数据库而不全量加载

解析而非加载

流式解析器逐行 emit，内存恒定。把文件管道给解析器，任一时刻内存中不超过一个批次。

import { parse } from 'csv-parse';
fs.createReadStream('export.csv').pipe(parse({ columns: true }))

批量提交

每 1000 行一组准备插入并独立提交。脚本崩溃时从最近提交的批次恢复。用 ON CONFLICT DO NOTHING 保证幂等，重试不会产生重复行。

for await (const batch of batches(rows, 1000)) {
  await client.query('BEGIN');
  for (const r of batch) await stmt.execute([r.id, r.value]);
  await client.query('COMMIT');
}

预期结果

导入内存恒定，每秒提交数个批次，按字节偏移与行数报告进度。崩溃后从最近提交恢复，坏行被记录跳过而非中断整次运行。导入时长与文件大小成正比，应安排在非高峰时段。