Nodejs将csv文件从流解析到S3存储桶答案

【问题标题】：Nodejs parsing csv file from stream to S3 bucketNodejs将csv文件从流解析到S3存储桶
【发布时间】：2022-08-04 17:45:52
【问题描述】：

我正在尝试解析从 s3 存储桶下载的 CSV 文件，然后基于 Stream 将其上传到 s3 以避免将文件加载到内存 - 大文件。

const objectStream = object?.Body as Readable | undefined;

    if (objectStream === undefined) {
        throw new Error(\'No data\');
    }

    const transformationStream = new PassThrough();
    const parserStream = parse({ headers: (headers) => headers.map((header) => header + \'TEST\') })
        .on(\'error\', (error) => this.log.error(error))
        .on(\'data\', (row) => this.log.info(JSON.stringify(row)))
        .on(\'end\', (rowCount: number) => this.log.info(`Parsed ${rowCount} rows`));

    objectStream.pipe(parserStream).pipe(transformationStream);

    const upload = new Upload({
        client: s3Client,
        params: {
            Bucket: this.bucketName,
            Key: key,
            Body: transformationStream,
        },
    });

    try {
        await upload.done();
    } catch (error) {
        this.log.error(error);
        throw error;
    }

但我收到如下错误：

\"The \\\"chunk\\\" argument must be of type string or an instance of Buffer or Uint8Array. Received an instance of Object\",
\"stack\": \"TypeError [ERR_INVALID_ARG_TYPE]: The \\\"chunk\\\" argument must be of type string or an instance of Buffer or Uint8Array. Received an instance of Object
    at new NodeError (internal/errors.js:322:7)
    at PassThrough.Writable.write (internal/streams/writable.js:285:13)
    at uo.ondata (internal/streams/readable.js:731:22)
    at uo.emit (events.js:412:35)
    at uo.emit (/var/task/index.js:37:9384)
    at addChunk (internal/streams/readable.js:293:12)
    at readableAddChunk (internal/streams/readable.js:267:9)
    at uo.Readable.push (internal/streams/readable.js:206:10)
    at uo.Transform.push (internal/streams/transform.js:166:32)
    at uo.pushRow (/var/task/index.js:37:11093)\"

当我退出管道 pipe(parserStream) 时，代码工作正常并且文件已成功上传到 S3，否则会出现上述错误。

我对流的使用还不够，无法在这里得到可靠的答案，但是根据错误消息，我敢打赌 parserStream 没有输出字符串数据。它输出的对象正在通过 transformationStream 并进入不喜欢获取对象的 S3 上传。如何解决这个问题，嗯，不确定。这就是为什么这是评论而不是答案。
我认为问题可能是解析器流是一个对象流，而您正试图将其通过管道传输到非对象流，这正是 s3 上传想要的

标签： node.js csv amazon-s3

【解决方案1】：

parse（来自 csv-parse 的方法）的输出是数组。而 node.js 中可写流的输入必须是缓冲区字符串。
所以，据我了解parse 不能用于标准管道。

似乎来自 csv-parse 的人实现了他们自己的转换器，可以在您的情况下使用：

stream-transform (https://csv.js.org/transform/)

请参阅documentation 中的示例：

import { parse } from 'csv-parse';
import { generate } from 'csv-generate';
import { transform } from 'stream-transform';

const generator = generate({
  length: 20
});
const parser = parse({
  delimiter: ':'
});
const transformer = transform((record, callback) => {
  setTimeout(() => {
    callback(null, record.join(' ')+'\n');
  }, 500);
}, {
  parallel: 5
});
generator.pipe(parser).pipe(transformer).pipe(process.stdout);

【讨论】：