Transform Streams
Build reusable, composable stream stages that reshape data on the fly.
A Transform is a duplex stream: data written in one end is reshaped and read out the other. It is the unit of a streaming pipeline — parse, filter, map, aggregate — each stage doing one job and composing with the next.
You implement _transform(chunk, encoding, callback), calling this.push(value) for each output and callback() when the chunk is handled. An optional _flush(callback) lets you emit trailing data at the end (a final aggregate, a closing bracket).
const { Transform } = require('node:stream');
const upper = new Transform({
transform(chunk, _enc, cb) {
cb(null, chunk.toString().toUpperCase()); // shorthand: push via cb
},
});
process.stdin.pipe(upper).pipe(process.stdout);Object-mode transforms
With objectMode: true, a transform can turn text lines into parsed records, drop invalid rows, enrich each with a DB lookup, and count — all as separate, testable stages you snap together in a pipeline.
Example
// A composable object-mode pipeline: raw text -> lines -> JSON -> filtered
// -> stringified NDJSON, each stage independently testable.
const { Transform } = require('node:stream');
// Split incoming buffers into complete lines, buffering partial tails.
class LineSplitter extends Transform {
constructor() { super({ readableObjectMode: true }); this.tail = ''; }
_transform(chunk, _enc, cb) {
const parts = (this.tail + chunk).split('\n');
this.tail = parts.pop(); // last piece may be incomplete
for (const line of parts) if (line) this.push(line);
cb();
}
_flush(cb) { if (this.tail) this.push(this.tail); cb(); } // emit final line
}
// Parse + drop malformed rows instead of crashing the whole stream.
const parseJson = new Transform({
objectMode: true,
transform(line, _enc, cb) {
try { cb(null, JSON.parse(line)); }
catch { cb(); } // skip bad line, keep the pipeline alive
},
});
const onlyActive = new Transform({
objectMode: true,
transform(obj, _enc, cb) { cb(null, obj.active ? obj : undefined); },
});
const { pipeline } = require('node:stream/promises');
const { Readable } = require('node:stream');
await pipeline(
Readable.from(['{"id":1,"active":true}\n{bad}\n{"id":2,"active":false}\n']),
new LineSplitter(),
parseJson,
onlyActive,
async function* (source) { for await (const o of source) yield JSON.stringify(o) + '\n'; },
process.stdout,
);When to use it
- A log parser implements a Transform stream that reads raw HTTP access-log lines and emits structured JSON objects for downstream processing.
- A compression middleware wraps `zlib.createGzip()` in a Transform stream to compress HTTP response bodies on-the-fly as they are streamed.
- A CSV-to-JSON converter is built as a Transform stream so it can be inserted into any pipeline between a file source and a database sink.
More examples
Simple Transform stream
Creates a Transform stream that parses each input chunk as a number, doubles it, and pushes the result.
const { Transform } = require('stream');
const doubler = new Transform({
transform(chunk, enc, cb) {
const num = parseInt(chunk.toString(), 10);
this.push(String(num * 2) + '\n');
cb();
},
});
process.stdin.pipe(doubler).pipe(process.stdout);
// Echo: 5 -> '10'Gzip Transform with zlib
Uses zlib's built-in Gzip Transform inside `stream.pipeline` to compress a file without intermediate buffers.
const fs = require('fs');
const zlib = require('zlib');
const { pipeline } = require('stream/promises');
await pipeline(
fs.createReadStream('data.json'),
zlib.createGzip(),
fs.createWriteStream('data.json.gz')
);
console.log('Compressed');Line-splitting Transform
Implements a Transform that buffers partial lines across chunks and emits one complete line at a time.
const { Transform } = require('stream');
class Liner extends Transform {
constructor() { super(); this._buf = ''; }
_transform(chunk, enc, cb) {
const lines = (this._buf + chunk).split('\n');
this._buf = lines.pop(); // keep incomplete line
lines.forEach(l => this.push(l));
cb();
}
_flush(cb) { if (this._buf) this.push(this._buf); cb(); }
}
fs.createReadStream('log.txt').pipe(new Liner()).on('data', console.log);
Discussion