Batch and Stream Processing Pipeline

Each of the previous four cases did its work either at write time or at query time; none needed both together in the same system. This case takes on that problem. The system counts an event stream in one-minute windows, and two paths are built: stream processing publishes a result within seconds, and batch processing recomputes the same window hours later from the log. The two produce different results from the same input, because the stream path must close a window and cannot see anything after that moment.

The decision to close is made by the watermark: a bound expressed in event time that says how far the stream has advanced along its time axis, and a window closes once this bound passes it. Despite the similar name, it is distinct from the Application Layer and Service Interaction course’s version stamp and the Resilience and Reliability course’s deployment stamp: one marks a record’s version, the other a deployment unit’s identity, this one marks progress in time. A record that arrives after the watermark has passed is late-arriving data, and it does not enter the published count.

Constraints

Functional requirement: accept an event, count it in 60-second windows by event time, publish the result, and correct it with late-arriving data.

Non-functional requirement, as a number: the first result is published at most 60 seconds after the window closes; its divergence does not exceed 30 per mille; the divergence drops below 0.5 per mille within 12 hours; the two paths’ combined overhead does not exceed 3 times the raw input.

Scope reduction: cross-stream joins, session windows, exactly-once processing guarantees, and a query language are not designed.

Assumptions

Code	Assumption	Value	Rationale
TA1	daily events	900,000,000	total produced by the edges
TA2	event record	180 bytes	ID, event time, type, field
TA3	window size	60 s	the finest grain of reporting
TA4	delay distribution	0.94 uniform 0-5 s; 0.055 exponential 90 s; 0.005 exponential 4 hours	most edges are fast, very few are offline
TA5	peak factor	3	ratio of the peak hour to the daily average
TA6	batch run interval	6 hours	four runs a day
TA7	log retention period	30 days	history the recompute can reach

TA4 is the load-bearing assumption: the entire difference between the two paths comes from the delay distribution’s tail. Without a tail, the two paths give the same result and there is no need for a second path.

Scale

// pipeline-scale.mjs — the scale calculation derived from the TA table and the cost of running both paths at once
const TA = { dailyEvents: 900_000_000, eventBytes: 180, windowSec: 60, peakFactor: 3, batchIntervalHours: 6,
  retentionDays: 30 };
const DAY = 86_400;
const eventsPerSec = TA.dailyEvents / DAY;
for (const [name, d] of [
  ["avg events/s", eventsPerSec],
  ["peak events/s", eventsPerSec * TA.peakFactor],
  ["daily log GB", (TA.dailyEvents * TA.eventBytes) / 1e9],
  ["log TB (retention)", (TA.dailyEvents * TA.eventBytes * TA.retentionDays) / 1e12],
]) console.log(name.padEnd(22) + d.toFixed(2).padStart(14));

console.log(`\nbatch runs every ${TA.batchIntervalHours} hours; the scan window determines the workload:`);
console.log("scan hours".padEnd(13) + "daily runs".padStart(14) + "batch multiple".padStart(15) +
  "combined total".padStart(15));
for (const W of [6, 12, 24]) {
  const runs = 24 / TA.batchIntervalHours;
  const mult = (runs * W) / 24;
  console.log(String(W).padEnd(13) + String(runs).padStart(14) + mult.toFixed(2).padStart(15) +
    (1 + mult).toFixed(2).padStart(15));
}

avg events/s                10416.67
peak events/s               31250.00
daily log GB                  162.00
log TB (retention)              4.86

batch runs every 6 hours; the scan window determines the workload:
scan hours       daily runs batch multiple combined total
6                         4           1.00           2.00
12                        4           2.00           3.00
24                        4           4.00           5.00

These numbers are in the calculation class. The lower table binds the last constraint: because the batch run executes every six hours, the same event is read more times as the scan window grows — once for a six-hour window, twice for twelve hours, four times for twenty-four hours. The stream path’s single pass brings the combined total to 2.00 / 3.00 / 5.00 times, and the 3-times constraint eliminates the twenty-four-hour scan outright.

The Difference Between the Two Paths

The watermark delay determines both publish latency and divergence, and the two move in opposite directions. The model below builds no cluster; it draws delay from TA4’s mixture and compares what the stream path counts against the batch path’s exact count.

