Node.js Streams — Complete Guide to Memory-Efficient Data Processing 2026

Introduction

Why This Matters

Loading a 2 GB CSV file into memory with fs.readFile will crash your Node.js process or trigger costly garbage collection pauses. Streams solve this by processing data chunk-by-chunk, keeping memory usage constant regardless of file size.

Streams are also foundational to Node.js internals — HTTP requests and responses, TCP sockets, file system reads and writes, and process.stdin/stdout are all streams. Understanding them deeply unlocks efficient I/O patterns that are impossible to replicate with buffered approaches.

Stream Types and When to Use Each

Node.js has four stream types:

Type	Direction	Example
Readable	Source of data	`fs.createReadStream`, HTTP request
Writable	Destination for data	`fs.createWriteStream`, HTTP response
Duplex	Both read and write	TCP socket
Transform	Read, modify, write	Gzip compression, encryption

import fs from 'fs';
import { pipeline } from 'stream/promises';
import zlib from 'zlib';
 
// Pipe a readable through a transform into a writable
// pipeline handles cleanup automatically on error
await pipeline(
  fs.createReadStream('large-input.txt'),
  zlib.createGzip(),
  fs.createWriteStream('output.txt.gz')
);
 
console.log('Compressed with constant memory usage');

Always prefer stream/promises pipeline over .pipe() — pipeline correctly propagates errors and cleans up streams.

Building Custom Readable Streams

import { Readable, ReadableOptions } from 'stream';
 
interface PaginatedSource {
  fetchPage(page: number, pageSize: number): Promise<Record<string, unknown>[]>;
}
 
class DatabaseReadStream extends Readable {
  private page = 0;
  private pageSize: number;
  private done = false;
 
  constructor(
    private source: PaginatedSource,
    options: ReadableOptions & { pageSize?: number } = {}
  ) {
    super({ objectMode: true, ...options });
    this.pageSize = options.pageSize ?? 100;
  }
 
  override async _read(): Promise<void> {
    if (this.done) {
      this.push(null); // Signal end of stream
      return;
    }
 
    try {
      const rows = await this.source.fetchPage(this.page++, this.pageSize);
      if (rows.length === 0) {
        this.done = true;
        this.push(null);
      } else {
        for (const row of rows) {
          this.push(row); // Push each record as an object
        }
      }
    } catch (err) {
      this.destroy(err as Error);
    }
  }
}

Object mode (objectMode: true) lets streams carry arbitrary JavaScript objects instead of Buffers or strings — essential for database row streaming.

Building Custom Transform Streams

Transform streams are the workhorses of data pipelines — they consume input, modify it, and produce output.

import { Transform, TransformCallback } from 'stream';
 
interface CsvRow {
  [key: string]: string;
}
 
class CsvParserTransform extends Transform {
  private headers: string[] = [];
  private buffer = '';
 
  constructor() {
    super({ readableObjectMode: true, writableObjectMode: false });
  }
 
  override _transform(
    chunk: Buffer,
    _encoding: BufferEncoding,
    callback: TransformCallback
  ): void {
    this.buffer += chunk.toString('utf8');
    const lines = this.buffer.split('\n');
    this.buffer = lines.pop() ?? ''; // Keep incomplete last line
 
    for (const line of lines) {
      const trimmed = line.trim();
      if (!trimmed) continue;
 
      const values = trimmed.split(',');
 
      if (this.headers.length === 0) {
        this.headers = values;
      } else {
        const row: CsvRow = {};
        this.headers.forEach((h, i) => {
          row[h] = values[i] ?? '';
        });
        this.push(row); // Push parsed object downstream
      }
    }
    callback();
  }
 
  override _flush(callback: TransformCallback): void {
    if (this.buffer.trim()) {
      const values = this.buffer.split(',');
      const row: CsvRow = {};
      this.headers.forEach((h, i) => {
        row[h] = values[i] ?? '';
      });
      this.push(row);
    }
    callback();
  }
}

The _flush method is called when the writable side ends — use it to process any buffered incomplete chunks.

Backpressure: The Most Important Stream Concept

Backpressure prevents a fast producer from overwhelming a slow consumer. Without it, data queues in memory indefinitely.

import { Writable, WritableOptions } from 'stream';
 
