Node.js Streams: A Practical Guide

Streams are one of the most powerful features in Node.js and also one of the most misunderstood. They let you process data piece by piece instead of loading everything into memory at once. Reading a 2GB file? With streams, your process uses a few megabytes of memory regardless of file size. Without streams, you need 2GB of RAM.

Why streams matter

Consider reading a file and sending it as an HTTP response:

// Without streams: loads entire file into memory
const http = require('http');
const fs = require('fs');

http.createServer((req, res) => {
  fs.readFile('./large-file.csv', (err, data) => {
    res.end(data); // 'data' holds the entire file in memory
  });
}).listen(3000);

// With streams: processes in chunks
const http = require('http');
const fs = require('fs');

http.createServer((req, res) => {
  const stream = fs.createReadStream('./large-file.csv');
  stream.pipe(res); // Data flows chunk by chunk
}).listen(3000);

The streamed version starts sending data immediately (lower latency) and uses constant memory regardless of file size.

The four types of streams

Type	Description	Example
Readable	Source of data	fs.createReadStream, http.IncomingMessage
Writable	Destination for data	fs.createWriteStream, http.ServerResponse
Transform	Modifies data as it passes through	zlib.createGzip, custom parsers
Duplex	Both readable and writable	net.Socket, crypto.createCipher

Readable streams

Reading a file

const fs = require('fs');

const readable = fs.createReadStream('./data.txt', {
  encoding: 'utf8',
  highWaterMark: 64 * 1024, // 64KB chunks (default is 64KB)
});

readable.on('data', (chunk) => {
  console.log(`Received ${chunk.length} characters`);
});

readable.on('end', () => {
  console.log('No more data');
});

readable.on('error', (err) => {
  console.error('Error:', err.message);
});

Flowing vs paused mode

Readable streams operate in two modes:

• Flowing mode: Data is read automatically and provided via events.
• Paused mode: You must explicitly call read() to get chunks.

// Flowing mode (triggered by .on('data') or .pipe())
readable.on('data', (chunk) => { /* automatic */ });

// Paused mode (manual reading)
readable.on('readable', () => {
  let chunk;
  while ((chunk = readable.read()) !== null) {
    console.log(`Read ${chunk.length} bytes`);
  }
});

Async iteration (modern approach)

const fs = require('fs');

async function processFile() {
  const readable = fs.createReadStream('./data.txt', { encoding: 'utf8' });

  for await (const chunk of readable) {
    console.log(`Chunk: ${chunk.length} chars`);
  }

  console.log('Done');
}

processFile();

This is the cleanest way to consume a readable stream in modern Node.js.

Writable streams

const fs = require('fs');

const writable = fs.createWriteStream('./output.txt');

writable.write('Hello, ');
writable.write('world!\n');
writable.end('Final line.\n'); // Signals that no more data will be written

writable.on('finish', () => {
  console.log('All data has been flushed to the file');
});

Handling backpressure

write() returns false when the internal buffer is full. You should stop writing until the drain event fires:

function writeData(writable, data) {
  for (let i = 0; i < data.length; i++) {
    const canContinue = writable.write(data[i]);
    if (!canContinue) {
      // Buffer is full, wait for drain
      writable.once('drain', () => {
        writeData(writable, data.slice(i + 1));
      });
      return;
    }
  }
  writable.end();
}

Piping streams

pipe() connects a readable stream to a writable stream and handles backpressure automatically:

const fs = require('fs');

const readable = fs.createReadStream('./input.txt');
const writable = fs.createWriteStream('./output.txt');

readable.pipe(writable);

writable.on('finish', () => {
  console.log('Copy complete');
});

Pipeline (recommended)

stream.pipeline() is better than pipe() because it handles errors and cleanup:

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

async function compressFile() {
  await pipeline(
    fs.createReadStream('./input.txt'),
    zlib.createGzip(),
    fs.createWriteStream('./input.txt.gz')
  );
  console.log('Compression complete');
}

compressFile().catch(console.error);

