基于流处理的大体积CSV拆分并打包为Zip归档的实现难题

我明白你遇到的痛点了——在流式处理场景下，要动态把CSV按行数拆分并打包成Zip，archiver的API看起来好像只支持添加完整的流或缓冲区，没法逐行往不同的Zip文件里写内容。其实这里的关键是用PassThrough流作为每个拆分CSV文件的中间载体，把逐行的数据写入这个流后，archiver会自动读取并打包到对应的文件中。

下面是重构后的完整实现，我会标记出关键的改进点：

import archiver from 'archiver';
import { stringify, stringifySync } from 'csv-stringify';
import { Readable, PassThrough } from 'stream';

export const makeCsvStreamWrite = (header: Record<string, string>, rowLimit = 1000) =>
  async function* (input: AsyncIterable<Record<string, string | number>>): AsyncGenerator<Buffer> {
    let fileCounter = 1;
    let rowCounter = 0;
    let currentCsvStream: PassThrough | null = null;

    // 初始化Zip归档，配置高压缩比
    const archive = archiver('zip', { zlib: { level: 9 } });
    
    // 必须监听archiver的错误事件，避免静默失败
    archive.on('error', (err) => {
      throw new Error(`Zip归档失败: ${err.message}`);
    });

    // 辅助函数：创建新的CSV文件流并添加到Zip归档
    const createNewCsvFile = () => {
      // PassThrough流是双工流，写入的内容会原封不动输出给archiver读取
      const csvStream = new PassThrough();
      // 把这个流作为新文件添加到Zip，指定文件名
      archive.append(csvStream, { name: `data_${fileCounter}.csv` });
      // 给新文件写入CSV头
      csvStream.write(stringifySync([header]));
      fileCounter++;
      rowCounter = 0;
      return csvStream;
    };

    // 初始化第一个CSV文件
    currentCsvStream = createNewCsvFile();

    // 将输入转为可读流并字符串化（关闭自动加头，我们手动处理）
    const stringifier = stringify({ header: false });
    const stringifiedInput = Readable.from(input).pipe(stringifier);

    // 遍历每一行字符串化后的CSV内容
    for await (const row of stringifiedInput) {
      if (!currentCsvStream) throw new Error('无活跃的CSV文件流');
      
      // 将当前行写入到当前的CSV文件流
      currentCsvStream.write(row);
      rowCounter++;

      // 达到行限制时，切换到新的CSV文件
      if (rowCounter >= rowLimit) {
        // 结束当前CSV流，告诉archiver这个文件已写完
        currentCsvStream.end();
        // 创建并切换到新的CSV文件流
        currentCsvStream = createNewCsvFile();
      }
    }

    // 处理最后一个未完成的CSV文件
    if (currentCsvStream) {
      currentCsvStream.end();
    }

    // 完成Zip归档的创建
    await archive.finalize();

    // 将archiver的输出流转为async generator返回
    const archiveStream = Readable.from(archive);
    for await (const chunk of archiveStream) {
      yield chunk;
    }
  };

关键实现细节解释： #

1. PassThrough流的核心作用：
   这个流是Node.js提供的轻量级双工流，你写入它的内容会直接输出。我们把它传给archive.append()后，archiver会自动从这个流读取内容，相当于把它作为Zip中某一个文件的数据源。这样就实现了“逐行往Zip文件里写”的需求。

2. 文件切换逻辑：
   每当行计数器达到限制时，我们先调用currentCsvStream.end()结束当前流（这会触发archiver完成该文件的打包），然后通过createNewCsvFile()创建新的流并添加到Zip，同时自动写入新文件的CSV头。

3. 归档的收尾处理：
   所有行处理完成后，必须结束最后一个CSV流并调用archive.finalize()，这会让archiver完成Zip的收尾工作（比如写入文件目录结构）。

4. 错误处理：
   一定要监听archiver的error事件，否则如果归档过程中出现错误（比如内存不足、压缩失败），程序会静默崩溃而没有任何提示。

额外提示： #

• 如果你的上游调用方是直接将这个async generator用于流式上传（比如到S3），可以直接把生成的chunk传给S3的上传流，完全不需要把整个Zip加载到内存中，符合你流式处理的核心需求。
• 可以根据需要调整zlib的压缩级别，level:9是最高压缩比但会消耗更多CPU，平衡性能的话可以设为level:6。

备注：内容来源于stack exchange，提问作者florian norbert bepunkt