9 — Stream processing

Chapter 8 covered moving messages between services, including log streams such as Kafka, where messages stay in an ordered log and each consumer keeps its own offset. This chapter is about computing over those streams continuously: counting, aggregating, joining and detecting patterns while the data is still arriving.

Typical jobs include fraud scores computed as payments happen, dashboards of orders per minute, sessions built from click events, and search indexes or caches kept up to date from a database’s change stream.

Bounded and unbounded data

Batch processing runs over a bounded dataset: yesterday’s logs, a table snapshot, a month of orders. The input is complete before the job starts, so the job can read everything, sort it, and produce one answer.

Stream processing runs over an unbounded dataset that never ends. The job never sees all the data, so every answer is computed from what has arrived so far, and some of it may arrive late.

The Dataflow paper from Google prefers the terms bounded and unbounded over batch and streaming, because the difference lies in the data, not the engine. A batch engine can process an unbounded stream by running repeatedly on slices, and a streaming engine can process a bounded file by reading it as a stream.

Streaming isn’t simply faster batch. The hard parts are deciding when an answer is complete enough to emit, keeping state safe across failures, and making sure a restart doesn’t count anything twice.

A job is a graph of operators

A streaming job is a directed graph. Sources read from logs, such as Kafka topics. Operators transform records. Sinks write results to another topic, a database, a cache or a search index.

[orders topic] -> parse -> filter -> keyBy(customer) -> window count -> [counts topic]
[payments topic] -> parse ------------------------------^ (join)

Each operator runs as several parallel instances. Records are routed between instances by key, so every record for one customer reaches the same instance. That routing is a shuffle, or repartition, and it costs a network hop and serialization for every record. When the input topic is already partitioned by the key the job needs, the shuffle can be skipped.

Stateless and stateful operators

A stateless operator looks at one record at a time. Parsing, filtering, reformatting and routing are stateless. They’re easy to scale, and after a crash they can restart from any point without losing anything except work in flight.

A stateful operator remembers something across records. Counting, summing, deduplicating, windowing and joining are all stateful.

State is what makes stream processing hard. It has to be partitioned by key so each instance holds only its share, it has to survive crashes, and it has to be moved when the job scales out. It also grows: a deduplication set with no expiry, or a join with no time bound, grows without limit.

Event time and processing time

Every event has two times.

• Event time is when the event happened, recorded by the device or service that produced it.
• Processing time is when the stream processor handles it, by the clock of the machine running the operator.

The two differ, and the gap isn’t constant. Network delays, retries, consumer lag, a phone that was offline for an hour, or a job restarted from an old offset all push processing time later. Flink’s documentation and the Dataflow paper both define the terms this way, and both build their models on event time.

A count of “orders per minute” can mean either.

Processing time is simpler and has the lowest latency, and it suits questions about the system itself, such as throughput. Event time is what most business questions mean, and it’s the only choice when results must be reproducible. Its cost is that the job has to decide when it has seen enough of a minute to report on it.

Event-time timestamps come from producers’ clocks, which can be wrong (chapter 13). Jobs usually reject or clamp timestamps far in the future, because one such event can push the job’s notion of time forward and close windows early.

Windows

An unbounded stream has no end, so an aggregate like “count” needs boundaries. A window groups records by time, and each window produces its own result.

Tumbling, size 10:      [0,10)     [10,20)    [20,30)
Hopping, size 10, every 5:
                        [0,10)
                             [5,15)
                                  [10,20)
Session, gap 5, one key:  x x  x        x x       x
                         [ session 1 ] [ s2 ]   [s3]

Names vary between engines. Flink calls hopping windows sliding windows. Kafka Streams uses hopping for the fixed-interval kind, and reserves sliding for windows defined by the time difference between records. Its documentation gives a concrete example of hopping windows: with a size of 5,000 ms and an advance of 3,000 ms, the windows are [0;5000), [3000;8000), and so on, aligned to the epoch.

Window assignment is a small calculation:

def tumbling(t, size):
    start = t - t % size
    return [(start, start + size)]

def hopping(t, size, every):
    first = t - t % every - size + every       # earliest window that can hold t
    return [(s, s + size) for s in range(first, t + 1, every) if s <= t < s + size]

print(tumbling(23, 10))        # [(20, 30)]
print(hopping(23, 10, 5))      # [(15, 25), (20, 30)]

Hopping windows multiply state and output: with a 10-minute window advancing every minute, each event updates 10 windows. Session windows are the most complex, because a late event can fall into the gap between two sessions and merge them into one.

Watermarks

With event-time windows, the job has to decide when a window is done. A watermark is the job’s estimate of how far event time has progressed. In Flink’s words, a watermark with time t declares that event time has reached t in that stream, so no more elements with a timestamp at or before t are expected. When the watermark passes the end of a window, the window fires and emits its result.

