Node.js向Firestore导入60万+CSV数据的最优方案及内存泄漏解决

你当前的代码用Promise.all配合data.map一次性发起60万+个写请求，所有Promise同时处于pending状态，内存会被大量未完成的请求对象和数据占用，直接触发内存泄漏。

最优方案核心是流式处理+批量写入，从根源上控制内存占用，同时提升写入效率：

一、核心思路 #

1. 避免一次性加载全量数据：如果是从CSV文件读取，用流式读取逐行处理，不把60万条记录一次性塞进内存。
2. 批量提交写操作：利用Firestore的WriteBatch API，每积累500条记录（Firestore批量操作上限）提交一次，减少HTTP请求次数，同时控制内存中待处理的任务数量。

二、流式读取CSV+批量写入示例代码 #

如果还没把CSV数据加载到内存，推荐用这种方式，内存占用最低：

const { Firestore } = require('@google-cloud/firestore');
const csv = require('csv-parser');
const fs = require('fs');

const db = new Firestore();
const PROMO_CODES = 'promo_codes';
const RULE_NAME = '你的规则名称';
const BATCH_SIZE = 500; // Firestore批量操作的最大条数

let batch = db.batch();
let totalCount = 0;

// 流式读取CSV文件
fs.createReadStream('你的CSV文件路径.csv')
  .pipe(csv())
  .on('data', (row) => {
    // 假设CSV中存储docId的字段名为docId，根据实际调整
    const docRef = db.collection(PROMO_CODES).doc(row.docId);
    batch.set(docRef, { name: RULE_NAME });
    totalCount++;

    // 每积累到BATCH_SIZE条，提交一次批次
    if (totalCount % BATCH_SIZE === 0) {
      return batch.commit()
        .then(() => {
          console.log(`已完成 ${totalCount} 条记录写入`);
          batch = db.batch(); // 重置批次
        })
        .catch(err => console.error(`批次提交失败: ${err.message}`));
    }
  })
  .on('end', () => {
    // 处理剩余不足一批的记录
    if (totalCount % BATCH_SIZE !== 0) {
      batch.commit()
        .then(() => console.log(`所有记录写入完成，总计 ${totalCount} 条`))
        .catch(err => console.error(`剩余记录提交失败: ${err.message}`));
    }
  })
  .on('error', (err) => console.error(`CSV文件读取错误: ${err.message}`));

三、已加载全量数据的分批处理方案 #

如果已经把所有记录加载到了data数组中，可以用分批循环处理，避免一次性创建大量Promise：

const { Firestore } = require('@google-cloud/firestore');

const db = new Firestore();
const PROMO_CODES = 'promo_codes';
const RULE_NAME = '你的规则名称';
const BATCH_SIZE = 500;

async function batchWriteAll(data) {
  for (let i = 0; i < data.length; i += BATCH_SIZE) {
    const batch = db.batch();
    const currentBatch = data.slice(i, i + BATCH_SIZE);

    currentBatch.forEach(docId => {
      const docRef = db.collection(PROMO_CODES).doc(docId);
      batch.set(docRef, { name: RULE_NAME });
    });

    await batch.commit();
    console.log(`已完成 ${i + currentBatch.length} 条记录写入`);
  }
  console.log('全部记录写入完成');
}

// 调用执行
batchWriteAll(data).catch(err => console.error('整体写入失败:', err.message));

额外优化建议 #

• 错误重试：可以给批次提交失败的逻辑添加重试机制（比如最多重试3次），避免因网络波动导致数据丢失。
• 监控进度：可以结合日志或监控工具跟踪写入进度，确保任务正常推进。

内容的提问来源于stack exchange，提问作者Alex Javier Ulloa