Stream processing

A ride-hailing app raises prices in a zone when riders there outnumber drivers. To do that it needs one number, again and again: how many ride requests came from each zone in the last minute. A report that runs every night over the day's requests is correct and useless. By the time it finishes, the concert crowd that needed more drivers has gone home.

Stream processing computes over data as it arrives. The ride requests flow into a topic on a log-based broker (Message queues & pub/sub), and a long-running stream job reads them, keeps running counts, and publishes a fresh result every minute:

The difference is not only speed. A batch job reads bounded data, a set with an end, and can wait until it has seen all of it. A stream job reads unbounded data that never ends, so every question that batch answers by "wait for the end" needs a new answer:

The rest of this page answers the stream job's questions in order: where its counts live, how an endless stream is cut into minutes, which clock decides the minute, when a minute is finished, and how a crash is survived.

Some steps look at one event at a time and forget it. Others must remember earlier events. The difference decides how hard a step is to scale and to recover:

The memory a stateful step keeps is its state. A busy city has hundreds of zones, more requests than one machine can count, so the job runs as several instances, each reading some partitions of the topic. That only works if every request for one zone reaches the same instance, which holds that zone's count. The job routes events by key, here the zone name, exactly as a database shards rows by key (Sharding & partitioning):

Local state is fast because no network call is needed per event. It also brings the two problems the rest of the page deals with. One zone can be so busy that its instance falls behind (Hot partitions & skew). And local state dies with its machine, unless the job saves it somewhere (see the checkpoints section).

"How many requests from Midtown?" has no answer on an endless stream: the count only grows. The useful question is always about a stretch of time, such as "in the minute from 18:00 to 18:01". That stretch is a window. The job keeps one count per zone per open window and publishes it when the window ends.

Three shapes of window cover almost every need:

The shapes cost different amounts of memory. A tumbling window holds one count per zone. A 5-minute window that hops every minute keeps 5 overlapping windows open, so 5 counts per zone, and each request updates all 5. A session window has no fixed end at all: two bursts of taps 8 minutes apart merge into one session, because they are within the 10-minute gap.

A window has a start and an end on a clock. Which clock? The obvious answer, the clock on the job's machine, is wrong surprisingly often.

A rider in a tunnel taps "request ride" at 18:00:50. The phone has no signal, so the app holds the request and sends it when the train leaves the tunnel. It reaches the job at 18:01:20:

There are two clocks, and every stream job must choose one:

• Event time: when the event happened, stamped by the device that produced it and carried in the event.
• Processing time: when the job's machine handles the event.

Between them sits ingestion time, the moment the broker stored the event, which is fixed once written but still says nothing about the tunnel.

The gap between event time and processing time is not a small constant. It changes with signal, load and failures, and it is largest exactly when something goes wrong. Suppose the job is down for 20 minutes and then reads the backlog in 2 minutes:

So jobs whose results describe the real world count by event time. Processing time is fine when the question really is about the job, such as "how many events did we handle this minute?". One caveat: event time is only as good as the clock that stamped it, and phone clocks drift (Time, clocks & ordering).

Counting by event time creates a new problem. At 18:01:00 on the job's clock, is the 18:00 minute finished? A request stamped 18:00:50 might still be in a tunnel. The job can never be sure it has seen everything, yet it has to publish a count at some point.

A watermark is the job's declaration of progress. A watermark of 18:01:00 says: "no more events stamped before 18:01:00 are expected". When it passes the end of a window, the window's result is published and its count can be freed. A common way to compute it:

watermark = latest event time seen − allowed delay

With an allowed delay of 30 seconds, the 18:00 minute closes once the job has seen an event stamped 18:01:30. The delay is the one dial the designer turns, and it trades speed for completeness. Assuming most events arrive within a second of happening:

The tunnel request, 30 seconds late, makes it into the 18:00 count with a 30-second delay. Pricing that waits 5 minutes for every count is too slow to be useful, so some events will always be late. A watermark is a guess, not a guarantee, and the job needs a policy for events that arrive after it:

Results can also come early. A job can publish an estimate for the current minute every 10 seconds and the final count when the watermark passes, so the pricing service never waits a full minute to react. One trap: the watermark is the slowest input's progress. If one topic partition receives no events, the job's watermark stops moving and every window stays open; stream frameworks let you mark such an input as idle so the others can move on.

Instance A holds the open counts for Midtown and Harbor. Its machine dies. Those counts are gone, and the requests that built them were already read from the topic. Re-reading the whole topic is too slow; skipping ahead loses requests. The fix combines two things the job already has: a log it can rewind, and state it can copy.

At a regular interval the designer chooses, the job takes a checkpoint: a consistent snapshot of every instance's state together with each input's position in the log, saved to durable storage. "Consistent" means all instances snapshot at the same point in the input. Flink does this by sending markers through the stream along with the events. On a crash, the job goes back to the last checkpoint:

This gives exactly-once state: every request affects the counts exactly once, even though some were read twice. It is the same idea as storing a consumer's position with its result (Delivery semantics), applied to a whole job at once. Kafka Streams keeps the same promise differently: each instance's state is always written to a changelog topic, from which a replacement instance rebuilds it, and with exactly-once on, the changelog writes, the output and the input offsets commit together.

The guarantee covers the job's state, not what it has already sent out. If the job published Midtown's 18:00 count after checkpoint 7 and then crashed, the replay publishes it again. Two ways to make that harmless:

• Overwrite by key. Write the count as "Midtown, 18:00 → 42" into a table keyed by zone and minute. A repeat writes the same value again.
• Commit output with the checkpoint. Hold output in a transaction that commits only when the checkpoint completes. Nothing is published twice, but consumers see results one checkpoint later.

Demand alone doesn't set a price; it has to be compared with supply. Drivers' apps report their position every few seconds, so the job keeps a table of free drivers per zone, updated from that second stream. Each ride request is joined with the table as it arrives, a stream-table join:

Bugs need a different tool. Suppose the map step put part of Midtown into Harbor for a week. A batch system would rerun the report. A stream job can do the same thing, as long as the topic still holds the week's events: start a second copy of the fixed job from the start of the retained log, let it write to a new table, and point the pricing service at the new table once it has caught up. Then delete the old one.

Using one job for both live results and reprocessing is the idea behind the Kappa architecture, which avoids running a batch system and a stream system side by side with the same logic written twice (Lambda vs Kappa architecture).

Streaming is not always the right tool. If the answer is needed tomorrow, a batch job is simpler to write, test and rerun. Choose a stream job when a result loses its value within minutes, and then settle these decisions up front:

• Key: what must be counted together, so it lands on one instance.
• Window: tumbling, hopping or session, and its size.
• Clock: event time for anything about the real world.
• Lateness: how long to wait, and what happens to events that arrive after that.
• Output: overwrite by key, or commit with checkpoints, so a replay doesn't double-count.