stream-processor

How it is built

Architecture overview

A stateful stream processor runs a program over data that never ends. Events arrive from logs and queues, flow through operators (filter, map, join, aggregate) and leave as results, continuously and with low latency. What makes it hard, and what separates it from a loop around a message consumer, is three things at once: state (a running count per user, a join buffer, a session) that must survive crashes, time (events arrive late and out of order, so "the last five minutes" has to be defined by when events happened, not when they were seen), and correctness after failure (a result must not be counted twice or lost when a machine dies mid-flight). The standard answer is a dataflow of parallel operator instances, each owning a slice of the keyed state, with periodic consistent checkpoints of all state plus input positions, and a commit protocol at the sinks. Apache Flink is the reference open design; Spark Structured Streaming, Kafka Streams and others solve the same problems with different trade-offs.

Parts:

• Event-time clock — tells every operator how far event time has progressed, so a window can close — watermarks (no page yet), built on d-ary heap and Consumer offsets
• Windowed aggregation — groups events by key and time window and keeps partial aggregates — Stream windowed aggregation, built on watermarks (no page yet), hash-aggregation (no page yet), LSM tree and Hierarchical timing wheel
• State store — per-key state, larger than memory, quick to snapshot — LSM tree
• Checkpointing — a consistent snapshot of all operator state and source positions — chandy-lamport-snapshots (no page yet), built on distributed-barrier (no page yet), copy-on-write-snapshots (no page yet) and Object storage
• Exactly-once results — state effects and sink output happen once even after replay — exactly-once-processing (no page yet), built on Idempotency keys, Event deduplication, Consumer offsets and Two-phase commit
• Source positions — where each input partition has been read up to — Consumer offsets
• Shuffle and key assignment — repartitions records so each key reaches one operator instance — data-shuffle (no page yet), built on Fast hash functions, Range partitioning and External merge sort; Consistent hashing for moving key ranges
• Joins — match two streams or a stream and a table — Hash join
• Back-pressure and wire format — slow consumers slow producers, records cross the network cheaply — credit-based-flow-control (no page yet), Zero-copy serialization

 sources (log partitions, offsets) ──► operator ──shuffle by key──► operator (keyed state, windows) ──► sink
        │                                 │                              │   ▲ watermarks flow with records
        │      ◄── credits (back-pressure)┴──────────────────────────────┘   │
        │                                                                    │
  coordinator ──inject barrier n──► sources ─► barrier flows with records ─► every operator snapshots state
        ▲                                                                    │
        └─────────────── checkpoint n complete (state in object storage) ◄───┘
  sink: pre-commit at barrier n ──► commit when checkpoint n is complete (two-phase)
  failure: restore state n, rewind sources to offsets n, replay

There are two products of one design: the running job (long-lived, thousands of operator instances, latency and recovery time matter) and the control plane (a coordinator that schedules tasks, triggers checkpoints and restarts failed work). Flink calls them TaskManagers and a JobManager (source: Flink documentation, architecture).

How the parts fit together

Event-time clock

Processing time (the machine clock) is easy and wrong for analytics: a mobile client that was offline for ten minutes sends old events late. Event time uses the timestamp inside the record. Flink states the consequence plainly: in event time, progress depends on the data, not on wall clocks, and the mechanism that signals progress is the watermark (source: Flink documentation, timely stream processing). A watermark W(t) says "no more events with timestamp ≤ t are expected". Each operator takes the minimum over its input channels' watermarks, which is why one idle partition can hold back the whole job, and why sources need idle-detection. Per-key pending timers (close this window at watermark ≥ t) are kept in a priority structure (d-ary heap). The source positions that bound the watermark come from the log's Consumer offsets. Out-of-order data forces a trade: wait longer (higher latency, fewer dropped events) or close sooner (lower latency, late events go to a side output or are discarded).

Windowed aggregation

