System Design: [Stream processing engine name]

Reference design for a distributed stream processing engine that runs long-lived, stateful jobs over unbounded event streams. It covers event time versus processing time, windows, watermarks, triggers and late data, keyed state with periodic checkpoints, exactly-once state with transactional sinks, backpressure and rescaling. The concepts follow the Dataflow model (Akidau et al., PVLDB vol. 8 no. 12, 2015) and asynchronous barrier snapshots (Carbone et al., 2015); the text and the worked design are original. It follows the structure of the System Design Document template. Replace every [bracketed] value and every row marked Example.

Document control

Approvals

Revision history

Contents

1. Overview
2. Requirements
3. Capacity estimation
4. API design
5. Data model
6. High-level design
7. Detailed design
8. Scalability and performance
9. Reliability, security and operations
10. Trade-offs and alternatives
11. Delivery plan and risks
12. Appendices

1. Overview

1.1 Problem statement

[Example: Product teams compute campaign metrics (impressions, clicks, spend per campaign and country per minute) with a nightly batch job and a fragile set of consumers that count in memory. Dashboards are 12 hours stale, counts change after every consumer restart, and mobile events that arrive late (devices offline, then reconnecting) are either dropped or counted in the wrong minute. Advertisers dispute numbers that do not match the batch reports.]

1.2 Proposed solution

[Example: A shared stream processing engine runs declarative jobs that read partitioned event logs, assign each event to event-time windows, keep per-key state in an embedded log-structured store on local SSD, and take a consistent distributed snapshot every 30 s by injecting barriers into the streams. Watermarks track event-time progress per source partition; triggers emit early speculative results every 10 s, an on-time result when the watermark passes the window end, and corrections for late events for up to 10 minutes. Sinks either commit transactionally on checkpoint completion or write idempotent upserts keyed by window. Jobs rescale by restoring a savepoint with a different parallelism; state moves in key groups.]

1.3 Decisions requested from reviewers

1.4 45-minute interview path

2. Requirements

2.1 Core functional requirements

2.2 Non-functional requirements

2.3 Below the line (out of scope)

• [Example: A SQL interface for ad-hoc queries; jobs are defined in code or the pipeline format in 4.2 for this release]
• [Example: Ingestion from devices; the engine reads from the existing message logs (system-design/design-message-queue)]
• [Example: Machine-learning feature serving; results land in stores that a feature service reads]
• [Example: Exactly-once delivery to external HTTP endpoints that offer no idempotency key]
• [Item deferred, with the release that will take it]

3. Capacity estimation

3.1 Assumptions

3.2 Ingest and parallelism

Consequence: the keyed shuffle moves roughly the same 2 GB/s (16 Gbit/s aggregate) across the network on top of ingest, so worker network interfaces (Example: 25 Gbit/s) are sized for ingest, shuffle and checkpoint upload together, and the job is split into at least 64 parallel tasks from day one. The maximum parallelism (number of key groups) is fixed at 1 024 so the job can grow 16 times without rehashing state.

3.3 State size

Open windows per key are one current window plus the lateness horizon:

Total keyed state is about 25 GB, or about 0.39 GB per task at parallelism 64. Consequence: deduplication, not the windows, dominates state, so it gets its own TTL and is the first thing to shrink (shorter horizon or idempotent producers upstream). State lives in an embedded log-structured store on local SSD rather than on the heap, which keeps garbage collection pauses away from the watermark path and enables incremental checkpoints.

3.4 Checkpoints and recovery

With Example 20% of state changing per interval, each incremental checkpoint uploads 5 GB, or about 167 MB/s sustained to object storage. Consequence: checkpoint upload is a steady background load that must be budgeted in network capacity, not an occasional burst.

After a worker loss the job rewinds to the last checkpoint. Worst-case backlog is one interval plus detection and restore time:

Spare capacity is $64 \times 40\,000 - 2 \times 10^{6} = 560\,000$ events/s, so catching up takes about $88 \times 10^{6} / 5.6 \times 10^{5} \approx 157$ s. Consequence: recovery to fresh output is under 3 minutes at peak, which meets NFR-04 but not a 60 s freshness promise during incidents; raising headroom to 1.5x or shortening the interval are the two levers.

