javascript - TypeError [ERR_INVALID_ARG_TYPE] 与 Node 管道

Question

我在 ubuntu 上使用 Node 18.7。我正在尝试将一堆 csv 文件解析为对象(使用 csv-parse)，最终加载到数据库中。因为有很多这样的我决定尝试流，我想使用异步等待风格。

基于 Using async/await syntax with node stream ，我已将我的代码更改为

   const { parse } = require('csv-parse');
const path = __dirname + '/file1.csv';
const opt = { columns: true, relax_column_count: true, skip_empty_lines: true, skip_records_with_error: true };
console.log(path);
const { pipeline } = require('node:stream/promises');

 async function readByLine(path, opt) {
    const readFileStream = fs.createReadStream(path);
    const writeFileStream = fs.createWriteStream(__dirname + '/file2');
    var csvParser = parse(opt, function (err, records) {
        if (err) throw err;
    });
   await pipeline(readFileStream, csvParser, writeFileStream);

}

readByLine(path, opt);

当我运行该文件时，我得到:

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 (node:internal/errors:387:5)
at _write (node:internal/streams/writable:315:13)
at Writable.write (node:internal/streams/writable:337:10)
at Parser.ondata (node:internal/streams/readable:766:22)
at Parser.emit (node:events:513:28)
at Readable.read (node:internal/streams/readable:539:10)
at Parser.<anonymous> (/home/gmail-username/node/maricopa/node_modules/csv-parse/dist/cjs/index.cjs:1357:28)
at Parser.emit (node:events:513:28)
at emitReadable_ (node:internal/streams/readable:590:12)
at process.processTicksAndRejections (node:internal/process/task_queues:81:21) {
code: 'ERR_INVALID_ARG_TYPE'
}

我该如何解决这个问题？

Answer 1

Reads from the csvParser stream在 object mode .

安fs.writeStream需要将字符串/缓冲区写入其中。

一个简单的transform stream可以通过设置 writableObjectMode 选项将对象转换为 JSON 字符串:

class JsonTransform extends Transform {
    constructor(opt) {
        super(Object.assign({}, { writableObjectMode: true }, opt));
    }
    _transform(obj, enc, callback) {
        let ret = JSON.stringify(obj) + '\n'
        this.push(ret)
        callback()
    }
}

然后在解析 + 文件写入之间使用它。

const toJSON = new JsonTransform()
await pipeline(readFileStream, csvParser, toJSON, writeFileStream);

我相信 csv stringify 项目与编写 CSV 类似。