Node.js Streams — Complete Guide to Stream API, Pipe, and Data Processing

In this tutorial, you will learn about Node.js Streams. We cover key concepts, practical examples, and best practices to help you master this topic.

Node.js streams Process data in chunks, enabling memory-efficient handling of large files, network responses, and data transformations without loading everything into memory.

What You'll Learn

By the end of this tutorial, you'll use readable, writable, and transform streams, pipe data between streams, handle backpressure, and process large datasets efficiently.

Why Streams Matter

Loading a 1GB file into memory crashes a server. Streams process data piece by piece, using constant memory regardless of file size. This is critical for video processing, log analysis, and file uploads in tools like DodaZIP.

Real-World Use

A video transcoding service reads a large video file via a readable stream, compresses it through a transform stream, and writes the output to disk, using less than 50MB of memory for a 4GB file.

Streams Learning Path

flowchart LR
  A[File System] --> B[Streams]
  B --> C[Buffers]
  C --> D[Events]
  D --> E[Express.js]
  A --> F{You Are Here}
  style F fill:#f90,color:#fff

Types of Streams

import { Readable, Writable, Transform, Duplex, PassThrough } from "node:stream";

Reading from a Stream

import fs from "node:fs";
const readStream = fs.createReadStream("large-file.txt", { encoding: "utf8", highWaterMark: 16384 });
readStream.on("data", (chunk) => {
  console.log(`Received ${chunk.length} bytes`);
});
readStream.on("end", () => {
  console.log("File read complete");
});
readStream.on("error", (err) => {
  console.error("Stream error:", err.message);
});

Writing to a Stream

import fs from "node:fs";
const writeStream = fs.createWriteStream("output.txt");
writeStream.write("First line\n");
writeStream.write("Second line\n");
writeStream.end("Final line\n");
writeStream.on("finish", () => console.log("Write complete"));

Piping Streams

Pipe connects a readable stream to a writable stream, managing backpressure automatically.

import fs from "node:fs";
import zlib from "node:zlib";
const readStream = fs.createReadStream("input.txt");
const gzipTransform = zlib.createGzip();
const writeStream = fs.createWriteStream("input.txt.gz");
readStream.pipe(gzipTransform).pipe(writeStream);
writeStream.on("finish", () => console.log("File compressed"));

Transform Stream Example

A transform stream modifies data as it passes through.

import { Transform } from "node:stream";
const upperCaseTransform = new Transform({
  transform(chunk, encoding, callback) {
    this.push(chunk.toString().toUpperCase());
    callback();
  }
});
process.stdin.pipe(upperCaseTransform).pipe(process.stdout);

Backpressure

When a readable stream pushes data faster than the writable stream can handle, backpressure occurs. The pipe method handles this automatically.

const readable = getReadableStream();
const writable = getWritableStream();
readable.pipe(writable); // Automatic backpressure management
readable.on("error", () => writable.destroy());
writable.on("error", () => readable.destroy());

Stream Utilities

import { pipeline } from "node:stream/promises";
import fs from "node:fs";
import zlib from "node:zlib";
try {
  await pipeline(
    fs.createReadStream("input.txt"),
    zlib.createGzip(),
    fs.createWriteStream("input.txt.gz")
  );
  console.log("Pipeline completed");
} catch (err) {
  console.error("Pipeline failed:", err);
}

Pipeline with promises provides cleaner error handling than manual piping.

Common Mistakes

1. Not Handling Errors on Streams

An unhandled error on a stream crashes the process. Always add error listeners or use pipeline.

2. Reading Entire Stream Into Memory

Stream chunks accumulate if you collect them in an array. Process data chunk by chunk instead.

3. Forgetting to Call callback() in Transform

The transform function must call callback() or data flow stops silently.

4. Ignoring Backpressure

Without pipe or proper drain handling, memory usage grows unbounded when writable is slower than readable.

5. Creating Streams Without highWaterMark

Default buffer size (16KB for streams, 64KB for files) may not suit your data size profile.

Practice Questions

1. What is backpressure in streams?

Backpressure occurs when the writable stream cannot keep up with the readable stream. pipe() manages this automatically by pausing the readable.

2. What is the difference between pipe and pipeline?

pipe returns the writable stream for chaining but doesn't forward errors. pipeline manages cleanup and error propagation correctly.

3. How do you create a custom transform stream?

Extend Transform class or pass a transform function to the constructor. Call this.push() with transformed data.

4. When should you use a stream instead of readFile?

Use streams for large files (over 100MB), continuous data, or when memory is constrained.

5. Challenge: Create a transform stream that counts lines passing through it.

import { Transform } from "node:stream";
class LineCounter extends Transform {
  constructor() { super({ objectMode: true }); this.lines = 0; }
  _transform(chunk, encoding, callback) {
    this.lines += chunk.toString().split("\n").length - 1;
    this.push(chunk);
    callback();
  }
  _flush(callback) {
    this.push(`\n--- Total lines: ${this.lines} ---\n`);
    callback();
  }
}
process.stdin.pipe(new LineCounter()).pipe(process.stdout);

FAQ

Mini Project: CSV Processor

Build a transform stream that reads a CSV file, converts it to JSON lines, and writes the output.

import { Transform } from "node:stream";
import { pipeline } from "node:stream/promises";
import fs from "node:fs";
class CsvToJson extends Transform {
  constructor() {
    super({ objectMode: true });
    this.headers = null;
  }
  _transform(chunk, encoding, callback) {
    const line = chunk.toString().trim();
    if (!line) return callback();
    if (!this.headers) {
      this.headers = line.split(",");
      return callback();
    }
    const values = line.split(",");
    const obj = {};
    this.headers.forEach((h, i) => obj[h] = values[i]);
    this.push(JSON.stringify(obj) + "\n");
    callback();
  }
}
await pipeline(
  fs.createReadStream("data.csv"),
  new CsvToJson(),
  fs.createWriteStream("data.json")
);

What's Next

Node.js Buffers Node.js Events Express.js