The most common way to generate one is bounded out-of-orderness: track the largest event time seen, and set the watermark a fixed amount behind it.

A larger bound waits longer, so fewer events arrive after their window has fired, but every result comes out later. An operator with several inputs takes the minimum of its inputs’ watermarks, so the slowest input sets the pace for everything downstream. An input partition that stops receiving data stops advancing its watermark and holds back the whole job, which is why engines provide a way to mark idle inputs.

The Dataflow paper names the two ways a watermark goes wrong. It’s sometimes too fast: for most distributed sources, a perfect watermark is impossible, so some data arrives behind it. It’s sometimes too slow: one slow input holds back the whole pipeline. The paper’s example of the first case is a phone that goes offline and uploads its events days later, which no watermark can predict.

So watermarks are a heuristic, and a design has to say what happens to data that arrives after them.

Late data

An event is late when it arrives after the watermark has passed its timestamp. The options are:

In Flink, the default allowed lateness is zero, so an element behind the watermark is dropped. With allowed lateness set, a window’s state is kept until the watermark passes its end plus the lateness, and a late element can make the window fire again. Elements later than that are dropped, or sent to a side output. Kafka Streams expresses the same idea as a grace period on windows.

Re-emitting a window means downstream consumers see more than one result for the same window. They have to treat a result as an update to a key, such as an upsert into a table keyed by window, not as a new fact to add to a total.

A simulation

The trade-offs are easier to see with numbers. This program generates 3,000 events, one every 0.1 seconds of event time over 300 seconds. Most arrive about a second late, with an exponential delay. 5% are stragglers, arriving 5 to 40 seconds late. The events are processed in arrival order and counted in 10-second tumbling windows, with a bounded out-of-orderness watermark and a configurable allowed lateness.

import random
from collections import defaultdict

random.seed(7)
SIZE = 10.0                      # tumbling window length, seconds of event time

# 3,000 events, one every 0.1 s of event time (0 to 300 s).
# Most arrive about 1 s late; 5% are stragglers, 5 to 40 s late.
events = []
for i in range(3000):
    t = i * 0.1
    delay = random.uniform(5, 40) if random.random() < 0.05 else random.expovariate(1.0)
    events.append((t + delay, t))          # (arrival time, event time)
events.sort()                               # the order the processor sees them

