Transform Gigabyte Files with Constant Memory Streams

1. Establish the current memory ceiling with the whole-file version.

node -e "const fs=require('fs');const d=fs.readFileSync(process.argv[1],'utf8');console.log('read', (d.length/1e6).toFixed(0)+'MB')" big.jsonl; ps -o rss= -p $$ | awk '{print "peak RSS ~", $1/1024, "MB"}'

Expected output: The read prints the file size and the RSS snapshot shows memory proportional to input, confirming the scaling problem.

2. Rewrite as a pipeline with a transform stream.

import { createReadStream, createWriteStream } from 'node:fs';
import { createInterface } from 'node:readline';
import { pipeline } from 'node:stream/promises';
import { Transform } from 'node:stream';

const scrub = new Transform({
  transform(line, _enc, cb) {
    if (!line.trim()) return cb();
    try {
      const row = JSON.parse(line);
      row.email = row.email?.toLowerCase().trim() ?? null;
      row.ssn = '[redacted]';
      this.push(JSON.stringify(row) + '\n');
      cb();
    } catch (err) {
      cb(new Error(`bad line: ${line.slice(0, 80)}`)); // fail loud, never skip
    }
  },
});

await pipeline(
  createReadStream(process.argv[2], { encoding: 'utf8' }),
  createInterface({ crlfDelay: Infinity }), // readline is a valid pipeline leg
  scrub,
  createWriteStream(process.argv[3]),
);

Expected output: Running the pipeline on the same file completes with RSS pinned near baseline regardless of input size.

3. Watch memory stay flat while throughput stays high.

node scrub-stream.mjs big.jsonl out.jsonl & PID=$!; for i in 1 2 3 4; do ps -o rss= -p $PID | awk '{printf "RSS %dMB\n", $1/1024}'; sleep 30; done; wait $PID; wc -l out.jsonl

Expected output: RSS samples print flat (tens of MB) while the file processes, and the output line count matches the input minus blank lines.

4. Verify equivalence on a sample before swapping the job.

head -n 10000 big.jsonl > sample.jsonl && node scrub-wholefile.mjs sample.jsonl sample-old.jsonl && node scrub-stream.mjs sample.jsonl sample-new.jsonl && diff <(jq -S . sample-old.jsonl) <(jq -S . sample-new.jsonl) && echo "outputs equivalent"

Expected output: The sorted, normalized sample outputs match byte-for-byte, proving the stream version is behaviorally identical on the comparison set.

5. Add integrity guards to the production job.

#!/usr/bin/env bash
set -euo pipefail
IN="$1"; OUT="$2"
in_lines=$(grep -c . "$IN" || true)
node scrub-stream.mjs "$IN" "$OUT"
out_lines=$(grep -c . "$OUT" || true)
if [ "$in_lines" -ne "$out_lines" ]; then
  echo "row count mismatch: in=$in_lines out=$out_lines" >&2
  exit 1
fi
echo "OK: $out_lines rows transformed"

Expected output: The nightly job now fails loudly on any row count mismatch instead of writing a silently truncated file and reporting success.

Problem

Transform writes garbled multi-byte characters

Solution

A chunk boundary split a UTF-8 sequence; either set encoding on the read stream and operate on strings, or wrap with a StringDecoder in the transform.

Problem

Pipeline aborts mid-file with EMFILE

Solution

Too many open handles elsewhere; the pipeline itself opens one stream, so audit concurrent jobs or raise ulimit -n, and rely on pipeline to close streams on error.

Problem

Throughput is worse than the whole-file version

Solution

Per-line JSON.parse dominates; batch the transform with a highWaterMark-sized buffer, parsing and pushing every N lines to amortize per-call overhead.