With pipe(), if the readable stream emits an error, the writable stream is not automatically closed, potentially causing memory leaks. pipeline() handles all cleanup.

Transform streams

Transform streams modify data as it passes through. They are both readable and writable.

Built-in transforms

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

// Compress
await pipeline(
  fs.createReadStream('data.json'),
  zlib.createGzip(),
  fs.createWriteStream('data.json.gz')
);

// Decompress
await pipeline(
  fs.createReadStream('data.json.gz'),
  zlib.createGunzip(),
  fs.createWriteStream('data.json')
);

Custom transform stream

A transform that converts text to uppercase:

const { Transform } = require('stream');

class UpperCaseTransform extends Transform {
  _transform(chunk, encoding, callback) {
    this.push(chunk.toString().toUpperCase());
    callback();
  }
}

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

await pipeline(
  fs.createReadStream('input.txt'),
  new UpperCaseTransform(),
  fs.createWriteStream('output.txt')
);

CSV line parser

const { Transform } = require('stream');

class CSVParser extends Transform {
  constructor() {
    super({ objectMode: true }); // Output objects instead of buffers
    this.remainder = '';
    this.headers = null;
  }

  _transform(chunk, encoding, callback) {
    const data = this.remainder + chunk.toString();
    const lines = data.split('\n');
    this.remainder = lines.pop(); // Save incomplete last line

    for (const line of lines) {
      if (!line.trim()) continue;

      const values = line.split(',').map(v => v.trim());

      if (!this.headers) {
        this.headers = values;
        continue;
      }

      const record = {};
      this.headers.forEach((header, i) => {
        record[header] = values[i];
      });
      this.push(record);
    }

    callback();
  }

  _flush(callback) {
    if (this.remainder.trim() && this.headers) {
      const values = this.remainder.split(',').map(v => v.trim());
      const record = {};
      this.headers.forEach((header, i) => {
        record[header] = values[i];
      });
      this.push(record);
    }
    callback();
  }
}

// Usage
const parser = new CSVParser();
const readable = fs.createReadStream('data.csv');

readable.pipe(parser).on('data', (record) => {
  console.log(record); // { name: 'Alice', age: '30', city: 'NYC' }
});

Creating readable streams from data

const { Readable } = require('stream');

// From an array
const readable = Readable.from(['hello\n', 'world\n']);

// From an async generator
async function* generateData() {
  for (let i = 0; i < 1000; i++) {
    yield `Line ${i}\n`;
    if (i % 100 === 0) {
      await new Promise(resolve => setTimeout(resolve, 10));
    }
  }
}

const stream = Readable.from(generateData());

Backpressure explained

Backpressure occurs when a writable stream cannot process data as fast as a readable stream produces it. Without handling it, data piles up in memory.

How pipe() handles it:

1. Readable emits data.
2. Writable’s internal buffer fills up.
3. write() returns false.
4. Readable pauses automatically.
5. Writable drains its buffer and emits drain.
6. Readable resumes.

This is why pipe() and pipeline() are preferred over manual data event handling.

Real-world example: log file processor

Process a large log file, filter for errors, and write a summary:

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

async function processLogs(inputPath, outputPath) {
  const errorFilter = new Transform({
    transform(chunk, encoding, callback) {
      const line = chunk.toString();
      if (line.includes('ERROR')) {
        this.push(line + '\n');
      }
      callback();
    }
  });

  const input = fs.createReadStream(inputPath);
  const rl = readline.createInterface({ input });
  const output = fs.createWriteStream(outputPath);

  let count = 0;
  for await (const line of rl) {
    if (line.includes('ERROR')) {
      output.write(line + '\n');
      count++;
    }
  }

  output.end();
  console.log(`Found ${count} errors`);
}

processLogs('app.log', 'errors.log');

Summary

Streams let you process data incrementally, keeping memory usage constant regardless of data size. Use pipeline() instead of pipe() for proper error handling. Transform streams let you modify data in flight. Async iteration with for await...of is the cleanest way to consume readable streams. Always respect backpressure — either by using pipe/pipeline or by checking the return value of write() and waiting for drain.