--- title: 'Batch and Stream Processing Pipeline' source: 'https://academia.sh/en/courses/case-studies/batch-and-stream-processing-pipeline' course: 'Case Studies' language: en updated: '2026-08-23T07:01:24+00:00' license: 'CC BY-SA 4.0' --- # Batch and Stream Processing Pipeline The closing case where two processing models are used together in the same system: how the watermark cuts off late-arriving data, how the stream path's first published result diverges from the batch path's recomputation, the divergence's closing curve, and the overhead cost of running both paths at once. 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 ```js // 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. ```js // 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.