Topic 14 of 55
Streams
Overview
Streams process data piece-by-piece without loading it all into memory — essential for large file processing, HTTP responses, and data transformation pipelines. They are 4x more memory-efficient than loading full files.
Syntax
javascript
import { createReadStream, createWriteStream } from "fs";
import { pipeline } from "stream/promises";
import { createGzip } from "zlib";
import { Transform } from "stream";
// Readable stream — produces data
const readStream = createReadStream("large-file.csv");
// Transform stream — processes data chunks
const upperCase = new Transform({
transform(chunk, encoding, callback) {
callback(null, chunk.toString().toUpperCase());
}
});
// Pipeline — safe streaming with error handling
await pipeline(
readStream, // source
createGzip(), // compress
createWriteStream("output.gz") // destination
);Common Pitfalls
- Use pipeline() instead of pipe() — pipeline() automatically destroys streams on error and handles cleanup; pipe() does not.
- Do not mix stream versions — use the modern Readable.from() and async iteration pattern for simplicity.
- Interview tip: Streams follow the backpressure pattern — readable pauses when writable is full. pipeline() handles this automatically.
Real-World Example
Processing a 1GB CSV file without loading it into memory
example
javascript
import { createReadStream } from "fs";
import { createInterface } from "readline";
async function processLargeCSV(inputPath) {
const rl = createInterface({
input: createReadStream(inputPath),
crlfDelay: Infinity,
});
let isHeader = true;
let headers = [];
let batch = [];
let lineCount = 0;
for await (const line of rl) {
if (isHeader) {
headers = line.split(",");
isHeader = false;
continue;
}
const values = line.split(",");
const record = Object.fromEntries(headers.map((h, i) => [h, values[i]]));
batch.push(record);
lineCount++;
if (batch.length >= 1000) {
await db.bulkInsert(batch); // process in batches
batch = [];
}
}
if (batch.length > 0) await db.bulkInsert(batch);
console.log("Processed", lineCount, "records");
}