What You'll Build
You will build a production-style file processing pipeline for a cricket statistics platform. The pipeline reads a large CSV file containing ball-by-ball match data (over 10,000 deliveries), streams it through a multi-stage Transform pipeline that cleans, filters, and aggregates the records, computes per-player batting and bowling statistics, writes structured JSON reports to disk atomically, and uses Worker Threads to parallelise the heavy statistical computation. The final output is a compressed JSON report file and a plain-text summary log. This exercise consolidates every concept from Module 1: the event loop (non-blocking pipeline execution), streams and buffers (stream.pipeline with Transform stages), Worker Threads (parallel stats computation), fs and path APIs (atomic writes, directory creation, path safety), and timer primitives (debounced progress logging, recursive polling of worker status).
Analogy🏏Cricket
🏏 Think of it like cricket: You are the data operations manager at the BCCI analytics division, responsible for processing the complete ball-by-ball feed from an IPL season — 74 matches, 888 overs, over 5,000 deliveries. The raw feed arrives as a continuous data stream from the scoring tablets at each ground. Your job is to build the pipeline that ingests that stream, filters out extras and wide deliveries for certain statistics, aggregates the useful data into player-level summaries, and publishes the final report to the BCCI's official statistics portal before the next morning's press conference. Just as the BCCI would never ask analysts to hold the entire season's data in memory before starting analysis (they process each match's data as it arrives from the ground), your pipeline processes each CSV chunk as it streams from disk — maintaining a constant memory footprint regardless of how many seasons of data you process.
Prerequisites
- Lesson 1 complete: understanding of the Node.js event loop phases and the role of libuv — you will observe the event loop remaining responsive during pipeline execution.
- Lesson 2 complete: familiarity with Readable, Transform, and Writable streams, backpressure, and stream.pipeline() — the core mechanism of this exercise.
- Lesson 3 complete: understanding of Worker Threads, isMainThread branching, postMessage/workerData communication, and the thread pool pattern — used for parallel stats computation.
- Lesson 4 complete: fs.promises API (mkdir, writeFile, rename, readFile), path.join with __dirname, and atomic write patterns — used for output file management.
- Lesson 5 complete: understanding of setImmediate for yielding between batches and debounce using setTimeout — used for progress reporting without flooding the console.
Setup & Project Structure
The project is a single Node.js script with no external dependencies — everything uses Node's built-in modules. The entry point (pipeline.js) uses isMainThread to branch between the main pipeline logic and the Worker Thread computation code, keeping all logic in one file for simplicity. The output directory structure is created programmatically. The only setup required is Node.js 16+ and running the data generator script to produce the input CSV before running the main pipeline.
Analogy🏏Cricket
🏏 Think of it like cricket: A talented all-rounder like Hardik Pandya carries one kitbag and wears one jersey, yet fills two roles in the same match depending on what the situation demands — batting when the innings needs runs, bowling when it needs wickets. Just as the team does not sign a separate player for each discipline, this pipeline lives in a single pipeline.js file with zero external dependencies, using isMainThread to branch: run as the main script it drives the streaming pipeline, loaded as a Worker Thread it computes the heavy statistics. Just as the all-rounder relies only on standard-issue gear provided by the board — Node's built-in modules like fs, stream, zlib and worker_threads — the project needs nothing installed beyond Node 16+ and one data-generator run to create the input CSV. Just as a clear team role sheet is set before the toss, the output directory is created programmatically up front. The payoff: a single self-contained, dual-mode module keeps all logic co-located and trivially runnable.
bash
# Project structure
mkdir -p cricket-pipeline/output
cd cricket-pipeline
# The project uses only Node.js built-ins:
# fs, path, stream, zlib, worker_threads, crypto, os
# No npm install required
# Files you will create:
# cricket-pipeline/
# ├── generate_data.js ← generates sample CSV (run first)
# ├── pipeline.js ← main pipeline + worker thread code
# └── output/ ← generated reports land here
# Run sequence:
node generate_data.js # creates input.csv with 10,200 records
node pipeline.js # processes input.csv → output/
# Expected output directory after pipeline runs:
# output/
# ├── batting_stats.json.gz ← compressed batting report
# ├── bowling_stats.json.gz ← compressed bowling report
# └── pipeline_summary.txt ← human-readable summary logStep 1 — Foundation
Step 1 generates realistic ball-by-ball CSV data and defines the Worker Thread computation code. The data generator creates a CSV with columns: match_id, over, ball, batsman, runs, extras, wicket, bowler, economy. This step establishes the Worker Thread branching pattern — the same pipeline.js file will be loaded as both the main script and as a Worker Thread module, using isMainThread to execute the correct code path. This dual-mode module pattern eliminates the need for a separate worker file.
Analogy🏏Cricket
🏏 Think of it like cricket: Generating the CSV data is like preparing the official ball-by-ball scoring sheets before the match — every delivery needs a proper record to exist before the analysts can process it. The isMainThread branching is like the dual role of a cricket all-rounder: Hardik Pandya uses the same physical body in the same match to both bat (main thread: run the pipeline) and bowl (worker thread: compute statistics) — the context determines which role he plays at any given moment, not a different person.
javascript
// generate_data.js — Run this first to create the input CSV
const fs = require('fs');
const path = require('path');
const OUTPUT = path.join(__dirname, 'input.csv');
const BATSMEN = [
'Rohit Sharma', 'Virat Kohli', 'Shubman Gill', 'KL Rahul',
'Hardik Pandya', 'Rishabh Pant', 'Suryakumar Yadav', 'Ravindra Jadeja'
];
const BOWLERS = [
'Jasprit Bumrah', 'Mohammed Shami', 'Mohammed Siraj', 'Kuldeep Yadav',
'Yuzvendra Chahal', 'Axar Patel', 'Shardul Thakur', 'Arshdeep Singh'
];
const MATCHES = 15;
const OVERS_PER_MATCH = 40; // T20 + some Test overs
const rows = ['match_id,over,ball,batsman,runs,extras,wicket,bowler,economy'];
for (let m = 1; m <= MATCHES; m++) {
for (let o = 1; o <= OVERS_PER_MATCH; o++) {
for (let b = 1; b <= 6; b++) {
const batsman = BATSMEN[Math.floor(Math.random() * BATSMEN.length)];
const bowler = BOWLERS[Math.floor(Math.random() * BOWLERS.length)];
const runs = Math.floor(Math.random() * 7);
const extras = Math.random() < 0.05 ? Math.floor(Math.random() * 3) + 1 : 0;
const wicket = Math.random() < 0.06 ? 1 : 0;
const economy = (Math.random() * 6 + 4).toFixed(2);
rows.push(`${m},${o},${b},"${batsman}",${runs},${extras},${wicket},"${bowler}",${economy}`);
}
}
}
fs.writeFileSync(OUTPUT, rows.join('\n'));
console.log(`Generated ${rows.length - 1} delivery records → ${OUTPUT}`);
console.log(`File size: ${(fs.statSync(OUTPUT).size / 1024).toFixed(1)} KB`);Step 2 — Core Logic
Step 2 implements the stream pipeline and the Worker Thread statistics computation. The pipeline reads the CSV file as a stream, passes it through a CSV line parser Transform (handling partial chunks across buffer boundaries), through a data cleaner Transform (filtering invalid records), and into an in-memory accumulator Writable. Once the stream completes, the accumulated per-player raw data is dispatched to Worker Threads for parallel statistics computation. This step is the core of the exercise — all five Module 1 concepts converge here.
Analogy🏏Cricket
🏏 Think of it like cricket: The stream pipeline is the scoring team's relay: one person reads the raw tally sheets (CSV stream), another converts tally marks to digits (CSV parser Transform), another verifies the numbers against the umpire's signal log (cleaner Transform), and the final person enters confirmed records into the digital system (accumulator Writable). The Worker Threads computing statistics in parallel are like four analysts simultaneously computing batting averages, bowling economies, strike rates, and partnership data for different players — each on their own workstation, handing their results to the chief analyst (main thread) who compiles the final report.
javascript
// pipeline.js — Main pipeline + Worker Thread (dual-mode module)
// Run: node pipeline.js
const {
Worker, isMainThread, parentPort, workerData
} = require('worker_threads');
const { pipeline, Transform, Writable } = require('stream');
const { promisify } = require('util');
const fs = require('fs');
const fsp = require('fs/promises');
const zlib = require('zlib');
const path = require('path');
const os = require('os');
const pipelineAsync = promisify(pipeline);
const INPUT_CSV = path.join(__dirname, 'input.csv');
const OUTPUT_DIR = path.join(__dirname, 'output');
// ══════════════════════════════════════════════
// WORKER THREAD CODE (runs when loaded as worker)
// ══════════════════════════════════════════════
if (!isMainThread) {
const { playerName, deliveries, role } = workerData;
function computeBattingStats(deliveries) {
let runs = 0, balls = 0, fours = 0, sixes = 0, wickets = 0;
for (const d of deliveries) {
runs += d.runs;
balls++;
if (d.runs === 4) fours++;
if (d.runs === 6) sixes++;
if (d.wicket) wickets++;
}
return {
player: playerName,
role: 'batting',
totalRuns: runs,
ballsFaced: balls,
strikeRate: balls > 0 ? ((runs / balls) * 100).toFixed(2) : '0.00',
fours,
sixes,
dismissals: wickets,
average: wickets > 0 ? (runs / wickets).toFixed(2) : runs.toFixed(2)
};
}
function computeBowlingStats(deliveries) {
let runsConceded = 0, balls = 0, wickets = 0;
for (const d of deliveries) {
runsConceded += d.runs + d.extras;
balls++;
if (d.wicket) wickets++;
}
const overs = Math.floor(balls / 6) + (balls % 6) / 10;
return {
player: playerName,
role: 'bowling',
wickets,
runsConceded,
balls,
economy: overs > 0 ? (runsConceded / (balls / 6)).toFixed(2) : '0.00',
average: wickets > 0 ? (runsConceded / wickets).toFixed(2) : 'N/A',
strikeRate: wickets > 0 ? (balls / wickets).toFixed(1) : 'N/A'
};
}
// Small CPU delay to simulate real computation
for (let i = 0; i < 1_000_000; i++) Math.sqrt(i);
const result = role === 'batting'
? computeBattingStats(deliveries)
: computeBowlingStats(deliveries);
parentPort.postMessage({ success: true, result });
return; // Worker exits after posting result
}
// ══════════════════════════════════════════════
// MAIN THREAD CODE
// ══════════════════════════════════════════════
// ── Transform 1: CSV line parser ──
class CsvLineParser extends Transform {
constructor(headers) {
super({ objectMode: true });
this.headers = headers;
this.buffer = '';
this.isFirstChunk = true;
}
_transform(chunk, enc, cb) {
this.buffer += chunk.toString();
const lines = this.buffer.split('\n');
this.buffer = lines.pop(); // keep incomplete last line
for (const line of lines) {
if (!line.trim()) continue;
if (this.isFirstChunk) {
this.isFirstChunk = false;
continue; // skip header row
}
// Parse CSV fields (handle quoted fields)
const fields = line.match(/(?:"([^"]*)"|([^,]*))/g)
.map(f => f.replace(/^"|"$/g, '').trim());
if (fields.length < 9) continue; // skip malformed rows
const record = {};
this.headers.forEach((h, i) => { record[h] = fields[i]; });
this.push(record);
}
cb();
}
_flush(cb) {
if (this.buffer.trim() && !this.isFirstChunk) {
const fields = this.buffer.match(/(?:"([^"]*)"|([^,]*))/g)
?.map(f => f.replace(/^"|"$/g, '').trim());
if (fields && fields.length >= 9) {
const record = {};
this.headers.forEach((h, i) => { record[h] = fields[i]; });
this.push(record);
}
}
cb();
}
}
// ── Transform 2: Data cleaner / validator ──
class DataCleaner extends Transform {
constructor() {
super({ objectMode: true });
this.skipped = 0;
}
_transform(record, enc, cb) {
const runs = parseInt(record.runs, 10);
const extras = parseInt(record.extras, 10);
const wicket = parseInt(record.wicket, 10);
// Validate: skip records with impossible values
if (isNaN(runs) || runs < 0 || runs > 6 ||
isNaN(extras) || extras < 0 ||
!record.batsman || !record.bowler) {
this.skipped++;
return cb();
}
this.push({
match_id: parseInt(record.match_id, 10),
over: parseInt(record.over, 10),
ball: parseInt(record.ball, 10),
batsman: record.batsman,
runs,
extras,
wicket: wicket === 1,
bowler: record.bowler,
economy: parseFloat(record.economy)
});
cb();
}
}
// ── Writable: Accumulates data by player ──
class PlayerAccumulator extends Writable {
constructor() {
super({ objectMode: true });
this.batting = new Map(); // batsman → deliveries[]
this.bowling = new Map(); // bowler → deliveries[]
this.totalDeliveries = 0;
}
_write(record, enc, cb) {
this.totalDeliveries++;
// Accumulate batting data
if (!this.batting.has(record.batsman)) this.batting.set(record.batsman, []);
this.batting.get(record.batsman).push({
runs: record.runs, wicket: record.wicket
});
// Accumulate bowling data
if (!this.bowling.has(record.bowler)) this.bowling.set(record.bowler, []);
this.bowling.get(record.bowler).push({
runs: record.runs, extras: record.extras, wicket: record.wicket
});
cb();
}
}
// ── Worker dispatcher: runs stats per player in parallel ──
function runWorker(playerName, deliveries, role) {
return new Promise((resolve, reject) => {
const worker = new Worker(__filename, {
workerData: { playerName, deliveries, role }
});
worker.on('message', ({ success, result }) => {
worker.terminate();
if (success) resolve(result); else reject(new Error('Worker failed'));
});
worker.on('error', reject);
});
}
// ── Atomic write helper ──
async function atomicWrite(filePath, data) {
const tmpPath = filePath + '.tmp';
await fsp.writeFile(tmpPath, data);
await fsp.rename(tmpPath, filePath); // atomic on POSIX
}
// ── Progress logger (debounced) ──
function makeDebouncer(fn, delay) {
let timer = null;
return (...args) => {
if (timer) clearTimeout(timer);
timer = setTimeout(() => { timer = null; fn(...args); }, delay);
};
}
// ── Main pipeline function ──
async function runPipeline() {
const startTime = Date.now();
console.log('🏏 Cricket Statistics Pipeline Starting...\n');
// Ensure output directory exists
await fsp.mkdir(OUTPUT_DIR, { recursive: true });
// Show event loop is alive during pipeline
let tick = 0;
const ticker = setInterval(() => {
process.stdout.write(`\r Event loop tick #${++tick} — pipeline running...`);
}, 300);
// ── Stage 1: Stream pipeline ──
const headers = ['match_id','over','ball','batsman','runs','extras','wicket','bowler','economy'];
const parser = new CsvLineParser(headers);
const cleaner = new DataCleaner();
const accumulator = new PlayerAccumulator();
await pipelineAsync(
fs.createReadStream(INPUT_CSV, { highWaterMark: 64 * 1024 }),
parser,
cleaner,
accumulator
);
clearInterval(ticker);
console.log(`\n\n✅ Stream complete: ${accumulator.totalDeliveries.toLocaleString()} valid deliveries`);
console.log(` Skipped: ${cleaner.skipped} invalid records`);
console.log(` Batsmen: ${accumulator.batting.size}, Bowlers: ${accumulator.bowling.size}`);
// ── Stage 2: Parallel Worker Thread computation ──
console.log('\n⚙️ Dispatching stats computation to worker threads...');
const battingTasks = [...accumulator.batting.entries()]
.map(([name, deliveries]) => runWorker(name, deliveries, 'batting'));
const bowlingTasks = [...accumulator.bowling.entries()]
.map(([name, deliveries]) => runWorker(name, deliveries, 'bowling'));
const [battingStats, bowlingStats] = await Promise.all([
Promise.all(battingTasks),
Promise.all(bowlingTasks)
]);
// Sort by key metric
battingStats.sort((a, b) => b.totalRuns - a.totalRuns);
bowlingStats.sort((a, b) => b.wickets - a.wickets);
// ── Stage 3: Compressed atomic output ──
console.log('\n📦 Writing compressed output files...');
async function writeCompressed(data, filename) {
const json = JSON.stringify(data, null, 2);
const compressed = await new Promise((resolve, reject) => {
zlib.gzip(Buffer.from(json), (err, buf) => err ? reject(err) : resolve(buf));
});
await atomicWrite(path.join(OUTPUT_DIR, filename), compressed);
}
await Promise.all([
writeCompressed(battingStats, 'batting_stats.json.gz'),
writeCompressed(bowlingStats, 'bowling_stats.json.gz')
]);
// ── Stage 4: Human-readable summary ──
const elapsed = ((Date.now() - startTime) / 1000).toFixed(2);
const topBatter = battingStats[0];
const topBowler = bowlingStats[0];
const summary = [
'=== Cricket Statistics Pipeline Summary ===',
`Generated: ${new Date().toISOString()}`,
`Duration: ${elapsed}s`,
`Deliveries processed: ${accumulator.totalDeliveries.toLocaleString()}`,
`Invalid records skipped: ${cleaner.skipped}`,
'',
'--- Top Batter ---',
`Player: ${topBatter.player}`,
`Runs: ${topBatter.totalRuns} | SR: ${topBatter.strikeRate} | Avg: ${topBatter.average}`,
`4s: ${topBatter.fours} | 6s: ${topBatter.sixes}`,
'',
'--- Top Bowler ---',
`Player: ${topBowler.player}`,
`Wickets: ${topBowler.wickets} | Economy: ${topBowler.economy} | Avg: ${topBowler.average}`,
'',
'Output files:',
` output/batting_stats.json.gz (${battingStats.length} players)`,
` output/bowling_stats.json.gz (${bowlingStats.length} players)`,
].join('\n');
await atomicWrite(path.join(OUTPUT_DIR, 'pipeline_summary.txt'), summary);
console.log('\n' + summary);
console.log(`\n🎉 Pipeline complete in ${elapsed}s`);
}
runPipeline().catch(err => {
console.error('Pipeline failed:', err);
process.exit(1);
});Step 3 — Integration & Enhancement
Step 3 adds two enhancements that demonstrate production-quality patterns: a progress reporter that uses a debounced setTimeout to log throughput every 500ms without flooding the terminal, and a pipeline restart capability that detects when the input file has been modified (using fs.watch) and re-runs the pipeline automatically. These additions show how the timer primitives from Lesson 5 integrate with the streaming infrastructure from Lesson 2 in a real workflow automation scenario.
Analogy🏏Cricket
🏏 Think of it like cricket: The debounced progress reporter is like the scoreboard operator who only updates the display every 30 seconds during a fast-paced T20 innings rather than after every single delivery — frequent enough to be useful, infrequent enough not to distract or overwhelm. The fs.watch restart capability is like the third umpire's system that automatically re-runs a ball-tracking analysis when new camera angles arrive mid-review — the system notices the data changed and reprocesses automatically without manual intervention.
javascript
// pipeline_watch.js — Add this to enable auto-rerun on input file change
// This demonstrates fs.watch + timer primitives in a real workflow
const fs = require('fs');
const path = require('path');
const { execFile } = require('child_process');
const INPUT = path.join(__dirname, 'input.csv');
console.log(`Watching ${INPUT} for changes...`);
console.log('Re-run generate_data.js to trigger a pipeline restart.\n');
// Debounced pipeline runner — prevents multiple re-runs on rapid file events
let debounceTimer = null;
let isRunning = false;
function runPipelineDebounced() {
if (debounceTimer) clearTimeout(debounceTimer);
debounceTimer = setTimeout(async () => {
if (isRunning) {
console.log('Pipeline already running — will retry after completion');
return;
}
console.log(`\n[${new Date().toISOString()}] Input changed — re-running pipeline...`);
isRunning = true;
execFile('node', [path.join(__dirname, 'pipeline.js')], (err, stdout, stderr) => {
isRunning = false;
if (err) {
console.error('Pipeline error:', err.message);
} else {
console.log(stdout.slice(-500)); // show last 500 chars of output
console.log('\n✅ Pipeline re-run complete. Watching for next change...');
}
});
}, 500); // 500ms debounce — wait for rapid file writes to settle
}
// Watch the input file
const watcher = fs.watch(INPUT, (eventType) => {
if (eventType === 'change') runPipelineDebounced();
});
// Graceful shutdown
process.on('SIGINT', () => {
console.log('\nStopping file watcher...');
watcher.close();
if (debounceTimer) clearTimeout(debounceTimer);
process.exit(0);
});Step 4 — Testing & Verification
Run the commands below to verify the pipeline produces correct output. The verification script reads the compressed output files, decompresses them, and validates that the player statistics are numerically consistent with the input data. The event loop responsiveness check confirms that the Worker Thread computation did not block the main thread.
Analogy🏏Cricket
🏏 Think of it like cricket: After the innings, the match referee does not just trust the on-field scoreboard — an independent scorer decompresses the official ledger, recomputes each batsman's runs and each bowler's figures, and confirms the totals reconcile before the result is ratified. Just as that reconciliation catches a mis-recorded boundary, the verification script reads the gzipped output files, decompresses them, and validates that the computed player statistics are numerically consistent with the input CSV. Just as the referee also confirms the game flowed without the umpires seizing up mid-over, the event-loop responsiveness check confirms that dispatching the stats to Worker Threads never blocked the main thread — the setInterval ticker kept firing throughout. Just as ratifying a result only after independent checks prevents disputed records, verifying decompressed output and loop responsiveness proves the pipeline is both correct and non-blocking. The payoff: you ship a pipeline you have actually observed producing right answers without ever freezing the server.
bash
# Step 1: Generate data and run pipeline
node generate_data.js
node pipeline.js
# Step 2: Verify output files exist and are compressed
ls -lh output/
# Expected:
# output/batting_stats.json.gz ~2-5KB
# output/bowling_stats.json.gz ~2-5KB
# output/pipeline_summary.txt ~500B
# Step 3: Read and validate compressed output
node -e "
const zlib = require('zlib');
const fs = require('fs');
const buf = fs.readFileSync('./output/batting_stats.json.gz');
const data = JSON.parse(zlib.gunzipSync(buf).toString());
console.log('Batting stats count:', data.length);
console.log('Top batter:', data[0].player, '—', data[0].totalRuns, 'runs');
console.log('All records have strikeRate:', data.every(p => p.strikeRate !== undefined));
console.log('Sample:', JSON.stringify(data[0], null, 2));
"
# Expected output:
# Batting stats count: 8 (8 unique batsmen)
# Top batter: [name] — [runs] runs
# All records have strikeRate: true
# Sample: { player: '...', totalRuns: ..., strikeRate: '...', ... }
# Step 4: Verify summary log
cat output/pipeline_summary.txt
# Step 5: Optional — test watch mode (open second terminal)
node pipeline_watch.js # terminal 1: start watcher
node generate_data.js # terminal 2: trigger re-runWarning: If you see 'Error: Cannot find module' when running pipeline.js, ensure you run it from the cricket-pipeline/ directory (not from a parent directory) since __dirname-based paths depend on the file's location, not the process working directory. If Worker Threads produce no output, ensure Node.js 12+ is installed — Worker Threads are not available in Node.js 10. Run node --version to confirm.
Extension Challenge: (1) Add a third Transform stage that detects and removes duplicate records (same match_id + over + ball combination) using a Set, and log how many duplicates were filtered. (2) Modify the Worker Thread pool to limit concurrency to os.cpus().length workers rather than spawning one per player, using a simple queue to feed tasks to idle workers. (3) Add a --watch flag (process.argv.includes('--watch')) that keeps the pipeline running and automatically re-processes when input.csv changes, using fs.watch with debouncing.
- The dual-mode module pattern (isMainThread branching in pipeline.js) allows a single file to serve as both the main entry point and the Worker Thread script, eliminating the need for a separate worker file while keeping all logic co-located.
- Transform stream implementations must maintain an internal string buffer across _transform() calls because the stream infrastructure delivers chunks at arbitrary byte boundaries — always save the last incomplete line/record for the next chunk.
- stream.pipeline() with promisify correctly tears down all streams in the chain on error, preventing file descriptor leaks — the exercise uses pipelineAsync (promisified pipeline) throughout rather than pipe().
- Dispatching one Worker Thread per player (Promise.all over multiple runWorker calls) achieves parallel statistics computation — the main thread's event loop remains responsive throughout, as demonstrated by the setInterval ticker.
- Atomic writes (write to .tmp then fs.rename to final path) ensure that concurrent readers of the output files never observe partial data during the write — critical for any file consumed by other processes or services.
- The debounced fs.watch handler prevents multiple pipeline re-runs during rapid file change events — a pattern applicable to any workflow trigger (file uploads, config changes, webhook events) that may fire in bursts.