// two-paths.mjs — in-process model: the difference in the result the stream path and the
// batch path produce over the same event set. No network, no cluster; the delay distribution is
// TA4's three-part mixture, the generator is hand-written, and the seed is visible. IT IS A MODEL.
const WINDOW = 60, WINDOWS = 120, EVENTS = 2000, SEED = 20260803;
const MIX = [[0.94, "uniform", 5], [0.055, "exponential", 90], [0.005, "exponential", 14_400]];  // TA4

let state = SEED;
const random = () => { state = (state * 1103515245 + 12345) % 2147483648; return state / 2147483648; };
function delay() {
  let u = random(), i = 0;
  while (i < MIX.length - 1 && u > MIX[i][0]) { u -= MIX[i][0]; i += 1; }
  const [, kind, p] = MIX[i];
  return kind === "uniform" ? random() * p : -Math.log(1 - random()) * p;
}
const S = (x) => MIX.reduce((a, [w, kind, p]) =>          // independent of the run: tail probability
  a + w * (kind === "uniform" ? Math.max(0, 1 - x / p) : Math.exp(-x / p)), 0);
const expectedLate = (L) => {                             // via a uniform position within the window
  let t = 0;
  for (let i = 0; i < 6000; i += 1) t += S(WINDOW + L - (i + 0.5) * (WINDOW / 6000));
  return t / 6000;
};

const total = WINDOWS * EVENTS;                           // arrival instant measured from the window's start
const arrivals = Array.from({ length: total }, () => random() * WINDOW + delay());

console.log(`model: ${WINDOWS} windows x ${EVENTS} events = ${total}, window ${WINDOW} s, seed ${SEED}`);
console.log(`delay mixture: ${MIX.map(([w, t, p]) => `${w}/${t}/${p}s`).join(" ")}\n`);
console.log("watermark delay".padEnd(16) + "publish delay".padStart(16) +
  "late per mille".padStart(16) + "expected per mille".padStart(20));
for (const L of [0, 15, 60, 120, 300]) {
  const late = arrivals.filter((o) => o > WINDOW + L).length;
  console.log(`${L} s`.padEnd(16) + `${WINDOW + L} s`.padStart(16) +
    ((late / total) * 1000).toFixed(2).padStart(16) + (expectedLate(L) * 1000).toFixed(2).padStart(20));
}

console.log(`\nthe divergence's closing (against the result published with a 60 s watermark delay):`);
console.log("after window closes".padEnd(21) + "still missing".padStart(17) + "divergence per mille".padStart(22));
for (const [name, T] of [["1 hour later", 3600], ["6 hours later", 21_600], ["12 hours later", 43_200],
  ["24 hours later", 86_400]]) {
  const remaining = arrivals.filter((o) => o > WINDOW + T).length;
  console.log(name.padEnd(21) + String(remaining).padStart(17) + ((remaining / total) * 1000).toFixed(3).padStart(22));
}

model: 120 windows x 2000 events = 240000, window 60 s, seed 20260803
delay mixture: 0.94/uniform/5s 0.055/exponential/90s 0.005/exponential/14400s

watermark delay    publish delay  late per mille  expected per mille
0 s                         60 s           82.46               84.30
15 s                        75 s           37.42               38.96
60 s                       120 s           24.67               25.58
120 s                      180 s           13.32               15.53
300 s                      360 s            5.37                6.32

the divergence's closing (against the result published with a 60 s watermark delay):
after window closes      still missing  divergence per mille
1 hour later                       711                 2.962
6 hours later                      251                 1.046
12 hours later                      69                 0.287
24 hours later                       0                 0.000

The measured per-mille figure is a measurement tied to this run; the expected per-mille figure is a calculation independent of the run, derived from the mixture’s tail probability. On all five rows, the two sit at the same order of magnitude.

The upper table binds both constraints at once. At a watermark delay of 0, the divergence is 82.46 per mille; at 15 seconds, 37.42 — both break the 30-per-mille constraint; at 60 seconds, 24.67 per mille satisfies it. On the other hand, publish delay is window size plus the watermark, and the “at most 60 seconds later” constraint bounds the watermark to 60 seconds. The two constraints intersect at a single value: watermark delay of 60 seconds.