3.5 Output

Each of the 5 million keys emits one on-time row per minute, about 83 000 rows/s, plus early firings every 10 s (up to 500 000 rows/s if every key is active). Consequence: sinks must accept batched upserts, and early firings are enabled only for keys a dashboard actually reads.

3.6 Summary

4. API design

4.1 Control plane API

4.2 Pipeline definition

A job declares what is computed, where in event time, when results are emitted and how refinements relate (the four questions of the Dataflow model, paraphrased):

job: campaign-metrics
version: 14
source:
  log: ad-events
  timestamp_field: event_time
  watermark:
    strategy: bounded_out_of_orderness
    max_delay: 30s
    idle_partition_timeout: 60s
key_by: [campaign_id, country]
window:
  type: tumbling
  size: 1m
  allowed_lateness: 10m
trigger:
  early: every 10s processing_time
  on_time: at_watermark
  late: every_element
accumulation: accumulating   # each firing carries the full value so far
aggregate: {impressions: count, clicks: count, spend_micros: sum}
dedup: {field: event_id, horizon: 10m}
late_side_output: ad-events-too-late
sink:
  type: upsert_store
  table: campaign_minute
  key: [campaign_id, country, window_start]
checkpoint: {interval: 30s, mode: aligned, unaligned_after: 60s}

4.3 Output record

Every emitted result states which firing it is, so consumers can tell a speculative value from a final one:

{"campaign_id": "c-7781", "country": "VN", "window_start": "2026-10-22T09:41:00Z", "window_end": "2026-10-22T09:42:00Z", "impressions": 18240, "clicks": 311, "spend_micros": 9120000, "pane": {"timing": "LATE", "index": 3, "is_last": false}, "watermark": "2026-10-22T09:42:31Z"}

4.4 Errors

5. Data model

5.1 Entities

A key maps to a key group by hash modulo max parallelism; key groups map to tasks in contiguous ranges. Rescaling moves whole key groups, never single keys, which keeps state handles coarse.

5.2 Storage choices

5.3 Access patterns

6. High-level design

7. Detailed design

7.1 Deep dive: time, watermarks and progress

Processing time is when a worker sees an event; event time is when it happened. Mobile and retried events arrive minutes late, and a replay after an outage delivers hours of events in minutes, so grouping by processing time puts events in the wrong window and makes results depend on when the job ran.

Choice: C, with per-partition watermarks combined by taking the minimum across inputs at every operator. Reason: it produces reproducible results under replay and a clear point at which a window is considered complete, while triggers (7.2) cover freshness before that point and corrections after.

Rules that keep watermarks honest:

• Generate watermarks per source partition, not per task, so a slow partition holds back the watermark instead of producing late data.
• Mark a partition idle after idle_partition_timeout so an empty partition does not stall all windows; un-idle it on the next event.
• Never let the watermark move backwards; on restore, start from the checkpointed watermark.
• Track watermark lag (processing time minus watermark) as the main freshness signal; it is the number a dashboard should show.

7.2 Deep dive: windows, triggers and late data

Windows decide where in event time results are computed; triggers decide when in processing time results are emitted; accumulation decides how successive results for the same window relate.

Options for late data: drop silently; keep windows open forever; keep windows for an allowed lateness then send later events to a side output. Choice: allowed lateness of 10 minutes with accumulating firings and a side output. Reason: state cost is bounded (3.3), dashboards stay correct through ordinary device delays, and nothing is dropped silently: the side output feeds the daily reconciliation that compares stream totals with the batch recomputation (NFR-03).

7.3 Deep dive: keyed state and checkpoints

Choice: C by default, switching a checkpoint to D when alignment exceeds 60 s. Reason: aligned barriers give a consistent cut of all operator state and source offsets with no global pause and small checkpoints; the fallback keeps checkpoints completing under sustained backpressure, which is exactly when a recovery point matters most.

