Reading a CSV File in Node With Streams and Backpressure

Reading a CSV in Node usually starts with fs.readFileSync and a parse call, and that works until the file is bigger than you planned for. The alternative is to treat the file as a stream: bytes flow from disk to the parser to your code in small pieces, and nothing waits for the whole file. This article shows the pattern on a 2,000,000 row file, what it costs in memory and time, the three ways to consume the stream, and the error and encoding behavior you need to handle for a file you did not write. The overall tour of CSV in JavaScript is in the parsing guide.

Why streams

A CSV parser that returns an array needs the whole file in memory twice: once as the text and once as the rows. I parsed the 2,000,000 row orders file both ways and summed the amount column:

for m in split csv-parse-stream csv-parse-columns csv-parse-sync papaparse-stream papaparse-string; do node --max-old-space-size=4096 bench.mjs $m; done

split              rows=2000000 total=160174646.36 time=1.15s peak_rss=70MB
csv-parse-stream   rows=2000000 total=160174646.36 time=4.16s peak_rss=74MB
csv-parse-columns  rows=2000000 total=160174646.36 time=6.26s peak_rss=78MB
csv-parse-sync     rows=2000000 total=160174646.36 time=4.31s peak_rss=1132MB
papaparse-stream   rows=2000000 total=160174646.36 time=1.12s peak_rss=84MB
papaparse-string   rows=2000000 total=160174646.36 time=2.31s peak_rss=1179MB

csv-parse as a stream (csv-parse-stream) took 4.16 s and a peak of 74 MB. csv-parse/sync (csv-parse-sync) took 4.31 s and 1,132 MB, about fifteen times more, for the same total of 160174646.36. The time is similar because the parser dominates; the memory is not, because the stream never holds more than a few thousand rows. If a file might grow, the stream is the version that keeps working.

Three ways to consume it

The parser is a Transform stream: bytes in, records out. You can use it in three ways, and they differ in how much they ask of you.

Async iteration is the shortest. for await pulls one record at a time and pauses the file when your loop body is slow:

import fs from "node:fs";
import { parse } from "csv-parse";

const parser = fs.createReadStream(path).pipe(parse({ columns: true }));
for await (const record of parser) {
  // one record at a time; the stream waits while this runs
}

This is the form the benchmark above used. One thing to know: .pipe does not forward errors from the file stream to the parser, so a missing file raises ENOENT on the read stream and the loop may hang or crash separately. For anything that must survive bad input, use pipeline.

**pipeline** from node:stream/promises connects the stages, forwards errors from any of them and cleans up all the streams on failure or on success. I used it to filter the file with a Transform and count what came out of a sink:

import fs from "node:fs";
import { pipeline } from "node:stream/promises";
import { Transform, Writable } from "node:stream";
import { parse } from "csv-parse";

const big = "/batches/csv-0001-0200/work/orders-2m.csv";

// 1. pipeline with a transform and a sink: filter refunded rows, count them
let kept = 0;
const t = performance.now();
await pipeline(
  fs.createReadStream(big),
  parse({ columns: true }),
  new Transform({
    objectMode: true,
    transform(rec, _enc, cb) { cb(null, rec.status === "refunded" && Number(rec.amount) > 500 ? rec : undefined); },
  }),
  new Writable({ objectMode: true, write(_rec, _enc, cb) { kept++; cb(); } }),
);
console.log(`pipeline: kept ${kept} rows in ${((performance.now() - t) / 1000).toFixed(2)}s, rss ${(process.resourceUsage().maxRSS / 1024).toFixed(0)} MB`);

// 2. leaving an async loop early closes the file
const t2 = performance.now();
let seen = 0;
for await (const r of fs.createReadStream(big).pipe(parse({ from_line: 2 }))) { if (++seen === 3) break; }
console.log(`break after ${seen} records: ${(performance.now() - t2).toFixed(0)} ms`);

// 3. errors carry a code and a line number
for (const [file, opts] of [["ragged2.csv", {}], ["unterminated.csv", {}], ["ragged2.csv", { relax_column_count: true }], ["ragged2.csv", { skip_records_with_error: true }]]) {
  const rows = [];
  try {
    const p = fs.createReadStream("/batches/csv-0001-0200/lab/data/py/" + file).pipe(parse({ columns: true, ...opts }));
    for await (const r of p) rows.push(r);
    console.log(`${file} ${JSON.stringify(opts)} -> ${rows.length} records`);
  } catch (e) {
    console.log(`${file} ${JSON.stringify(opts)} -> ${e.code}: ${e.message.slice(0, 90)}`);
  }
}

node --max-old-space-size=4096 streams.mjs

pipeline: kept 25 rows in 6.82s, rss 63 MB
break after 3 records: 17 ms
ragged2.csv {} -> CSV_RECORD_INCONSISTENT_COLUMNS: Invalid Record Length: columns length is 4, got 5 on line 3
unterminated.csv {} -> CSV_QUOTE_NOT_CLOSED: Quote Not Closed: the parsing is finished with an opening quote at line 5000
ragged2.csv {"relax_column_count":true} -> 8 records
ragged2.csv {"skip_records_with_error":true} -> 4 records

The first line is the pipeline: the read stream, the parser with columns: true, a Transform that keeps refunded orders above 500, and a Writable that counts. It kept 25 rows, the same as the DuckDB and Miller counts in the shell guide, in 6.82 s at 63 MB. It is slower than the bare loop because every record passes through three object-mode streams, each with its own buffering; object mode streams default to a buffer of 16 objects, and the highWaterMark that sets that is the lever if you need to tune it. The memory stayed flat. When the sink is slow, say an HTTP call or a database write, pipeline slows the file read to match, which is backpressure. Without it, a fast reader and a slow writer fill memory with the difference.