The first row also corrects a misconception. At a watermark of zero, most of what is missed is not late-arriving data — it is an event that lands in the window’s final seconds and arrives at normal speed; even some events with a delay under five seconds cross the boundary. At 15 seconds, this share disappears entirely, and the drop from 82.46 to 37.42 per mille comes from there. What is left of the divergence really does come from the tail.

The lower table gives the closing curve: 2.962 per mille an hour later, 1.046 six hours later, 0.287 twelve hours later. The “drops below 0.5 per mille within 12 hours” constraint is not satisfied by a six-hour scan, but it is by a twelve-hour one. The workload constraint had already eliminated twenty-four hours; the closing constraint eliminates six hours. What remains is a twelve-hour scan window, and the total work comes out to exactly 3.00 times the raw input.

Design

The single source of truth is the event log, and both paths are its projection (the Scaling the Data Layer course’s Read–Write Separation topic, Event-Sourced Design). The write model is the log, the read model is the window counts; separating the two is the same topic’s Command and Query Separation. The table the batch path produces is a materialized view (the Data Distribution topic), and its refresh is not incremental — it is a recomputation of the twelve-hour window.

Overwriting the published value must be idempotent (the Resilience and Reliability course’s Distributed Correctness topic); the uniqueness key is the pair of window and path, so the result does not change even if the same run executes twice. The intake side sits on a message broker and competing consumers (the Application Layer and Service Interaction course’s Queues and Workflows topic); the peak of 31,250 events/s is this layer’s input, and when a consumer falls behind, backpressure (the Fault Isolation topic) propagates all the way to the edge. What the reader sees is eventually consistent (the Introduction to System Design course’s Fundamental Properties topic), and the staleness window here is a measured number: 0.287 per mille.

Deliberately unused pattern: edge caching. The edge caching from the Traffic Layer course’s Entry Points topic is not placed in front of the published counts, because the same window’s value keeps being corrected for twelve hours; opening a second staleness window at the edge would make the design’s one measurable guarantee — the closing time — unmeasurable. The second is the compensating transaction (the Distributed Correctness topic): what the batch path does is not a rollback but an overwrite, because the log does not change.

Eliminated Alternatives

Stream only is the cheapest: one times the work, publish in 120 seconds. But the 24.67-per-mille divergence never closes; there is no second pass to correct it.

Batch only is flawless on accuracy and its workload is one times. A window’s result is published, in the worst case, six hours later: 360 times the publish constraint.

Widening the watermark brings the divergence down to 5.37 per mille at 300 seconds and satisfies the 30-per-mille constraint with a single path; publish delay rises to 360 seconds, six times over the first constraint. Which constraint changes and the alternative wins: if the publish constraint is relaxed to 6 minutes, a single stream path suffices, and the second path’s 2.00-times extra overhead drops.

Failure Behavior and What Is Given Up

When the stream path stops entirely, the system keeps responding: the batch path produces the correct result once every six hours, publish delay rises from 120 seconds to six hours, and the divergence drops to zero. This is an unusual form of graceful degradation (the Fault Isolation topic) — the dimension that degrades is not accuracy but freshness. The reverse happens when the batch path stops: publish stays at 120 seconds, but the 24.67-per-mille divergence freezes. Neither failure loses data, because both paths derive from the log, which is kept for thirty days.

What is given up: the same event is processed twice. Total work is 3.00 times the raw input, and the same count is kept in two separate code paths; that the two agree is guaranteed not by sharing code but by a measurement. The first number given to the reader is also never final.

Summary

• The watermark is the decision to close a window; because its delay moves publish latency and divergence in opposite directions, it is squeezed between two constraints.
• At a watermark of 0 / 15 / 60 / 120 / 300 seconds, the divergence is 82.46 / 37.42 / 24.67 / 13.32 / 5.37 per mille; the 30-per-mille constraint and the 60-second publish constraint intersect at a single value, 60 seconds.
• At a watermark of zero, most of what is missed is not late-arriving data but a normal event that lands in the window’s final seconds; at 15 seconds, this share disappears entirely.
• The scan window is squeezed from both sides: twenty-four hours violates the workload constraint at 5.00 times, six hours violates the closing constraint at 1.046 per mille; twelve hours satisfies both at the margin.

