Node.js Streams

Process data piece by piece with streams: read and write large files without loading them into memory, transform data on the fly, connect streams safely with pipeline, read lines with readline, compress with zlib, and stream HTTP responses.

Lesson 17 of 43Intermediate17 min read

Imagine processing a 5 GB log file, or letting users download a 2 GB video. Reading the whole thing into memory with readFile would need gigabytes of RAM for every request, and nothing could be sent until everything was read. Streams solve this by handling data in small chunks, one after another, so memory use stays small and work starts immediately.

Streams are everywhere in Node. HTTP requests and responses are streams, files can be read and written as streams, compression and encryption are streams, and process.stdin and process.stdout are streams too. There are four kinds: readable streams produce data, writable streams consume it, duplex streams do both, and transform streams change data as it passes through.

The real power comes from connecting them. pipeline joins a readable source, any number of transforms and a writable destination into one chain, handles errors from every part, and manages backpressure, slowing a fast source when the destination can't keep up. In this lesson you'll read and write file streams, iterate over chunks and lines, write your own transform, compress files and stream responses to HTTP clients.

Reading and writing with streams

JavaScript
import { createReadStream, createWriteStream } from 'node:fs';

const input = createReadStream('access.log', { encoding: 'utf8' });
for await (const chunk of input) {
  console.log(`Got ${chunk.length} characters`);
}

const out = createWriteStream('report.txt');
out.write('Line 1\n');
out.end('Last line\n');          // finish the stream
  • Readable streams can be consumed with for await, the simplest modern way.
  • Chunks are 64 KB by default for files; highWaterMark changes the size.
  • Writable streams have write(chunk) and end(). Always end them, or the file may not be completed.
  • Streams are event emitters: data, end, error, finish and close events are still common in older code.

pipeline and transforms

pipeline from node:stream/promises connects streams and returns a promise:

JavaScript
import { pipeline } from 'node:stream/promises';
import { createGzip } from 'node:zlib';

await pipeline(
  createReadStream('big.log'),
  createGzip(),
  createWriteStream('big.log.gz'),
);
  • If any stream fails, pipeline destroys the others and rejects, so files and sockets aren't left open.
  • A Transform stream receives chunks, changes them, and pushes results onward. Write one with new Transform({ transform(chunk, encoding, callback) { ... } }), or pass an async generator function into pipeline as a step.
  • Readable.from(iterable) turns an array or generator into a readable stream.

Lines and backpressure

Chunks don't respect line boundaries, so a chunk may end in the middle of a line. readline handles this for you:

JavaScript
import { createInterface } from 'node:readline';
const rl = createInterface({ input: createReadStream('data.csv'), crlfDelay: Infinity });
for await (const line of rl) { /* one complete line at a time */ }

Backpressure happens when data arrives faster than it can be written, for example reading from a fast disk and sending over a slow network. write() returns false when the destination's buffer is full. pipeline and for await handle this automatically; manual write loops should wait for the drain event.

Node.js Advanced Syntax

Stream essentials

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

for await (const chunk of createReadStream('a.txt', 'utf8')) { }
await pipeline(createReadStream('a'), createGzip(), createWriteStream('a.gz'));
const upper = new Transform({ transform(chunk, enc, cb) { cb(null, chunk.toString().toUpperCase()); } });
await pipeline(Readable.from(['a', 'b']), upper, process.stdout);
for await (const line of createInterface({ input: createReadStream('x.csv') })) { }

Node.js Advanced Summary

  • Streams process data in chunks, keeping memory low and starting work immediately.
  • Readable, writable, duplex and transform are the four kinds of stream.
  • Consume readable streams with for await; end writable streams with end().
  • pipeline from node:stream/promises connects streams and handles errors and backpressure.
  • readline reads complete lines; zlib's createGzip compresses streams.

Node.js Advanced Examples

Each example shows the code, the output it produces and what each part does.

Example 1

Processing a large file line by line

Node.js
import { createReadStream, createWriteStream } from 'node:fs';
import { createInterface } from 'node:readline';
import { once } from 'node:events';

const out = createWriteStream('access-demo.log');
for (let i = 1; i <= 50000; i++) {
  const status = i % 97 === 0 ? 500 : i % 10 === 0 ? 404 : 200;
  out.write(`2026-03-01 GET /page/${i % 25} ${status}\n`);
}
out.end();
await once(out, 'finish');

