'use strict';
const { Transform } = require('stream');
class LineSplitter extends Transform {
constructor(options = {}) {
super({ ...options, readableObjectMode: true });
this.tail = '';
this.lineNo = 0;
}
_transform(chunk, _enc, cb) {
const data = this.tail + chunk.toString('utf8');
const lines = data.split('\n');
// Last element is a partial line (or '') until the next chunk arrives.
this.tail = lines.pop();
for (const text of lines) {
this.push({ lineNo: ++this.lineNo, text, bytes: Buffer.byteLength(text) + 1 });
}
cb();
}
_flush(cb) {
if (this.tail.length > 0) {
this.push({ lineNo: ++this.lineNo, text: this.tail, bytes: Buffer.byteLength(this.tail) });
}
cb();
}
}
module.exports = { LineSplitter };
'use strict';
const fs = require('fs');
const path = require('path');
const { once } = require('events');
const { Transform } = require('stream');
const DATE_RE = /^(\d{4}-\d{2}-\d{2})/;
class DateChunker extends Transform {
constructor(outDir, options = {}) {
super({ ...options, objectMode: true });
this.outDir = outDir;
this.streams = new Map();
this.bytesByDate = new Map();
this.currentDate = 'unknown';
}
writeStreamFor(date) {
let dest = this.streams.get(date);
if (!dest) {
const file = path.join(this.outDir, `${date}.log`);
dest = fs.createWriteStream(file, { flags: 'a' });
this.streams.set(date, dest);
this.bytesByDate.set(date, 0);
}
return dest;
}
async _transform(line, _enc, cb) {
try {
const match = DATE_RE.exec(line.text);
if (match) this.currentDate = match[1];
const date = this.currentDate;
const dest = this.writeStreamFor(date);
this.bytesByDate.set(date, this.bytesByDate.get(date) + line.bytes);
// Honor backpressure from the slower disk stream.
if (!dest.write(line.text + '\n')) {
await once(dest, 'drain');
}
cb();
} catch (err) {
cb(err);
}
}
async closeAll() {
const totals = {};
for (const [date, dest] of this.streams) {
dest.end();
await once(dest, 'close');
totals[date] = this.bytesByDate.get(date);
}
return totals;
}
}
module.exports = { DateChunker };
'use strict';
const fs = require('fs');
const { pipeline } = require('stream/promises');
const { LineSplitter } = require('./line-splitter');
const { DateChunker } = require('./date-chunker');
async function splitLogs(inputPath, outDir) {
await fs.promises.mkdir(outDir, { recursive: true });
const source = fs.createReadStream(inputPath, { highWaterMark: 256 * 1024 });
const chunker = new DateChunker(outDir);
await pipeline(source, new LineSplitter(), chunker);
const totals = await chunker.closeAll();
return totals;
}
if (require.main === module) {
const [input, outDir = './chunks'] = process.argv.slice(2);
if (!input) {
console.error('usage: node split-logs.js <input.log> [outDir]');
process.exit(1);
}
splitLogs(input, outDir)
.then((totals) => {
for (const [date, bytes] of Object.entries(totals)) {
console.log(`${date}\t${bytes} bytes`);
}
})
.catch((err) => {
console.error('split failed:', err.message);
process.exit(1);
});
}
module.exports = { splitLogs };
This snippet shows how to slice a huge, multi-gigabyte log file into separate files partitioned by day, without ever loading the whole thing into memory. The core idea is a streaming pipeline: bytes flow through a line splitter, then through a Transform that groups records by their date prefix and writes each group to a per-day file handle. Because everything is a stream, memory stays flat regardless of input size, and Node's built-in backpressure keeps the reader from outrunning the slower disk writes.
In line-splitter.js, LineSplitter extends Transform and buffers partial data across chunk boundaries. A single fixed-size read almost never lands on a newline, so the leftover tail after the last \n is stashed in this.tail and prepended to the next chunk. On _flush, any final line without a trailing newline is emitted so no record is silently dropped. It pushes objects rather than strings so downstream stages can keep metadata alongside the text.
date-chunker.js is where the routing happens. DateChunker extends Transform in objectMode and, for each line, extracts the leading YYYY-MM-DD via a regex. Lines that fail to parse are attributed to the current date so continuation lines and stack traces stay with their event rather than being discarded. writeStreamFor lazily opens a WriteStream per date and caches it in a Map, so a file is only created when a matching line appears. Crucially, when dest.write() returns false the chunker awaits the stream's drain event before continuing — this is the manual backpressure that prevents an unbounded number of buffered writes.
split-logs.js wires the stages together with stream.pipeline, which propagates errors and cleans up every stream on failure — a common pitfall when chaining .pipe() calls manually. After the pipeline resolves, it awaits chunker.closeAll() to flush and close every open file handle and reports the byte totals per day.
This pattern is the right tool when input is too large for memory, when output must be partitioned, and when ordering within each partition must be preserved. The main trade-offs are holding one open file descriptor per active date and relying on well-formed date prefixes, so the fallback-to-current-date rule matters for real-world messy logs.
Related snips
export type Settled<R> =
| { status: 'fulfilled'; value: R }
| { status: 'rejected'; reason: unknown };
export interface ConcurrencyOptions {
limit: number;Simple concurrency limiter for batch operations
import type { IncomingMessage, ServerResponse } from "http";
const MIN_BYTES = 1024;
const INCOMPRESSIBLE = /^(image|video|audio)\/|application\/(zip|gzip|x-brotli|pdf|octet-stream)/i;Response compression (only when it helps)
import pino, { Logger } from 'pino';
import { AsyncLocalStorage } from 'node:async_hooks';
export interface Store {
requestId: string;
logger: Logger;Request ID + structured logging (Express + pino)
import { NodeSDK } from '@opentelemetry/sdk-node';
import { OTLPTraceExporter } from '@opentelemetry/exporter-trace-otlp-http';
import { BatchSpanProcessor } from '@opentelemetry/sdk-trace-base';
import { getNodeAutoInstrumentations } from '@opentelemetry/auto-instrumentations-node';
import { Resource } from '@opentelemetry/resources';
import {OpenTelemetry tracing for Node HTTP
class ReindexCheckpoint < ApplicationRecord
enum status: { idle: 0, running: 1, done: 2, failed: 3 }
validates :index_name, presence: true, uniqueness: true
def self.for(index_name)Safer Background Reindex: slice batches + checkpoints
import { onCLS, onINP, onLCP, onFCP, onTTFB, type Metric } from 'web-vitals';
export interface VitalPayload {
id: string;
name: Metric['name'];
value: number;Web Vitals reporting to an API endpoint
Share this code
Here's the card — post it anywhere.