class SlowDatabaseWriter extends Writable {
  private batch: object[] = [];
 
  constructor(options: WritableOptions = {}) {
    super({ objectMode: true, highWaterMark: 50, ...options });
  }
 
  override async _write(
    chunk: object,
    _encoding: BufferEncoding,
    callback: (error?: Error | null) => void
  ): Promise<void> {
    this.batch.push(chunk);
 
    if (this.batch.length >= 50) {
      try {
        await this.flushBatch();
        callback();
      } catch (err) {
        callback(err as Error);
      }
    } else {
      callback();
    }
  }
 
  override async _final(callback: (error?: Error | null) => void): Promise<void> {
    try {
      await this.flushBatch();
      callback();
    } catch (err) {
      callback(err as Error);
    }
  }
 
  private async flushBatch(): Promise<void> {
    if (this.batch.length === 0) return;
    // await db.batchInsert(this.batch);
    this.batch = [];
  }
}

highWaterMark controls the buffer size. When the internal buffer exceeds this threshold, writable.write() returns false — a signal to the producer to pause. pipeline handles this automatically.

Full Data Pipeline Example

Streaming a large CSV from S3, parsing it, transforming records, and writing to a database:

import { pipeline } from 'stream/promises';
 
async function importCsvToDatabase(filePath: string): Promise<void> {
  const readStream = fs.createReadStream(filePath, { highWaterMark: 64 * 1024 });
  const csvParser = new CsvParserTransform();
  const dbWriter = new SlowDatabaseWriter();
 
  const startTime = Date.now();
  let count = 0;
 
  // Count records inline with a pass-through transform
  const counter = new Transform({
    objectMode: true,
    transform(chunk, _enc, cb) {
      count++;
      this.push(chunk);
      cb();
    },
  });
 
  await pipeline(readStream, csvParser, counter, dbWriter);
 
  const duration = ((Date.now() - startTime) / 1000).toFixed(1);
  console.log(`Imported ${count} records in ${duration}s`);
}

Memory usage stays flat (a few MB) regardless of whether the file has 1,000 or 10,000,000 rows.

Common Mistakes

Using .pipe() without error handling — errors on individual streams will not propagate. Always use pipeline from stream/promises.
Ignoring backpressure when not using pipeline — if you manually call .write() without checking the return value, you will buffer unbounded data.
Mixing object mode and byte mode without explicit flags — readableObjectMode and writableObjectMode must be set separately for Transform streams with different modes on each side.
Forgetting _flush — any buffered data at stream end must be emitted in _flush, otherwise the last chunk is silently dropped.
Not destroying streams on error — failing to call this.destroy(err) in _read or _transform leaves downstream consumers hanging.

Best Practices

Always use stream/promises pipeline — it handles cleanup, backpressure, and error propagation automatically.
Set highWaterMark deliberately: larger values improve throughput, smaller values reduce memory pressure.
Use object mode for application-level data (parsed rows, events); keep byte mode for raw I/O (files, network).
Prefer async iteration (for await...of) over event listeners for consuming readable streams in modern Node.js.
Monitor stream events (drain, finish, error, close) in production to detect stalled pipelines.
Test streams with Readable.from([...]) to create test fixtures without touching the filesystem.

Key Takeaways

Node.js streams process data in chunks, keeping memory usage constant regardless of total data size.
The four stream types are Readable, Writable, Duplex, and Transform — each serving a distinct role in data pipelines.
pipeline from stream/promises is the correct API for composing streams — it handles errors, cleanup, and backpressure automatically.
Backpressure prevents fast producers from overwhelming slow consumers; highWaterMark controls the buffer threshold.
Custom Transform streams require implementing _transform for chunk processing and _flush for end-of-stream buffer draining.
Object mode allows streams to carry arbitrary JavaScript objects, enabling typed database row streaming without serialization.
for await...of on a readable stream is the idiomatic modern way to consume streams with async/await syntax.
HTTP requests and responses in Node.js are streams — understanding streams directly improves HTTP handling efficiency.

Introduction#

Why This Matters#

Stream Types and When to Use Each#

Building Custom Readable Streams#

Building Custom Transform Streams#

Backpressure: The Most Important Stream Concept#

Full Data Pipeline Example#

Common Mistakes#

Best Practices#

Key Takeaways#

Related reading