Stream windowed aggregation groups events by key and event-time window (tumbling, sliding, session) and keeps one partial aggregate per window or slice, not the raw events; a sliding window of 1 hour every minute shares slices instead of re-aggregating 60 times. When the watermark passes the end of a window, the result is emitted and the state dropped, with an allowed-lateness period if late updates should still correct it. The aggregation itself is a hash aggregation (no page yet) keyed by (key, window), spilled to the state store when it outgrows memory, and the window-end timers sit in a Hierarchical timing wheel or a heap, because there can be millions of open windows.

State store

State per key can far exceed memory: a day of session data for every user. Flink offers two state backends: a heap-based one that keeps state as Java objects in memory, and an embedded RocksDB one that keeps it on local disk (source: Flink documentation, state backends). RocksDB is an LSM tree: writes go to memory and are flushed as immutable sorted files, which suits a workload of constant small updates and, importantly, makes incremental snapshots natural, since a snapshot only needs the new immutable files since the last one. The heap backend is faster per access and limited by memory and garbage collection. The choice is between speed and state size.

Checkpointing

Recovery needs a consistent picture of all operators' state together with the input positions that produced it. The method is the one from Chandy and Lamport, adapted to dataflows: the coordinator injects barriers into the sources; a barrier flows with the records, never overtaking them, and splits the stream into records that belong to this snapshot and records that belong to the next. When an operator has seen the barrier on all inputs it snapshots its state and forwards the barrier (source: Flink documentation, stateful stream processing; Carbone et al., Lightweight Asynchronous Snapshots for Distributed Dataflows). The paper's key result is that on acyclic dataflows only operator state needs to be persisted, not the records in transit, which keeps snapshots small, and the snapshot runs asynchronously so processing continues. Since Flink 1.11 a checkpoint can also be taken without aligning barriers, which helps when back-pressure makes alignment slow (same Flink page). Snapshots are written to durable shared storage (Object storage), copy-on-write or incremental files make them cheap, and a distributed barrier (no page yet) coordinates the "all instances have finished" step.

Exactly-once results

"Exactly-once" means each event affects the result once, not that it is processed once; after a failure the job replays events from the last checkpoint. For state, replay from a consistent snapshot already gives that: state and source offsets are restored together. For output, the sink must not publish replayed results twice. Flink can guarantee exactly-once state updates only if the source takes part in the snapshotting; its connector table lists Kafka and Kinesis as exactly once, others as at least once or at most once (source: Flink documentation, fault-tolerance guarantees). For sinks there are two techniques: make the write idempotent (Idempotency keys, Event deduplication: the same result written twice overwrites itself), or make it transactional with Two-phase commit: pre-commit the output when the barrier passes, and commit only after the whole checkpoint has completed. Flink added a TwoPhaseCommitSinkFunction in version 1.4.0 (December 2017) to support exactly that, with Kafka 0.11 and later as the first target (source: Flink blog, An Overview of End-to-End Exactly-Once Processing in Apache Flink). Kafka's own transactions, which tie produced records and consumed offsets into one atomic commit, are described in KIP-98.

Shuffle and key assignment

A keyed operator needs all events of a key in one place. The shuffle (no page yet for the general block) repartitions records by a hash of the key (Fast hash functions), sending each record over the network to the instance that owns that key. Range partitioning (Range partitioning) is used when order matters, and a sort-based exchange uses External merge sort when data does not fit in memory. To rescale, state must move: Flink splits keyed state into as many key groups as the job's maximum parallelism and assigns ranges of key groups to instances; the maximum parallelism therefore bounds how far a job can later scale (source: Flink documentation, tuning state). This is the stream-processor form of Consistent hashing: a stable intermediate level (key groups) so that changing the number of workers moves whole groups, not single keys.

Joins

A stream-stream join keeps recent records of both sides in state and matches new records against the other side's buffer. The Hash join idea carries over: build a hash table on one side, probe with the other. In streaming both sides are build and probe sides at once, and state is bounded only by a time condition (join events within 10 minutes of each other) or by a retention rule. A stream-table (enrichment) join probes a slowly changing table, which is kept in state as a changelog.

