如何在Node.js Stream Pipeline中实现批量数据处理与写入？

针对超大CSV文件内存溢出问题，不能直接让MapData每X行触发一次，但可以通过维护批次缓存+异步流处理实现分批插入，核心思路是在流转换过程中积累数据，达到指定批次大小就执行数据库插入，同时处理流结束时的剩余数据。

具体实现步骤 #

1. 替换同步的mapSync为支持异步操作的转换流（推荐使用through2，它能很好处理流的背压和异步逻辑）
2. 在转换流内部维护批次数组和计数器，每积累X条数据就执行批量插入
3. 流结束时处理剩余的未达批次大小的数据，避免丢失
4. 捕获插入错误并传递给pipeline，确保错误能被及时处理

代码示例 #

const { pipeline } = require('stream');
const through2 = require('through2');
const split = require('split');

// 配置每批次处理的行数
const BATCH_SIZE = 100;

/**
 * 创建批量插入转换流
 * @param {Function} batchInsert - 批量插入数据库的异步回调
 * @returns {stream.Transform}
 */
createBatchInsertStream(batchInsert) {
  let batchCache = [];
  let rowCount = 0;

  // 创建异步转换流
  return through2.obj(async (filteredData, _, callback) => {
    const trimmedData = filteredData.trim();
    if (!trimmedData) {
      return callback(); // 跳过空行
    }

    batchCache.push(trimmedData);
    rowCount++;

    // 达到批次大小，执行批量插入
    if (rowCount >= BATCH_SIZE) {
      try {
        await batchInsert([...batchCache]);
        // 清空缓存和计数器
        batchCache = [];
        rowCount = 0;
        callback();
      } catch (err) {
        callback(err); // 传递错误，终止pipeline
      }
    } else {
      callback();
    }
  }, async (callback) => {
    // 流结束时，处理剩余的未达批次的数据
    if (batchCache.length > 0) {
      try {
        await batchInsert([...batchCache]);
        callback();
      } catch (err) {
        callback(err);
      }
    } else {
      callback();
    }
  });
}

// 使用方式
pipeline(
  storageStream,
  split(),
  this.FilterPipe(),
  this.createBatchInsertStream(async (batch) => {
    // 这里替换为你的数据库批量插入逻辑
    // 例如：await db.collection('your-collection').insertMany(batch);
    await yourDatabaseBatchInsertFunction(batch);
  }),
  (err) => {
    if (err) {
      console.error('CSV处理失败:', err);
    } else {
      console.log('CSV处理完成');
    }
  }
);

关键说明 #

• 内存控制：每次仅缓存最多BATCH_SIZE条数据，内存占用稳定，不会随文件大小增长
• 异步处理：通过through2的异步回调支持数据库插入的异步操作，避免阻塞流
• 背压处理：through2会自动管理流的背压，当批量插入未完成时，暂停读取上游数据，防止数据堆积
• 完整性保障：流结束的flush回调确保最后一批数据也能被插入，不会丢失

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