Node.js Knex流ETL管道大数据量时await查询挂起的解决方法

我用Node.js Stream实现ETL管道，通过Knex.js从Postgres的Books表读取数据，经转换流处理后，用批量转换流按500条为一批分组，最终用可写流更新author表。读写操作共用同一个Knex事务，小数据量时运行正常，但数据量较大时，流执行到await transaction.raw(queryText)处即挂起，求解决办法。

相关代码 #

const transaction = await knex.transaction();
const readStream = transaction.raw("SELECT * from Books").stream();
const transformStream = new Transform({
  objectMode: true,
  highWaterMark: 1000,
  transform(chunk, _encoding, next) {
    this.push(chunk);
  },
});
let batch: unknown[] = [];
const batchTransform = new Transform({
  objectMode: true,
  highWaterMark: 1,
  transform(chunk, encoding, next) {
    batch.push(chunk);
    if (batch.length >= 500) {
      next(null, batch);
      batch = [];
    }
  },
  flush(next) {
    next(null, batch);
  },
});
const writableStream = new Writable({
  objectMode: true,
  highWaterMark: 1,
  async write(chunk, _encoding, callback) {
    const queryText = `
      UPDATE "author"
      SET "Book" = 'test'
      WHERE "id" = '1';
    `;
    await transaction.raw(queryText); // stream ends here
    callback();
  },
});

await pipeline([readStream, transformStream, batchTransform, writableStream]);

问题根源与解决办法 #

1. 核心问题分析 #

• 事务连接阻塞：Knex的事务基于单个数据库连接，读流持续占用该连接拉取数据时，可写流里的异步更新请求会被阻塞，大数据量下这种冲突直接导致流程挂起。
• 批量流背压失效：当前batchTransform的transform方法中，未达到500条时未调用next()，破坏了Node.js Stream的背压机制，数据堆积后阻塞整个管道。

2. 具体修复步骤 #

修复批量转换流的背压逻辑

每次处理完chunk后必须调用next()，不管批次是否满额：

let batch: unknown[] = [];
const batchTransform = new Transform({
  objectMode: true,
  highWaterMark: 1,
  transform(chunk, encoding, next) {
    batch.push(chunk);
    if (batch.length >= 500) {
      next(null, batch);
      batch = [];
    } else {
      // 必须调用next()，否则流会卡住
      next();
    }
  },
  flush(next) {
    // 处理剩余的不足批量的数据
    if (batch.length > 0) {
      next(null, batch);
    } else {
      next();
    }
  },
});

分离读写的事务/连接

不要让读流绑定到写事务的连接上，读操作使用独立连接，写操作单独用事务：

// 单独创建事务用于写操作
const writeTx = await knex.transaction();
// 读操作使用默认连接池，不绑定事务
const readStream = knex.raw("SELECT * from Books").stream();

优化可写流的异步错误处理

确保异常能被正确捕获并传递给流管道，避免静默挂起：

const writableStream = new Writable({
  objectMode: true,
  highWaterMark: 1,
  async write(chunk, _encoding, callback) {
    try {
      // 这里可以根据chunk中的数据生成动态更新语句，而非固定id
      const queryText = `
        UPDATE "author"
        SET "Book" = 'test'
        WHERE "id" = '1';
      `;
      await writeTx.raw(queryText);
      callback();
    } catch (err) {
      // 传递错误，让pipeline终止并触发回滚
      callback(err);
    }
  },
});

完善事务的收尾逻辑

在pipeline完成后根据结果提交或回滚事务：

try {
  await pipeline([readStream, transformStream, batchTransform, writableStream]);
  await writeTx.commit();
} catch (err) {
  await writeTx.rollback();
  throw err;
}

额外优化建议 #

• 调整流的水位参数：根据服务器内存情况，调整读流的highWaterMark，避免一次性加载过多数据到内存。
• 批量更新优化：利用批次数据生成批量更新语句（如UPDATE ... WHERE id IN (...)或使用Knex的批量更新API），减少数据库请求次数，提升效率。
• 监控流状态：给各流添加error事件监听，快速定位异常点；通过data事件监控数据处理速度，排查瓶颈。

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