Node.js Streams Explained: pipeline and Async Iteration

Reading a 5 GB log file with fs.readFile loads all 5 GB into memory. Streams process data in small chunks, so memory use stays flat no matter how big the input is.

The four stream types

• Readable - a source (file, HTTP request, database cursor).
• Writable - a destination (file, HTTP response).
• Transform - modifies data as it passes (compression, parsing).
• Duplex - both readable and writable (a TCP socket).

Use pipeline, not pipe

pipeline connects streams, handles backpressure and, unlike .pipe(), cleans up and reports errors from every stage:

import { pipeline } from 'node:stream/promises';
import { createReadStream, createWriteStream } from 'node:fs';
import { createGzip } from 'node:zlib';

await pipeline(
  createReadStream('access.log'),
  createGzip(),
  createWriteStream('access.log.gz')
);

Async iteration: the easy way to read

import { createReadStream } from 'node:fs';
import readline from 'node:readline';

const lines = readline.createInterface({ input: createReadStream('access.log') });

let errors = 0;
for await (const line of lines) {
  if (line.includes(' 500 ')) errors++;
}
console.log(`${errors} server errors`);

Custom transforms with async generators

await pipeline(
  createReadStream('users.ndjson'),
  async function* (source) {
    let buffer = '';
    for await (const chunk of source) {
      buffer += chunk;
      const lines = buffer.split('\n');
      buffer = lines.pop();
      for (const line of lines) {
        const user = JSON.parse(line);
        if (user.active) yield `${user.email}\n`;
      }
    }
  },
  createWriteStream('active-emails.txt')
);

Streaming HTTP responses

app.get('/export.csv', async (req, res) => {
  res.setHeader('Content-Type', 'text/csv');
  await pipeline(db.queryStream('SELECT * FROM orders'), toCsv(), res);
});

Backpressure: when the destination is slower than the source, a correctly connected pipeline pauses reading automatically. That's what keeps memory flat.