truth = defaultdict(int)
for _, t in events:
    truth[int(t // SIZE)] += 1

def run(bound, lateness):
    """bound: watermark lag behind the max event time seen.
    lateness: how long a window stays open after the watermark passes its end."""
    counts, fired, first, final = defaultdict(int), set(), {}, {}
    latency, dropped, updates, max_open = [], 0, 0, 0
    max_seen = float("-inf")
    for arrival, t in events:
        w = int(t // SIZE)
        end = (w + 1) * SIZE
        watermark = max_seen - bound
        if watermark >= end + lateness:            # window already purged
            dropped += 1
        else:
            counts[w] += 1
            if w in fired:                          # late but allowed: re-emit
                updates += 1
                final[w] = counts[w]
        max_seen = max(max_seen, t)
        watermark = max_seen - bound
        for v in list(counts):
            v_end = (v + 1) * SIZE
            if v not in fired and watermark >= v_end:
                fired.add(v)
                first[v] = final[v] = counts[v]
                latency.append(arrival - v_end)
        max_open = max(max_open, sum(1 for v in counts
                                     if watermark < (v + 1) * SIZE + lateness))
    # flush at end of input
    for v in counts:
        if v not in fired:
            first[v] = final[v] = counts[v]
    in_first = 100 * sum(first.values()) / len(events)
    exact = sum(final.get(w) == truth[w] for w in truth)
    return dropped, updates, in_first, exact, len(truth), sum(latency) / len(latency), max_open

print("bound  lateness  dropped  updates  in first result  exact at end  wait after end  open windows")
for bound in (0, 2, 10):
    for lateness in (0, 10, 40):
        d, u, f, fin, n, lat, mo = run(bound, lateness)
        print(f"{bound:>4}s {lateness:>7}s {d:>8} {u:>8} {f:>14.1f}% {fin:>9}/{n} {lat:>12.1f} s {mo:>10}")

Output:

“Wait after window end” is the processing time between a window’s end and its first result. The one window that is always exact is the last, which is flushed when the input ends.

Several things follow.

• The watermark bound trades latency for completeness of the first answer. Raising it from 0 to 10 seconds raised the share of events in the first result from 88.8% to 96.5%, and delayed every result by about 10 seconds.
• No reasonable bound catches stragglers. With 5% of events up to 40 seconds late, almost every window missed something, even with a 10-second bound. Only a 40-second allowed lateness made every window exact.
• Allowed lateness gets correctness without delaying the first answer. With a zero bound and 40 seconds of lateness, first results came out after 0.3 seconds, and every window was eventually exact. The cost was 336 updates, which consumers must handle as corrections.
• Lateness costs state. The number of windows held open rose from 1 to 5 or 6. Here that’s a handful of counters, but in a job with a million keys, it multiplies the state by the same factor.

The usual pattern is a short watermark bound for prompt results, an allowed lateness matched to how late real data arrives, and a side output for anything later still. Measuring the actual distribution of event delays is the first step, because the right settings depend entirely on it.

Joins over streams

Joins between streams come in a few kinds, and they differ mainly in how much state they hold.

Stream-stream joins match records from two streams, such as an ad impression and a click on that ad. Since either side can arrive first, each side’s records are buffered until a match arrives. The join needs a time bound, such as “a click within 30 minutes of the impression”. Kafka Streams makes this explicit: its documentation says stream-stream joins are always windowed, because otherwise the state store used for the join would grow indefinitely. A record on one side produces one output for every matching record on the other side within the window.

Stream-table joins enrich each event with the current value from a table, such as adding a customer’s country to each order. The table is itself built from a stream, typically a compacted topic or a database change stream (chapter 15), and held in local state. This relies on what Kafka Streams calls the stream-table duality: a table is the latest value per key from a stream of updates, and a stream is the history of changes to a table.

A subtle point is which version of the table a join sees. A plain stream-table join uses whatever value is in the table when the event is processed. During a replay of last month’s orders, that’s today’s customer data, not the data at the time of each order. A temporal join uses the table’s value as of each event’s timestamp, which needs the table’s history to be kept.

Co-partitioning is required for both kinds when the state is partitioned. Records with the same key from both inputs must land on the same instance, so both inputs need the same partitioning by the join key, with the same number of partitions. If not, one side is repartitioned first. A small, slowly changing table can instead be copied in full to every instance, which Kafka Streams calls a global table, at the cost of storage on each instance.

State stores and checkpoints

Stateful operators keep their state locally, next to the computation, rather than calling a remote database for each record. Local state is fast, but it disappears with the machine, so it has to be made durable another way.

Kafka Streams and Flink do this differently.

Kafka Streams keeps each state store locally, often in RocksDB, and writes every update to a changelog topic in Kafka. The changelog is partitioned like the store, and log compaction keeps only the latest value per key, so it doesn’t grow forever. If a task moves to another machine, the store is rebuilt by replaying its changelog before processing resumes. The documentation notes that this restore time is the main cost of moving a task.

Flink takes periodic checkpoints of the whole job. Its documentation describes a method inspired by the Chandy-Lamport algorithm for distributed snapshots:

1. The sources insert a barrier into the stream, recording their current position, such as a Kafka offset.
2. Barriers flow with the records and never overtake them, so each barrier divides the records before the snapshot from those after.
3. When an operator has received the barrier from all its inputs, it snapshots its state to durable storage and passes the barrier on.
4. When every sink has acknowledged the barrier, the checkpoint is complete.

A completed checkpoint holds, for each source, the position in the stream, and for each operator, a pointer to its saved state. On failure, Flink restores every operator’s state from the last completed checkpoint and rewinds the sources to the recorded positions. Records after the checkpoint are processed again, but against state that hasn’t seen them yet, so the state ends up as if each record was applied once.

Operators with several inputs must align barriers: once a barrier arrives on one input, records on that input wait until the barrier arrives on the others. This usually adds little latency, but under backpressure, alignment can stall. Flink’s unaligned checkpoints let barriers overtake buffered records, saving those records as part of the state instead. The documentation describes this as suited to jobs with a slow data path, where alignment times can reach hours.

A savepoint is a checkpoint triggered by hand. It’s used to stop a job, deploy new code, and start it again from the same state and offsets.

Exactly-once processing

“Exactly once” is the most misunderstood term in streaming. Records are, in fact, often processed more than once, because recovery replays everything since the last checkpoint. What exactly-once processing promises is that the effects on the job’s state and output are as if each record had been processed once.

That requires three things together:

Kafka transactions

Kafka’s design documentation explains how this works between topics. Since version 0.11.0.0, the producer can be idempotent: the broker assigns it an ID and discards duplicates using a sequence number on each message. The producer can also write to several partitions in a transaction, so that all of the writes become visible or none do.

The key step is that a consumer’s position is itself stored as a message in an internal topic. So a job can write its output and its consumer offsets in the same transaction. If the transaction aborts, the output is discarded and the offsets revert together. Downstream consumers must use the read_committed isolation level. The default, read_uncommitted, returns messages from aborted transactions too.

Kafka Streams builds on this. Its documentation states that, with exactly-once enabled, each record read from the source topics has its results reflected exactly once in the output topics and in the state stores.

Flink and two-phase commit

Flink extends its checkpoints to sinks with a two-phase commit. A sink writes output into a pending transaction during a checkpoint interval, prepares it when the barrier arrives, and commits it only when the checkpoint completes. If the job fails before then, the transaction is aborted, and the replayed records write their output again.

Flink’s Kafka connector shows the costs plainly. With the EXACTLY_ONCE delivery guarantee, the sink writes all messages in a Kafka transaction committed on a checkpoint, so readers using read_committed see no duplicates after a restart. But record visibility is delayed until a checkpoint completes, so output latency is at least the checkpoint interval. Each application needs a unique transactional ID prefix. And the Kafka transaction timeout must exceed the maximum checkpoint duration plus the maximum restart time, or Kafka may expire an uncommitted transaction and data can be lost.

Where it stops

These guarantees hold inside a boundary: from a replayable source, through the engine’s state, to a sink that takes part in the commit. Anything else a job does, such as calling an HTTP API, sending an email, or writing to a database outside the transaction, can happen again on replay. Kafka’s documentation suggests the general fix for external systems: store the consumer’s offset in the same place as its output, in one atomic write, so the two can’t disagree. Otherwise, the answer is idempotent writes, as described in chapter 15.

Reprocessing: Lambda and Kappa

Code changes, and bugs are found. When the logic that derives output from a stream changes, past output usually has to be recomputed. Jay Kreps, in Questioning the Lambda Architecture, called reprocessing one of the key challenges of stream processing.

The Lambda architecture runs two systems over the same input. A batch layer periodically recomputes complete, correct results from all historical data. A speed layer processes recent data with low latency. Queries merge the two. It handles reprocessing, since the batch layer can always recompute, but the same logic has to be written and maintained twice, in two frameworks, and the two must agree.

The alternative Kreps proposed, which he suggested calling the Kappa architecture, uses only the stream processor:

1. Keep the input log for as long as reprocessing may need, for example 30 days of retention.
2. To change the logic, start a second instance of the new job that reads from the beginning of the log and writes to a new output table.
3. When it has caught up, switch readers to the new table.
4. Stop the old job and delete its output.

Running both side by side also allows an instant rollback, by pointing readers back at the old table.

Kappa only gives the same answer on replay if the job uses event time, so results don’t depend on how fast the replay runs, and if its logic is deterministic. A replay also runs much faster than live traffic, so watermarks advance quickly and resource needs spike. For long histories, logs are often kept in tiered or object storage rather than on brokers.

Backpressure

When an operator can’t keep up, records pile up in front of it. In a streaming engine, buffers between operators are bounded. When a downstream operator falls behind, its input buffers fill, the upstream operator can’t send, and it slows down in turn. The pressure travels back to the sources, which read more slowly.

Flink’s documentation describes how it measures this: a task is considered back pressured when it has no available output buffers, meaning it produces data faster than downstream operators consume it.

With a log as the source, backpressure is safe in a way it isn’t for a service that answers users. The sources simply read more slowly, and unread data waits in the log. The symptom is consumer lag, the same metric as in chapter 8. Lag that grows steadily means the job can’t keep up on average. Lag that spikes and recovers means bursts.

The fixes depend on the cause:

Backpressure also slows checkpoints, because barriers wait behind buffered records. Long checkpoints mean longer replays after a failure, and with transactional sinks, longer delays before output becomes visible. So a job under steady backpressure tends to get worse in several ways at once.

Queueing theory from chapter 10 applies here too. A job running near its capacity has long, unstable lag, so streaming jobs need headroom, especially for replays and recovery, when they must process a backlog faster than live traffic arrives.

Operating a streaming job

A streaming job runs indefinitely, so it needs the same care as a service (chapter 20). The signals worth watching are:

Changes to a stateful job need planning. The saved state has to be readable by the new code, so state schemas need versioning, like event schemas. Changing the number of partitions or the keying of a stream changes which instance owns which state.

The takeaway

Stream processing computes over unbounded data, so every result is based on what has arrived so far. Event time gives reproducible answers, and watermarks decide when a window is done. They are always a guess, so late data needs a policy, and allowed lateness buys correctness at the cost of state and corrections. Stateful operators keep local state that must be checkpointed with the input offsets. Exactly-once means exactly-once effects inside a boundary of a replayable source, consistent snapshots and a transactional or idempotent sink, and outside it, idempotency does the work. Keep the input log long enough to reprocess, and watch lag.