Course Wrap-Up

Fourteen cases carried out the same six steps, and each was defended with a number.

Case	The number that decides the design	Patterns chosen	Eliminated alternative	What is given up
URL Shortening	read/write 100; 3859.3 guesses at seven characters	edge caching, cache-aside, sharding	counter-based generation: 1.0 guesses per hit	300 s staleness window
News Feed	fan-out on write 1691.87 / fan-out on read 56,570.71 work/s (33.44x)	threshold-based hybrid fan-out, message queue, backpressure	pure fan-out on read: 33.44x work	203.63 box writes per post
Search Suggestion	266.67 bytes per query; 4.59 GB at depth 10	prefix tree, task queue, push-based distribution	a flat sorted list: 6593 units at p99	10-character limit, 600 s freshness
Content Distribution	request ratio 3.13, byte ratio 20.83; required hit rate 0.7440	content delivery network, edge caching, versioned name	short-lived single cache: 0.6188 at 2 GB	atomicity of publishing
Chat System	peak delivery 22,222.22/s; 1,250,000 connections, 25 nodes	uniqueness key, sequential convoy, offline mailbox	polling: 140.63x requests	chat-level parallelism
Notification System	single-channel delivery 0.9590, 3.01 points below the threshold	cascading order, circuit breaker, bulkhead pattern	parallel fan-out: 2.78x calls	30-60 s delivery delay
Rate Limiter	fixed window 1200 (2.000x), sliding window counter 600	sliding window counter, 0.2 s delayed synchronization	centralized counter: 40,000 touches/s	overshoot staying at 1.015x
Ticketing and Inventory	demand 56x the allotment; drains in 0.179 s	pessimistic locking, allotment row	optimistic locking: 2409 valid rejections	304,276 wait steps
Payment Flow	notification-only 59.74 per mille; with polling 9.76 per mille	idempotency, 15-min reconciliation cycle, audit trail	synchronous verification: 75.77 per mille	9.76 per mille and 30-min closure
Object Storage	1.0075 at a 2 MB part; erasure (10,4) factor 1.40	valet key, idempotency, sharding, supervisor	three copies: 3.00x, 438 lost per year	697,680 GB-month of extra storage
Video Streaming	watched/uploaded 225; 8.91 percent not playable	priority queue, claim check, edge caching	eight rungs: 2.58x, 1854 processors	bandwidth utilization 64.9 percent
Metrics Aggregation	240,000 samples/s; 31.50 TB against 393.57 TB	wide-column store, time partitioning, materialized view	five-minute rollup only: 100 percent error at the threshold	percentile questions older than 7 days
Location-Based	864 candidates at a 1 km cell, 51.4 percent hit rate	spatial index, composite index, hash sharding	8 km cell: 13,611 candidates	a scan roughly half wasted
Batch and Stream	24.67 per mille at a 60 s watermark; 0.287 per mille at 12 hours	watermark, event log and projection, idempotency	stream only: the divergence never closes	3.00x work, two code paths

The table’s rule is this curriculum’s rule: a design decision can be defended only together with a number and an alternative. A decision given without writing down the number’s class — assumption, calculation, measurement — cannot be argued, because the other side does not know what to ask for a reason: an assumption is defended with its rationale and sensitivity, a calculation with arithmetic that has been run, a measurement with its apparatus and seed.

Seven courses built this rule piece by piece. Introduction to System Design turned qualities into numbers and set up the back-of-the-envelope estimate; the Traffic Layer took on how a request enters the system, the Application Layer and Service Interaction how work is split across services, and Scaling the Data Layer how data is distributed. Resilience and Reliability took on how failure is contained, and Performance Anti-Patterns and Monitoring showed how these are made visible. Case Studies combined all of them, one problem at a time, and asked for the same thing every time: turn the constraint into a number, write down the assumption, run the calculation, choose the pattern by name, take on one alternative, say what is given up.

What comes after this is the reader’s own design problem. The constraints there will not resemble these fourteen cases’; the only thing that will resemble them is the method. The first question is always the same: which number decides this choice, what class is that number, and which alternative was eliminated.