const counts = {};
const rl = createInterface({ input: createReadStream('access-demo.log'), crlfDelay: Infinity });
for await (const line of rl) {
  const status = line.split(' ').at(-1);
  counts[status] = (counts[status] ?? 0) + 1;
}

console.log(counts);
console.log(`Memory used stays small: ${process.memoryUsage().heapUsed < 200 * 1024 * 1024}`);
Output
{ '200': 44536, '404': 4949, '500': 515 }
Memory used stays small: true
Line-by-line explanation
  1. Lines 5 to 10The demo writes a 50,000-line log file through a write stream.
  2. Line 11once waits for the finish event, which means everything has been written.
  3. Line 14readline turns the stream's chunks into complete lines.
  4. Line 15for await reads one line at a time, so the file is never fully in memory.
  5. Lines 16 to 17Each line updates the counts and is then discarded.
Example 2

A transform stream with pipeline

Node.js
import { pipeline } from 'node:stream/promises';
import { Readable, Transform, Writable } from 'node:stream';

const orders = [
  { id: 1, customer: 'ada', total: 59.9 },
  { id: 2, customer: 'ben', total: 12.5 },
  { id: 3, customer: 'cy', total: 210 },
];

const toCsv = new Transform({
  objectMode: true,
  transform(order, encoding, callback) {
    const line = `${order.id},${order.customer.toUpperCase()},${order.total.toFixed(2)}\n`;
    callback(null, line);
  },
});

let output = 'id,customer,total\n';
const collect = new Writable({
  write(chunk, encoding, callback) {
    output += chunk.toString();
    callback();
  },
});

await pipeline(
  Readable.from(orders),
  async function* (source) {
    for await (const order of source) if (order.total >= 20) yield order;
  },
  toCsv,
  collect,
);
console.log(output.trimEnd());
Output
id,customer,total
1,ADA,59.90
3,CY,210.00
Line-by-line explanation
  1. Lines 10 to 16A Transform turns each order object into a CSV line; objectMode lets it receive objects instead of bytes.
  2. Line 14callback(null, data) passes the result to the next stream.
  3. Lines 19 to 24A Writable collects the output; usually this would be a file or an HTTP response.
  4. Line 27Readable.from turns an array into a readable stream.
  5. Lines 28 to 30An async generator works as a filtering step inside the pipeline.
  6. Lines 26 to 33pipeline connects everything and rejects if any step fails.
Example 3

Compressing and streaming over HTTP

Node.js
import { createServer, get } from 'node:http';
import { pipeline } from 'node:stream/promises';
import { Readable } from 'node:stream';
import { createGzip, gunzipSync } from 'node:zlib';

function* rows(count) {
  yield 'id,product,qty\n';
  for (let i = 1; i <= count; i++) yield `${i},Item ${i % 7},${(i % 5) + 1}\n`;
}

const server = createServer(async (req, res) => {
  res.writeHead(200, { 'Content-Type': 'text/csv', 'Content-Encoding': 'gzip' });
  try {
    await pipeline(Readable.from(rows(20000)), createGzip(), res);
  } catch (err) {
    console.error('Export failed', err.message);
  }
});

server.listen(0, () => {
  get(`http://localhost:${server.address().port}/export.csv`, async (res) => {
    const parts = [];
    for await (const chunk of res) parts.push(chunk);
    const raw = Buffer.concat(parts);
    const text = gunzipSync(raw).toString();
    console.log(`Compressed: ${raw.length < text.length / 3}`);
    console.log(`Lines: ${text.trim().split('\n').length}`);
    console.log(text.split('\n')[1]);
    server.close();
  });
});
Output
Compressed: true
Lines: 20001
1,Item 1,2
Line-by-line explanation
  1. Lines 6 to 9A generator produces CSV rows one at a time instead of building one huge string.
  2. Line 14The rows are compressed and streamed straight into the HTTP response.
  3. Lines 15 to 17If the client disconnects, pipeline rejects and every stream is cleaned up.
  4. Lines 21 to 25The demo downloads the raw compressed bytes with http.get and decompresses them to check.
  5. Line 26Repetitive text like CSV compresses to a fraction of its size.

Common Node.js Advanced Mistakes

Reading huge files into memory