Events (parser.on("data", ...)) are the oldest form. I avoid them for new code, because they make backpressure and error handling manual.

Stopping early

A loop does not have to read the whole file. Leaving a for await early destroys the iterator, which destroys the parser, and the pipe stops reading from disk:

The second line of the output shows it: break after 3 records: 17 ms. The file has 2,000,000 records and the program touched three. This is how you build "show me the first ten rows" or "find the first record matching X" on a file of any size, and it is the behavior that fs.readFileSync plus a parse cannot give you. Use to_line or to in the parser options if you know the limit up front, and break if the stopping condition depends on the data.

When the file is not clean

The parser throws on structural problems, with a code you can branch on and a line number. Three files from the Python series, read as streams:

The remaining lines of the output show each. A row with the wrong number of fields raised CSV_RECORD_INCONSISTENT_COLUMNS: Invalid Record Length: columns length is 4, got 5 on line 3. An unclosed quote raised CSV_QUOTE_NOT_CLOSED: Quote Not Closed: the parsing is finished with an opening quote at line 5000. Read that second message carefully: the line number is where the file ended, not where the quote was opened. In the test file the stray quote is on line 3, and the parser reports line 5000 because it kept consuming lines looking for the close. Use the line number as "somewhere before here", and find the start with the same technique as in the Python field limit article.

You can make the parser tolerant, and each option has a price:

• relax_column_count: true accepted the ragged file as 8 records. The row that was a short record and the note that was broken over two lines each became their own record, so there are more records than the file has, and the column count of each varies.

• skip_records_with_error: true returned 4 records and dropped the rest, with no list of what was dropped unless you also pass an on_skip callback.

Neither is a repair. As in the other languages, I would count what was skipped and write the skipped records to a file, so a clean run and a lossy run are not the same exit code.

Encodings and the BOM

Streams give you bytes and the parser expects text. Two inputs commonly need help:

import fs from "node:fs";
import { Readable } from "node:stream";
import { parse } from "csv-parse";

const dir = "/batches/csv-0001-0200/lab/data/";
// BOM
for (const bom of [false, true]) {
  const recs = [];
  for await (const r of fs.createReadStream(dir + "py/bom.csv").pipe(parse({ columns: true, bom }))) recs.push(r);
  console.log(`bom:${bom}`.padEnd(10), JSON.stringify(Object.keys(recs[0])));
}
// encoding: a Windows-1252 file through TextDecoderStream
const out = [];
const web = Readable.toWeb(fs.createReadStream(dir + "enc/cp1252.csv")).pipeThrough(new TextDecoderStream("windows-1252"));
for await (const r of Readable.fromWeb(web).pipe(parse({ columns: true }))) out.push(r);
console.log("windows-1252 :", JSON.stringify(out[0]));
const wrong = [];
for await (const r of fs.createReadStream(dir + "enc/cp1252.csv", { encoding: "utf8" }).pipe(parse({ columns: true }))) wrong.push(r);
console.log("read as utf8 :", JSON.stringify(wrong[0]));
// info and line numbers
for await (const { record, info } of fs.createReadStream(dir + "py/people.csv").pipe(parse({ columns: true, info: true }))) {
  console.log(`record ${info.records} starts at line ${info.lines - (record.note.includes("\n") ? 1 : 0)}:`, record.name);
}

node streams2.mjs

bom:false  ["id","name"]
bom:true   ["id","name"]
windows-1252 : {"id":"1","name":"José Müller","city":"São Paulo","note":"price €4.50"}
read as utf8 : {"id":"1","name":"Jos� M�ller","city":"S�o Paulo","note":"price �4.50"}
record 1 starts at line 2: Ana Silva
record 2 starts at line 3: Smith, Jane
record 3 starts at line 4: Bo Chen
record 4 starts at line 6: Cy Dubois

A file saved with a UTF-8 byte order mark produced the key "id" with bom: false and "id" with bom: true, so I set bom: true for any file that came from a person. A Windows-1252 file read as UTF-8 returned {"name":"Jos� M�ller","city":"S�o Paulo","note":"price �4.50"}, with a replacement character wherever a byte was not valid UTF-8, and the damage is permanent since the original byte is gone. Piping the bytes through a TextDecoderStream("windows-1252") before the parser returned José Müller, São Paulo and price €4.50. Detecting which encoding the file is in, before you choose, is a separate problem, covered in the encoding detection article.

The info: true option adds a second property, info, to each record, with counters. In my run info.lines was the line on which the record ended: 2, 3, 5 and 6 for the four records of the people file, so the third record (which has a line break in a note) ends on line 5 and began on line 4. That is the number to put in an error message about a bad value.

How to tell it worked

Count the records and compare with qsv count or DuckDB; a stream bug that drops or duplicates a chunk shows as a different number. Check the peak memory of the process, process.resourceUsage().maxRSS, against what you expected; it should not grow with the file. Run it on a file with a BOM and a line break inside a field. Make sure an error in the middle of the file produces a failing exit code, with a test that feeds it a bad file. And confirm the cleanup: after an early break, the process should exit without waiting on the rest of the file.

When streams are not worth it

For a file of a few megabytes that you will read once, readFileSync and csv-parse/sync are shorter and the memory does not matter. For a browser, the equivalent is Papa Parse's step and chunk callbacks, covered in the browser article. And when the real task is loading the data somewhere, a stream into a database does the parsing and the loading in one pass, as in streaming a huge CSV into a database. If your only goal is speed on a large file, csv-parse is not the fastest parser on this list (Papa Parse took 1.12 s on the same file, as the guide shows), and the options in the csv-parse article are where its time goes.