Node.js Stream Processing Patterns

Intermediatev1.0.0

Master Node.js streams for processing large files, HTTP responses, and data pipelines efficiently without loading everything into memory.

Content

Overview

Node.js streams process data in chunks, enabling you to handle gigabytes of data with minimal memory usage. This skill covers readable/writable/transform streams, pipelines, and real-world streaming patterns.

Why This Matters

-Memory efficiency — process 10GB files with 64KB of memory
-Time to first byte — start processing before the entire payload arrives
-Backpressure — automatically slow producers when consumers can't keep up
-Composability — chain transforms like Unix pipes

Stream Types

javascript

import { Readable, Writable, Transform, pipeline } from 'node:stream';
import { pipeline as pipelineAsync } from 'node:stream/promises';

Step 1: Read Large Files as Streams

javascript

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

// Process a large file line by line
const rl = createInterface({
  input: createReadStream('large-file.csv'),
  crlfDelay: Infinity,
});

for await (const line of rl) {
  // Process each line — memory usage stays constant
  const [name, email] = line.split(',');
  await processUser({ name, email });
}

Step 2: Transform Streams

javascript

import { Transform } from 'node:stream';

const toUpperCase = new Transform({
  transform(chunk, encoding, callback) {
    this.push(chunk.toString().toUpperCase());
    callback();
  }
});

// Or use the simpler object mode
const parseCSV = new Transform({
  objectMode: true,
  transform(chunk, encoding, callback) {
    const lines = chunk.toString().split('\n');
    for (const line of lines) {
      if (line.trim()) {
        this.push(line.split(','));
      }
    }
    callback();
  }
});

Step 3: Pipeline for Safe Piping

javascript

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

// Compress a file with proper error handling and cleanup
await pipeline(
  createReadStream('input.log'),
  createGzip(),
  createWriteStream('input.log.gz')
);
// All streams are properly closed on success or error

Step 4: HTTP Streaming Response

javascript

import { createReadStream } from 'node:fs';

app.get('/download/:file', (req, res) => {
  const filePath = getValidatedPath(req.params.file);
  res.setHeader('Content-Type', 'application/octet-stream');

  const stream = createReadStream(filePath);
  stream.pipe(res);

  stream.on('error', (err) => {
    if (!res.headersSent) {
      res.status(500).json({ error: 'File read error' });
    }
    stream.destroy();
  });
});

Best Practices

-Always use pipeline() from node:stream/promises instead of .pipe()
-Handle errors on every stream in the pipeline
-Use object mode for structured data (JSON objects, parsed records)
-Set highWaterMark to control chunk size and memory usage
-Use for await...of for simple consumption of readable streams
-Destroy streams explicitly when done to free resources

Common Mistakes

-Using .pipe() without error handling (errors don't propagate)
-Reading entire files into memory with readFile when streaming would work
-Not handling backpressure (write returns false when buffer is full)
-Forgetting to end writable streams (call .end())

FAQ

Discussion

Loading comments...

Step 1: Read Large Files as Streams

javascript

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

// Process a large file line by line
const rl = createInterface({
  input: createReadStream('large-file.csv'),
  crlfDelay: Infinity,
});

for await (const line of rl) {
  // Process each line — memory usage stays constant
  const [name, email] = line.split(',');
  await processUser({ name, email });
}

Step 2: Transform Streams

javascript

import { Transform } from 'node:stream';

const toUpperCase = new Transform({
  transform(chunk, encoding, callback) {
    this.push(chunk.toString().toUpperCase());
    callback();
  }
});

// Or use the simpler object mode
const parseCSV = new Transform({
  objectMode: true,
  transform(chunk, encoding, callback) {
    const lines = chunk.toString().split('\n');
    for (const line of lines) {
      if (line.trim()) {
        this.push(line.split(','));
      }
    }
    callback();
  }
});

Step 3: Pipeline for Safe Piping

javascript

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

// Compress a file with proper error handling and cleanup
await pipeline(
  createReadStream('input.log'),
  createGzip(),
  createWriteStream('input.log.gz')
);
// All streams are properly closed on success or error

Step 4: HTTP Streaming Response

javascript

import { createReadStream } from 'node:fs';

app.get('/download/:file', (req, res) => {
  const filePath = getValidatedPath(req.params.file);
  res.setHeader('Content-Type', 'application/octet-stream');

  const stream = createReadStream(filePath);
  stream.pipe(res);

  stream.on('error', (err) => {
    if (!res.headersSent) {
      res.status(500).json({ error: 'File read error' });
    }
    stream.destroy();
  });
});

Best Practices

-Always use pipeline() from node:stream/promises instead of .pipe()

-Handle errors on every stream in the pipeline

-Use object mode for structured data (JSON objects, parsed records)

-Set highWaterMark to control chunk size and memory usage

-Use for await...of for simple consumption of readable streams

-Destroy streams explicitly when done to free resources