Don't
const data = await readFile('video.mp4');
res.end(data);                          // gigabytes of memory per request
Do
await pipeline(createReadStream('video.mp4'), res);

Using .pipe() without error handling

Don't
createReadStream('in.txt').pipe(createGzip()).pipe(createWriteStream('out.gz'));   // errors are lost, files left open
Do
await pipeline(createReadStream('in.txt'), createGzip(), createWriteStream('out.gz'));

Splitting chunks into lines by hand

Don't
for await (const chunk of stream) {
  chunk.toString().split('\n').forEach(handleLine);   // lines split across chunks get broken
}
Do
for await (const line of createInterface({ input: stream })) handleLine(line);

Node.js Advanced in Real Projects

Streams in production

File upload handlers stream request bodies straight to disk or cloud storage without buffering them. Export endpoints stream CSV or JSON from a database cursor directly into the response. Log processors, video servers, backup tools and ETL jobs are built on streams, and the AWS SDK, database drivers and HTTP clients all expose streams for large data.

Node.js Advanced FAQ

What are the four types of streams in Node.js?

Readable streams produce data, such as a file being read or an incoming request. Writable streams consume it, such as a file being written or an HTTP response. Duplex streams are both at once, like a network socket, and Transform streams are duplex streams that change data as it passes through, like gzip compression.

What is the difference between pipe() and pipeline()?

readable.pipe(writable) connects two streams, but if one of them fails, the others aren't closed and the error isn't passed along, which can leak file handles and connections. pipeline from node:stream/promises connects any number of streams, destroys all of them if one fails, and returns a promise you can await inside try/catch. Use pipeline in new code.

How do I read a whole stream into a string?

Use the helpers in node:stream/consumers: const body = await text(stream) collects everything into a string, and await json(stream) parses it as JSON. A for await loop that adds each chunk to a string does the same job by hand. Only do this when you know the data is small, since it defeats the memory benefit of streaming.

How do I stream a file as an Express response?

The response object is a writable stream, so pass it to pipeline: await pipeline(createReadStream(file), res), after setting the Content-Type header. For ordinary files on disk, res.sendFile does the same thing and also handles caching headers and range requests for you.

What does objectMode mean in Node.js streams?

Normal streams carry strings or Buffers, and their buffer size is measured in bytes. A stream created with objectMode: true carries any JavaScript values, such as one parsed CSV row per chunk, and its highWaterMark counts objects instead of bytes. Readable.from(array) creates an object-mode stream by default.

Node.js Advanced Exercises

Answer each one and check it instantly. If you get stuck, open the hint before the answer.

Fill in the blankseasy

Complete the line to connect the streams safely.

await (createReadStream('a.txt'), createGzip(), createWriteStream('a.txt.gz'));
Multiple choiceeasy

What's the main benefit of streams for large files?

Multiple choicemedium

Which kind of stream changes data as it passes through?

Multiple choicemedium

Why use readline instead of splitting chunks on newlines yourself?

Multiple choicehard

What is backpressure?

Node.js Advanced Quiz

5 questions to check what you learned.

  1. Which function creates a readable stream from a file?

    • createReadStream
    • readFile
    • openStream
    • streamFile

  2. What does pipeline do if one of its streams fails?

    • Continues with the others
    • Restarts the stream
    • Nothing
    • Destroys all the streams and rejects with the error

  3. How do you finish writing to a writable stream?

    • close()
    • stop()
    • finish()
    • end()

  4. What does Readable.from(array) do?

    • Turns the array (or any iterable) into a readable stream
    • Copies the array
    • Reads a file
    • Sorts the array

  5. Is an HTTP response object a stream?

    • No
    • Only for files
    • Yes, a writable stream
    • Only with Express

Related Node.js topics

04File SystemRead and write files with the fs/promises API, append to logs, work with JSON files, create and list folders, check whether files exist, handle errors like ENOENT, and know when to use the synchronous and callback versions.06EventsUse Node's event-driven style with EventEmitter: listen with on and once, emit events with data, remove listeners, handle the special error event, extend EventEmitter in your own classes, and await events with once from node:events.09HTTP ServerBuild a web server with Node's built-in http module: handle requests and send responses, set status codes and headers, route by method and URL, read query strings and JSON bodies, and call other servers with fetch.
Esc
↑↓ to navigateEnter to open