Direct answer
Build the pipeline as a chain of generators (functions that produce values one at a time, on demand, instead of building a whole list). Each stage pulls one item from the stage before it, does its work, and yields one item to the stage after it. Nothing asks for "all the records", so memory stays small and bounded at any moment (a chunk of the file, the lines readline has read ahead, the record being processed and the output buffer), whatever the file size. The final stage is a writable stream (an output sink with a bounded buffer); Node's stream.pipeline connects the stages and pauses the reader whenever the sink is full (back-pressure: a slow consumer slows the producer instead of letting data pile up between them).
Design: read, parse, filter, serialise, write
Four files make up the demo, all in one directory: three scripts and a shell runner. make-data.mjs creates the input:
js
// 1,000,000 lines "id,name,amount"; every 10th line is invalid (amount is not a number)
import { createWriteStream } from "node:fs";
const out = createWriteStream("records.csv");
for (let i = 1; i <= 1_000_000; i++) {
out.write(`${i},user-${i},${i % 10 === 0 ? "n/a" : (i % 977) + ".25"}\n`);
}
out.end();
stream.mjs is the streaming pipeline. Two Node pieces are used: readline (a built-in module that reads a stream one line at a time) and yield* (inside a generator, hand over each value of another iterable one by one). An async function* is an async generator: like a generator, but it may await and is consumed with for await ... of, which waits for each item in turn.
js
import { createReadStream, createWriteStream } from "node:fs";
import { createInterface } from "node:readline";
import { pipeline } from "node:stream/promises";
// parse: one line -> record, or null when invalid
const parseLine = (line) => {
const [id, name, amount] = line.split(",");
const n = Number(amount);
return Number.isFinite(n) && id && name ? { id: Number(id), name, amount: n } : null;
};
// each stage is a generator: it pulls one item, yields one item, and keeps no list of past items
async function* lines(path) {
yield* createInterface({ input: createReadStream(path), crlfDelay: Infinity });
}
async function* parsed(source) {
for await (const line of source) {
const rec = parseLine(line);
if (rec) yield rec; // invalid lines are dropped here
}
}
async function* toJsonLines(source) {
for await (const rec of source) yield JSON.stringify(rec) + "\n";
}
let peakRss = 0;
const timer = setInterval(() => (peakRss = Math.max(peakRss, process.memoryUsage().rss)), 20);
await pipeline(
async function* () { yield* toJsonLines(parsed(lines("records.csv"))); },
createWriteStream("out.jsonl")
);
clearInterval(timer);
console.log("streaming: done, peak RSS under 150 MB:", peakRss < 150 * 1024 * 1024);
The read-everything-first version of the same job, for comparison, is readall.mjs:
js
import { readFileSync, writeFileSync } from "node:fs";
const parseLine = (line) => {
const [id, name, amount] = line.split(",");
const n = Number(amount);
return Number.isFinite(n) && id && name ? { id: Number(id), name, amount: n } : null;
};
const records = readFileSync("records.csv", "utf8").split("\n").map(parseLine).filter(Boolean);
writeFileSync("out-all.json", JSON.stringify(records));
console.log("read-all: done,", records.length, "records");
run.sh generates a 1,000,000-line input, then runs both jobs with the JavaScript heap (the memory the engine manages for objects and strings) capped at 32 MB (--max-old-space-size=32). The pipeline's check uses RSS (resident set size: all physical memory the process holds, which also counts the Node runtime, compiled code and stream buffers that live outside the capped heap). That is why RSS is compared with its own generous 150 MB ceiling rather than with 32 MB: the cap limits the heap only, and the ceiling confirms the process as a whole stays small. The 25 MB test file stands in for a multi-gigabyte one: what matters is the ratio of input size to memory available, and a 32 MB cap against a 25 MB file reproduces the same failure at a size that runs in seconds.
run.sh:
sh
node make-data.mjs
sleep 1
echo "input bytes: $(stat -c %s records.csv)"
node --max-old-space-size=32 stream.mjs; echo "streaming exit code: $?"
echo "output lines: $(wc -l < out.jsonl)"
node --max-old-space-size=32 readall.mjs > readall.log 2>&1; echo "read-all exit code: $?"
grep -m1 -o "heap out of memory" readall.log
Run in a node:22 container (Linux) with sh run.sh, it prints:
text
input bytes: 25376416
streaming: done, peak RSS under 150 MB: true
streaming exit code: 0
output lines: 900000
read-all exit code: 134
heap out of memory
The streaming job writes all 900,000 valid records (every tenth line has n/a as its amount and is dropped) under the same cap that makes the read-all job abort with heap out of memory (exit code 134 is 128 plus signal 6, SIGABRT, the signal a process sends itself when it aborts). The RSS line is informational (it prints true or false and does not set the exit code); the evidence is the exit codes. A variant of stream.mjs whose toJsonLines first pushes every record into an array aborts the same way, with exit code 134 and heap out of memory, under the same cap.
Why it avoids the blowup
readFileSync(...).split("\n").map(...) holds the file as one string, then as an array of line strings, then as an array of objects, all alive at once, so memory grows linearly with the input and the process dies when it passes the limit. The generator chain is pull-based: for await asks the previous stage for the next item only when this stage is ready, so records flow through the stages one at a time and become garbage once written. One stage reads ahead: the readline async iterator queues lines and pauses its input when the queue reaches 1,024 lines (highWaterMark: 1024 in the Node 22 source). That is a fixed bound, so memory stays constant, but it is a queue of lines, not literally one line. The stream's buffer is bounded by its highWaterMark; Node's documentation describes that as a threshold at which the stream stops asking for more data, not a hard memory limit.
What breaks the property
- Collecting anywhere.
Array.from(stage), spreading, toArray, or await Promise.all(...) over the stream materialises everything (builds the full result in memory at once). So do operations that need all the data: global sort, group-by, and de-duplication with a Set (it grows with the number of distinct keys). Fix by an external sort (sort pieces small enough to fit in memory, write each sorted piece to disk, then merge the pieces by reading them in parallel), or by moving that step to a database.
- Ignoring back-pressure. If you write with a manual loop and never wait when
write() returns false, Node's documentation says it buffers every chunk until memory runs out and then aborts. pipeline (or await once(stream, "drain") in a loop, where drain is the event the stream emits when its buffer has emptied and it is safe to write again) is what keeps the producer paced.
- Unbounded concurrency. Starting one async task per line (
lines.map(async ...) collected in an array) holds every in-flight task in memory. Cap the concurrency (a small worker pool or a batch of N) if a stage does I/O.
- One enormous record. The chain is constant-memory per record; a single multi-hundred-megabyte line is buffered whole by the line reader.
- Keeping evidence. Pushing every rejected line into an error array re-creates the problem. Count the invalid lines and keep only a bounded sample.
Failure behaviour
pipeline destroys every stream in the chain when any stage throws and rejects its promise, so a bad file does not leave handles open. A malformed line is handled in parseLine (returns null, the stage drops it), so one bad record does not stop the job. The output is only complete if the promise resolves, so write to a temporary file and rename it on success if downstream readers must never see a partial file.
