TypeError [ERR_INVALID_ARG_TYPE] with node pipeline

Question

I'm using node 18.7 on ubuntu. I'm trying to parse a bunch of csv files to objects (using csv-parse), ultimately to load into a db. Because there are large numbers of these I decided to try streams and I'd like to use the async await style.

Based on Using async/await syntax with node stream , I have changed my code to

   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);

when I run the file, I'm getting:

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'
}

How can I fix this?

Answer 1

Reads from the csvParser stream are in object mode.

An fs.writeStream requires strings/buffers to be written to it.

A simple transform stream can be implemented to convert the objects to a JSON string by setting the writableObjectMode option:

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()
    }
}

Then use that between the parse + file write.

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

I believe the csv stringify project is similar for writing CSV's.