如何在NodeJS中创建一对多转换流（Transform Stream）？

核心思路 #

Transform流本身支持单输入多输出，关键在于多次调用this.push()输出多条记录，最后仅调用一次回调函数——_transform方法的回调只是用来标记当前输入记录的处理完成，而非限制输出次数。

实现步骤 #

• 开启objectMode: true：因为处理的是数据集对象，而非二进制Buffer，必须在Transform构造时配置该选项。
• 循环推送输出记录：对每条输入记录，根据配置生成多条输出后，循环调用this.push()将每条记录推入下游。
• 处理背压：若this.push()返回false，说明下游缓冲区已满，需暂停推送，直到drain事件触发后恢复。
• 单次回调：所有输出记录推送完成后，调用一次回调函数标记当前输入处理完成；若出错则将错误传入回调。

代码示例 #

const { Transform } = require('stream');

class ETLMultiTransform extends Transform {
  constructor(transformConfig) {
    // 开启对象模式，禁用默认的Buffer编码处理
    super({ objectMode: true });
    this.transformConfig = transformConfig;
  }

  _transform(inputRecord, _encoding, callback) {
    try {
      // 根据配置生成多条输出记录（替换为你的业务逻辑）
      const outputRecords = this.processRecord(inputRecord);

      // 推送所有输出记录，处理背压
      outputRecords.forEach(record => {
        if (!this.push(record)) {
          this.pause();
        }
      });

      // 标记当前输入记录处理完成（仅调用一次）
      callback();

      // 监听drain事件恢复推送
      this.once('drain', () => this.resume());
    } catch (error) {
      // 出错时传递错误，终止流（可选：也可自定义错误处理逻辑）
      callback(error);
    }
  }

  // 业务逻辑：根据配置将单条输入转换为多条输出
  processRecord(input) {
    const outputs = [];
    // 示例：根据配置的规则拆分/复制记录
    this.transformConfig.fields.forEach(field => {
      outputs.push({
        ...input,
        derivedField: field,
        timestamp: Date.now()
      });
    });
    return outputs;
  }
}

// 使用示例
const transformStream = new ETLMultiTransform({
  fields: ['user', 'order', 'product'] // 每条输入生成3条输出的配置
});
// 对接上下游流：readable.pipe(transformStream).pipe(writable);

关于Duplex流的误解 #

你不需要使用普通Duplex流——Transform流本身就是Duplex的子类，专门用于单向的输入转换输出场景。普通Duplex流需要同时实现_read和_write，而Transform只需要实现_transform，完全符合你的ETL单向转换需求。

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