State snapshots are asynchronous: the task flushes its memtable, records the list of immutable files and continues processing while files upload. Files already uploaded by earlier checkpoints are referenced, not copied. Retain the last 3 completed checkpoints and delete unreferenced files with a reference count, never by age alone.

7.4 Deep dive: exactly-once state and output

Checkpoints make state exactly-once: after a failure, state and source offsets rewind together, so each event affects state once. Output is a separate problem because records emitted after the last checkpoint are emitted again on replay.

Choice: B for databases and C for the output log. Reason: B gives exactly-once effect with no extra latency because replays overwrite the same rows with the same values; C is the only way to give downstream stream consumers exactly-once records, and the 30 s interval fits the 60 s freshness target (NFR-02). Record the pre-committed transaction IDs in the checkpoint so a restarted sink can commit or abort them; set the transaction timeout well above the longest expected recovery, or a commit after recovery fails and data is lost.

7.5 Backpressure, hot keys and rescaling

• Backpressure: use credit-based flow control between tasks: a receiver grants buffer credits and a sender sends only with credit, so a slow sink slows its upstream tasks instead of growing unbounded queues. Expose the share of time each task is backpressured and the source lag.
• Hot keys: a campaign that gets 10% of all traffic overloads one task. Pre-aggregate before the shuffle (each source task combines events per key for a short interval), then merge partial results after it. For non-mergeable logic, salt the key and add a second merge stage.
• Rescaling: take a savepoint, restart with the new parallelism, assign key-group ranges to tasks and restore only those ranges. Source partitions are reassigned; the source offsets in the savepoint keep reads exact.
• Upgrades: give every stateful operator a stable ID; refuse an upgrade whose operator IDs or state schemas cannot be matched (error 409 in 4.4).

7.6 How real systems do it

How real systems do it (paraphrased, not copied). Akidau and colleagues at Google described the Dataflow model in "The Dataflow Model: A Practical Approach to Balancing Correctness, Latency, and Cost in Massive-Scale, Unbounded, Out-of-Order Data Processing" (PVLDB vol. 8 no. 12, 2015). They argue that pipelines should stop assuming data will ever be complete and should instead let authors choose explicitly among correctness, latency and cost. The model separates what is computed (transformations), where in event time (windowing, including merging session windows), when results are materialised (watermarks and triggers) and how later results relate to earlier ones (discarding, accumulating, or accumulating with retractions). Separately, Carbone and colleagues described lightweight asynchronous barrier snapshots for dataflows (arXiv 1506.08603, 2015), the basis of the aligned checkpoint approach in 7.3. Read both papers for the reasoning; this template only reuses the concepts.

8. Scalability and performance

Latency budget for an on-time result (NFR-02, Example):

9. Reliability, security and operations

9.1 Failure-mode table

9.2 Security

9.3 Observability

10. Trade-offs and alternatives

11. Delivery plan and risks

11.1 Milestones

11.2 Rollout and rollback

[Example: Run the stream job in parallel with the batch job for one month; dashboards switch per team behind a flag. Rollback restores the previous job version from its last savepoint and points dashboards back to batch results.]

11.3 Risks

12. Appendices

12.1 Open questions

12.2 Glossary

12.3 Related templates

• Message Queue System Design (system-design/design-message-queue): the partitioned logs this engine reads and writes.
• CDC Pipeline System Design (system-design/design-cdc-pipeline): database changes as an input stream.
• Distributed Lock and Coordination Service System Design (system-design/design-distributed-lock-coordination): coordinator leader election and checkpoint pointer.
• Data Pipeline Specification (ai-data/data-pipeline-spec): contracts, quality checks and ownership for each job.
• Event Sourcing and CQRS (architecture/event-sourcing-cqrs): read models fed by stream jobs.

12.4 References

Standard and licence note

Based on the publicly described Dataflow model and asynchronous barrier snapshot technique cited in 12.4; no standard governs stream processing engines. This is a reference design to adapt, with illustrative numbers marked Example. Original text written for this collection; the cited papers and books were used as inspiration only and no text, figures or diagrams were copied. Check the current versions of the papers and of any engine you adopt. Template licence: MIT.