Streams let you process data piece by piece instead of loading everything into memory. Essential for handling large files and real-time data.
Why Streams?
// Bad: loads entire file into memory
const data = fs.readFileSync("huge-file.csv"); // 2GB in memory!
// Good: processes chunk by chunk
const stream = fs.createReadStream("huge-file.csv");
stream.on("data", (chunk) => processChunk(chunk)); // ~64KB at a timeStream Types
- Readable: Source of data (fs.createReadStream, http.IncomingMessage)
- Writable: Destination (fs.createWriteStream, http.ServerResponse)
- Transform: Modify data passing through (zlib.createGzip)
- Duplex: Both readable and writable (net.Socket)
The pipe() Method
// Compress a file
fs.createReadStream("input.txt")
.pipe(zlib.createGzip())
.pipe(fs.createWriteStream("input.txt.gz"));
// HTTP response streaming
app.get("/large-file", (req, res) => {
const stream = fs.createReadStream("huge.csv");
res.setHeader("Content-Type", "text/csv");
stream.pipe(res);
});Transform Streams
const { Transform } = require("stream");
const upperCase = new Transform({
transform(chunk, encoding, callback) {
this.push(chunk.toString().toUpperCase());
callback();
}
});
process.stdin.pipe(upperCase).pipe(process.stdout);Modern Async Iteration
async function processCSV(filePath) {
const stream = fs.createReadStream(filePath, { encoding: "utf8" });
let lineBuffer = "";
for await (const chunk of stream) {
lineBuffer += chunk;
const lines = lineBuffer.split("\\n");
lineBuffer = lines.pop(); // Keep incomplete line
for (const line of lines) {
await processLine(line);
}
}
}Backpressure
When the writable stream can't keep up with the readable stream, backpressure builds up. pipe() handles this automatically. If using manual events, check the return value of write() and wait for the drain event.