Back-pressure and wire format

When one operator is slower than its upstream, buffers fill and the job must slow the sources rather than drop data or run out of memory. Flink uses credit-based-flow-control (no page yet): a receiver announces how many network buffers it has free, and the sender sends only as much. The documentation notes that assigning credits and sending buffers takes around two round-trip messages between two nodes, and uses a roughly 1 ms buffer round trip in a healthy local network to size the buffer count for a target throughput (source: Flink documentation, network memory tuning). Records cross the network and the state store in a compact binary form; Zero-copy serialization formats that can be read in place avoid a decode step per hop, which matters because serialization is often the largest CPU cost in a shuffle-heavy job.

Data and control flow

One event, normal operation:

1. A source instance reads a record and its offset from a log partition and assigns an event timestamp.
2. The record is hashed by key and sent to the instance that owns the key group; credits limit what is in flight.
3. The operator updates the key's state (the window's partial aggregate) in the state store and registers a timer for the window end.
4. Watermarks advance from sources through operators; when an operator's watermark passes a timer, it emits the window result and clears the state.
5. The result goes to the sink: an idempotent write, or held in a pre-committed transaction.

One checkpoint:

1. The coordinator injects barrier n at every source; each source records its offsets.
2. Each operator, on receiving barrier n from all inputs, snapshots its state asynchronously to shared storage and forwards the barrier.
3. Sinks pre-commit their transaction on barrier n.
4. When every instance has acknowledged, checkpoint n is complete; the coordinator notifies the sinks, which commit.

One failure:

1. A task dies; the coordinator stops the affected tasks and restarts them.
2. State is restored from checkpoint n, sources rewind to the recorded offsets.
3. Events after the checkpoint are replayed; uncommitted sink transactions from the failed attempt are aborted; the output continues without duplicates.

Scaling

Limits, in the order they bite, and the lever for each:

1. Source parallelism. A job cannot read faster than the partitions of the input log allow. Lever: more partitions up front.
2. Skew. One hot key lands on one instance. Levers: pre-aggregate before the shuffle, split hot keys with a salt and merge afterwards.
3. State size. Heap state is bounded by memory. Levers: RocksDB-backed state, incremental checkpoints, time-to-live on state.
4. Checkpoint cost. Large state and slow shared storage make checkpoints long, and alignment under back-pressure makes them longer. Levers: incremental and unaligned checkpoints, longer interval, faster storage.
5. Rescaling. The number of key groups (maximum parallelism) is fixed when state is first created; Flink recommends setting it to a reasonable number up front and notes that it is changed only through a savepoint-based restart (source: Flink documentation, tuning state).
6. Network. Shuffle volume scales with record size times rate; narrower records and zero-copy formats help, and so does keeping chains of operators in one task so that records are not shuffled.

Failure modes

Real implementations

Sources

• Apache Flink documentation: Stateful stream processing (barriers, aligned and unaligned checkpoints), Timely stream processing (event time, watermarks), State backends (heap vs RocksDB), Fault-tolerance guarantees, Flink architecture, Tuning large state (key groups, maximum parallelism), Network memory tuning (credits).
• P. Carbone et al., Lightweight Asynchronous Snapshots for Distributed Dataflows, 2015. paper
• Flink blog, An Overview of End-to-End Exactly-Once Processing in Apache Flink (TwoPhaseCommitSinkFunction, Flink 1.4.0).
• K. M. Chandy and L. Lamport, Distributed Snapshots: Determining Global States of Distributed Systems, 1985. paper
• T. Akidau et al., The Dataflow Model: A Practical Approach to Balancing Correctness, Latency, and Cost in Massive-Scale, Unbounded, Out-of-Order Data Processing, PVLDB 8(12), 2015. paper
• Apache Kafka, KIP-98: Exactly Once Delivery and Transactional Messaging.