Node.js Streams: Process Gigabytes of Data Without Running Out of Memory

const fs = require('fs');

// WRONG: 10GB into RAM
const data = fs.readFileSync('huge.csv'); // OutOfMemory!

// RIGHT: constant 64KB memory regardless of file size
const stream = fs.createReadStream('huge.csv', { highWaterMark: 64*1024 });
for await (const chunk of stream) processChunk(chunk);

const { pipeline } = require('stream/promises');
const fs = require('fs');
const zlib = require('zlib');
const { Transform } = require('stream');

const csvToJson = new Transform({
  objectMode: true,
  transform(chunk, enc, cb) {
    chunk.toString().split('\n').filter(Boolean).forEach(line => {
      const [id, name, email] = line.split(',');
      this.push(JSON.stringify({ id, name, email }) + '\n');
    });
    cb();
  }
});

// pipeline: handles errors, cleanup, backpressure automatically
await pipeline(
  fs.createReadStream('users.csv'),
  csvToJson,
  zlib.createGzip(),
  fs.createWriteStream('users.json.gz')
);
// Entire 10GB processed with ~200KB peak memory

const readline = require('readline');
const fs = require('fs');

async function processFile(path) {
  const rl = readline.createInterface({
    input: fs.createReadStream(path),
    crlfDelay: Infinity
  });
  let count = 0;
  for await (const line of rl) {
    await processLine(line); // async per line, backpressure automatic
    count++;
  }
  return count;
}

// WRONG: ignore backpressure
readable.on('data', chunk => {
  writable.write(chunk); // write() may return false = buffer full!
});

// RIGHT: pause/resume
readable.on('data', chunk => {
  if (!writable.write(chunk)) {
    readable.pause();
    writable.once('drain', () => readable.resume());
  }
});

// BEST: pipeline handles all of this automatically
await pipeline(readable, transform, writable);