Everything computed so far in this module started from a closed dataset: the file for 14 September, seven days of clicks. But Kilometre Zero's data is not born closed. orders.events receives hundreds of stock.updated events per second during a campaign, the 140 couriers such as van-3 send 2.4 million positions a day, and there are questions that cannot wait for the nightly batch: which products are running out of stock in Girona now, where each courier is now, which courier has gone three minutes without a signal. Stream processing answers those questions by running the computation continuously, event by event, over a dataset that never ends. That breaks assumptions that came for free in batch: you cannot "wait until you have everything" before grouping, events arrive out of order and late with respect to the moment they happened, the state of the computation must survive failures without being able to re-run "from the beginning", and the input may arrive faster than it is processed. This lesson gives a name to each of those problems (event time, watermarks, windows, state and checkpoints, exactly-once, backpressure), compares the Lambda and Kappa architectures and the three usual tools (Kafka Streams, Flink, Spark Structured Streaming), and solves the two real-time cases of analytics: the low-stock alert and the delivery dashboard. Delivering those results to the operator's browser is left for 08-02; here we finish at a Kafka topic.

1. From batch to streaming: what changes
2. Event time, processing time and watermarks
3. Windows: tumbling, sliding and session
4. State and checkpoints
5. Exactly-once in streaming
6. Backpressure
7. Lambda and Kappa architectures
8. Tools: Kafka Streams, Flink and Spark Structured Streaming
9. The Kilometre Zero cases: low stock and the delivery dashboard
10. Hands-on: a Python simulation, PyFlink and Structured Streaming
11. Common Mistakes and Tips
12. Exercises
13. Conclusion

Let us go back to the table from 05-01 and sharpen it with what we have learnt since:

A stream engine runs the same kind of DAG of operators as Spark (05-03), but deployed permanently: each operator is a process (or several, in parallel by key) that waits for events, processes them and emits to the next one, with the DAG never finishing. The typical source is Kafka (02-04), which provides what a stream needs: partitions (parallelism), offsets (a resumable position) and retention (re-reading from further back). The sink is another topic, a database or a dashboard.

There are two ways of moving events through the DAG: event by event (Flink, Kafka Streams: millisecond latency) and in micro-batches (Spark Structured Streaming: every few hundred milliseconds or seconds it takes whatever has arrived and runs it as a small batch). The second reuses all the batch machinery and simplifies exactly-once; the first gives lower latencies. For the delivery dashboard and the stock alert, seconds are enough, so both will do.

In 01-05 we distinguished the moment something happened from the moment another node finds out about it. In streaming, that distinction is the one with the most consequences:

• Event time: when the fact happened, according to the clock of whoever generated it. It is the timestamp_ms of the envelope from 02-05. The position of van-3 taken at 10:04:58 has that time even if the phone sends it at 10:07 for lack of coverage.
• Processing time: when the operator sees the event, according to its own clock. It depends on the queue in Kafka, on the network, on the load.

The question "how many stock.updated events for aged-cheese were there between 10:00 and 10:05?" has only one correct answer in event time. With processing time, a consumer restart at 10:03 would put events that happened at 10:02 into the 10:05–10:10 window; the result would depend on the system's behaviour, not on the facts. But computing in event time creates a new problem: when an event with timestamp_ms = 10:04:58 arrives at 10:07, had the 10:00–10:05 window already been emitted? Does it have to be reopened? How long should you wait before considering it closed?

The answer is the watermark: a statement that the system sends flowing through the DAG saying "I expect no more events with an event time earlier than T". It is generated at the source with a heuristic, normally maximum event time seen − tolerated delay: with a tolerated delay of 2 minutes, after seeing an event from 10:07:00 the watermark is 10:05:00, and at that moment the 10:00–10:05 window is considered complete and is emitted. The tolerated delay is a matter of business judgement and of measurement: how long events really take to arrive (the 99th percentile of the gap arrival − timestamp_ms), against how much latency is acceptable in the result. A short tolerated delay: fast results that discard more events; a long one: more complete results, but later.

Events that arrive after the watermark has passed their window are late events. Every engine offers three destinations for them: discard them (the default behaviour), incorporate them by re-emitting the updated window for an additional margin (allowed lateness in Flink, which forces the window to be kept in state for longer), or divert them to a side output to count them, log them or reprocess them in batch. Counting the late ones is mandatory: it is the metric that tells you whether the tolerated delay has been well chosen.

With several Kafka partitions, each partition has its own watermark and the operator's is the minimum of them all: a partition with no traffic (a market closed for the night) holds back the global watermark and blocks the emission of everyone's windows. Engines deal with this through idle timeouts that exclude inactive partitions from the minimum.

Over an infinite stream, any aggregation ("how many", "the minimum") needs a boundary: the window. The three basic shapes:

Windows are combined with a key: "5-minute tumbling by (product, market)" keeps one window for each pair, and the operator's parallelism is by key (all the aged-cheese/girona events go to the same instance, as in the shuffle of 05-02). And with event time: it is timestamp_ms that decides which window an event falls into, and the watermark that decides when it is emitted.

An open window, a counter per key, the last known position of each courier: all of that is state, and in a stream it lives indefinitely in the operator. Engines keep it in a state backend local to the operator (memory, or RocksDB on local disk for gigabyte-sized states) partitioned by key, exactly like an actor per key (05-01). The problem is durability: if the node dies, its local state is lost, and you cannot "re-run from the beginning" because the beginning was months ago.

The solution is the checkpoint: periodically (every 10 s, every minute) the engine writes a consistent copy of the state of all the operators, together with the Kafka offsets that state reflects, to durable storage (HDFS, MinIO). On a failure, it restores the last checkpoint on healthy nodes and resumes consuming from those offsets: the events after the checkpoint are processed again, and the state ends up the same as if there had been no failure.

The hard word is consistent: the state of all the operators must correspond to the same point in the stream, even though each one is at a different event. Flink achieves it with an adapted Chandy-Lamport algorithm (checkpoint barriers): the source injects a marker with the checkpoint number into the stream; each operator, on receiving it on all its inputs, saves its state and forwards the marker; the saved state reflects exactly the events before the marker. It is a distributed snapshot taken without stopping the stream. Spark Structured Streaming has an easier time: each micro-batch is a unit, and the checkpoint records which micro-batches have been completed along with their offset ranges. Kafka Streams keeps its state in Kafka topics (changelog topics) and rebuilds it by re-reading them.

A savepoint is a manually triggered checkpoint, in a stable format, used to stop the job, change the code or the parallelism, and resume from the same state: the equivalent of a deployment without losing "the last five minutes". And the size of the state matters: a session window per courier is small, but "every order in the last 24 hours per customer" is gigabytes that have to be checkpointed every minute; incremental checkpoints (only what has changed since the previous one) and a TTL for state that is not touched are the tools.

In 02-05 we concluded that "exactly-once" delivery does not exist, and that what is achievable is at-least-once plus idempotent consumers. In streaming the term is used with a precise and achievable meaning: the engine's state reflects each event exactly once, even with failures and reprocessing. It is achieved with the checkpoint of the previous section: after a failure, the state goes back to that of the checkpoint and the later events are reapplied; since the state that reflected those events has been thrown away, nothing is counted twice.

What the checkpoint does not cover is what has already left the engine: the alert written to the inventory.alerts topic, the row inserted into PostgreSQL, before the failure and after the last checkpoint. On reprocessing, those effects happen again. For the external result to be exactly-once as well (end-to-end), the sink has to be one of two things:

• Idempotent. Writing the same result again changes nothing: an UPSERT by the key (window, product, market) in PostgreSQL, a PUT in Redis with the same key, a Kafka producer with a key and an idempotent consumer downstream (02-05). It is the simplest option and the one recommended whenever the output has a natural key.
• Transactional. The sink writes in a transaction that is only committed when the checkpoint completes (a two-phase commit sink): Flink with Kafka's transactional producer (DeliveryGuarantee.EXACTLY_ONCE), or with a table and one transaction per checkpoint. Between the pre-commit and the commit the data exists but is not visible to consumers with isolation.level=read_committed. It adds latency (the checkpoint interval) and complexity; it is reserved for sinks with no natural key (an event log in Kafka).

A streaming DAG is a chain of producers and consumers, and at any moment one may be going slower than the one before it: the window operator writing a large checkpoint, the PostgreSQL sink saturated, an Artisan Cheese Week spike that triples the stock.updated events. If the fast operator kept sending, the intermediate queues would grow until memory ran out. Backpressure is the mechanism by which slowness propagates backwards: the slow operator stops accepting, the previous one fills its output buffer and stops reading from its input, and so on back to the source, which stops consuming from Kafka. The events pile up in Kafka (which is designed for it, with days of retention), not in the engine's memory, and the consumer group's lag (02-04) becomes the metric that tells you the stream cannot keep up.

Flink implements it with credits between tasks (the receiver announces how much it can take); Kafka Streams gets it for free because each instance only polls when it has finished with the previous batch; Spark Structured Streaming approximates it by limiting how many offsets it reads per micro-batch (maxOffsetsPerTrigger). What none of them does is fix the cause: if the lag grows steadily, you have to increase the parallelism (more partitions and more instances), lighten the operator or accept approximate results. And mind the watermark: under backpressure, processing time drifts away from event time, but the windows are still correct because they are defined in event time; that is precisely what section 2 was buying.

When streaming was new and unreliable, the answer to "I want real-time results but exact ones too" was the Lambda architecture (Marz, 2011): keep two paths. The batch layer recomputes the complete, exact views from the lake every night; the speed layer computes the last few hours with streaming, approximately; a serving layer combines the two at query time. It works, but it forces you to write and maintain the same logic twice, in two engines, with two semantics, and to reconcile their differences. The Kappa architecture (Kreps, 2014) proposes a single path: everything is a stream, with Kafka retaining the history (or a re-readable event lake), and the "batch" is simply reprocessing the stream from an old offset with the same streaming application, writing to a new table and switching the pointer when it catches up with the present.

Kilometre Zero's data platform ends up being pragmatically mixed: daily_sales.py and ALS are batch because they need the complete dataset and are in no hurry; stock and delivery are streams because latency is the requirement. What Kappa contributes is the criterion: if the same logic is written twice, something is wrong; the fact that modern engines run the same code in batch and in streaming (Spark, Flink) makes the choice one of deployment, not of rewriting.

We choose Flink as the stream engine for analytics because of its event-time semantics and because of PyFlink, and we keep Structured Streaming as an alternative because it reuses the API of 05-03. Kafka Streams gets a mention: it is the right option if the streaming lives inside a Java service and not on a data platform.

Case (a): low-stock alert. inventory publishes stock.updated to orders.events on every change (04-05 used those events to invalidate Redis). Event data: product, market, current_stock, delta, replica (inv-bcn or inv-vlc). Requirement: every 5 minutes, per product and market, if the minimum stock observed has fallen below 10 units, emit an alert to the inventory.alerts topic with the window, the minimum and how many updates there were. A 5-minute tumbling window by (product, market), in event time (the timestamp_ms from inventory, because a lagging replica must not shift the alert), with a 2-minute watermark (measured: the 99th percentile of the gap is 40 s). An idempotent sink: the key of the alert message is window|product|market, so reprocessing produces the same message and the consumer of 08-02 will treat it as the same one.

Case (b): delivery dashboard. The positions of van-3 arrive over MQTT (02-01) and a bridge publishes them to the delivery.positions topic with key = courier id: {"courier":"van-3","lat":41.9794,"lon":2.8214,"timestamp_ms":...}. Two computations: the distance covered in the last 5 minutes, refreshed every minute (a 5/1 sliding window per courier: it detects couriers that are stopped or off route), and couriers with no signal (a session window with a 3-minute gap: when the session closes, the courier has gone 3 minutes without sending; when a new one opens, it is back). Both write to delivery.dashboard, which 08-02 will push to the operators' browsers.

Before using an engine, it helps to see the bare mechanism. This script processes a list of stock.updated events with two times each, the event time (timestamp_ms) and the arrival time (arrival_ms), in arrival order, keeping 5-minute tumbling windows per product and a watermark with a 2-minute tolerated delay. The times are in minutes since 10:00 to make them easy to read.

What the trace shows:

• The t=4.9 event that arrives at 10:07 does make it into the 0–5 window, because the watermark at that moment was 4.5 (< 5): the window was still open thanks to the tolerated delay. With processing time it would have fallen into 5–10, and the minimum of the 0–5 window would have been 12, with no alert.
• The 0–5 window is emitted when the t=7.3 event arrives: the watermark moves to 5.3 ≥ 5. Not before (it was not known whether events were missing) and not after (there is no need to wait any longer).
• The t=4.2 event that arrives at 10:09:30 is late: its window has already been emitted. Here it is discarded and counted; with allowed lateness the 0–5 window would be re-emitted with min=8, n=5 (the minimum does not change, but the count does).
• When the input ends, two windows are left open (10–15): in a real stream there is no "end", and they will be emitted when the watermark reaches 15. In a batch, the end of the input triggers the emission of everything.

Notice too that the watermark does not advance with the late event and never goes backwards: it is monotonic. And that every EMIT line is an output which, if the process died and reprocessed from event 1, would be produced again with the same values: the key (window, product) makes it idempotent.

The Flink cluster is added to docker-compose.yml with two services (a jobmanager and a taskmanager from the flink:1.19-python image, or a custom image with pip install apache-flink), and the job is submitted with flink run -py. With the Table API, the DAG of case (a) is three SQL statements:

Each piece corresponds to a section: ts AS TO_TIMESTAMP_LTZ(timestamp_ms, 3) declares the event time; WATERMARK FOR ts AS ts - INTERVAL '2' MINUTE the watermark (Flink generates it per Kafka partition and takes the minimum, with the idle-timeout for stalled partitions); TUMBLE(..., INTERVAL '5' MINUTE) the window, with window_start/window_end as columns; the GROUP BY is the keyBy that spreads the state by key across TaskManagers; execution.checkpointing.interval the checkpoint that saves state and offsets to the configured state.checkpoints.dir (HDFS /km0/checkpoints/); and upsert-kafka with a PRIMARY KEY the idempotent sink: reprocessing after a failure rewrites the same key with the same value. It is launched and observed like this:

The JobManager's web UI (localhost:8081) shows the deployed DAG, the parallelism of each operator, the current watermark of each one, the checkpoints (duration, size) and the backpressure per operator, colour-coded.

For case (b), the same structure with SESSION for the couriers with no signal:

Each row emitted means "the courier van-3 sent positions continuously between session_start and session_end, and then went at least 3 minutes with no signal": it is the dashboard's warning. The sliding window for distance is written with HOP(TABLE positions, DESCRIPTOR(ts), INTERVAL '1' MINUTE, INTERVAL '5' MINUTE) and a custom aggregate function (the distance between consecutive positions needs the order, which in SQL is handled with LAG over an OVER window before aggregating).

The same computation with the DataFrame API of 05-03, in streaming mode:

The correspondence with Flink is direct: withWatermark ↔ WATERMARK FOR, F.window("ts", "5 minutes") ↔ TUMBLE, checkpointLocation ↔ execution.checkpointing, maxOffsetsPerTrigger ↔ backpressure. Two important differences. outputMode("append") with a watermark emits each window only once, on closing it, and discards late events with no possibility of re-emitting (there is no allowed lateness); update would emit provisional results in every micro-batch, which demand a sink that knows how to overwrite. And Spark's Kafka sink is at-least-once: the message may be duplicated after a failure, and it is the key (identical in the duplicate) that makes the consumer of 08-02 treat it as a harmless repeat, exactly the idempotent consumer of 02-05. It is launched with spark-submit --packages org.apache.spark:spark-sql-kafka-0-10_2.12:3.5.1 stock_alerts_spark.py, and at localhost:4040 a Structured Streaming tab appears with the input rate, the duration of each micro-batch and the watermark.

• Using processing time because it is easier. The results depend on load, restarts and backpressure, and cannot be reproduced. If the event has a timestamp (and with the envelope from 02-05 it always does), use event time.
• An unmeasured watermark. A made-up tolerated delay discards valid events or delays the results. Measure the gap arrival − timestamp_ms in production (99th percentile) and count the late events as a permanent metric.
• Forgetting about idle partitions. A Kafka partition with no traffic holds back the global watermark and no window is emitted. Configure the idle timeout (Flink) or check that every partition receives events.
• Unbounded state. An aggregation by key with no window and no TTL grows for ever (one key per order_id, for example). Windows, a state TTL, or bounded keys.
• A non-idempotent sink with reprocessing. After a failure, the outputs after the checkpoint are repeated. A natural key + upsert, or a transactional sink. Never an INSERT with no key or an append to a file.
• Checkpoint on local disk. If the node dies, the checkpoint dies with it. HDFS, MinIO or replicated storage, always.
• Changing the DAG and resuming from the checkpoint. A checkpoint saves state per operator; if the DAG changes (a new operator, a different key) it may not be compatible. Savepoints with operator identifiers (uid) in Flink; in Spark, changes to the state schema usually require starting from scratch.
• Confusing lag with latency. The consumer group's lag (events pending in Kafka) grows under backpressure and is the sign that parallelism is lacking; the latency of a result in event time is always at least the tolerated delay plus the window size.
• Writing the logic twice (Lambda out of inertia). If the same aggregate is computed in batch and in streaming, the two will come out different and nobody will know which is the right one. One codebase, two deployments.

An analysis of the delivery.positions topic over a week shows that the gap between timestamp_ms and arrival in Kafka has this distribution: median 1.2 s; 95th percentile, 8 s; 99th percentile, 45 s; 99.9th percentile, 4 min (tunnels, areas with no coverage); maximum 22 min (a switched-off phone that resent on being switched on). The dashboard must show the position and the distance for the last 5 minutes with no more than 1 minute of delay. Choose the watermark's tolerated delay, the type and size of window, and what to do with late events. Justify the percentages of events that will be discarded and what happens with the phone that resends after 22 minutes.

Run tumbling_window.py in your head (or by modifying the script) with TOLERATED_DELAY = 0 and with TOLERATED_DELAY = 5. For each case, state when the 0–5 window for aged-cheese is emitted, with what minimum and count, and how many late events there are. Which of the three configurations (0, 2, 5) would give the correct alert soonest?

The PyFlink alerts job takes checkpoints every 30 s. At 10:07:50 the TaskManager running the aged-cheese/girona window dies; the last completed checkpoint is from 10:07:30, and at 10:07:40 the job had emitted the alert for the 10:00–10:05 window. Describe what Flink does on recovery: which offsets it reads from, what happens to the state of the 10:05–10:10 window, whether the 10:00–10:05 alert is emitted again and what the consumer of inventory.alerts sees. Then explain what would change if the sink were an INSERT into PostgreSQL with no primary key, and how you would fix it.

Exercise 1.

The latency budget is 1 minute, and the minimum latency of a result is the tolerated delay (plus the emission interval). A tolerated delay of 45 s (the 99th percentile) meets the budget and discards 1% of the positions as late; with 8 s (p95) 5% would be discarded, too much for a position trace; with 4 min (p99.9) the 1-minute requirement would be violated. A 5-minute sliding window with a 1-minute slide per courier (or 30 s if the dashboard must refresh more often, at the cost of more open windows per key: 10 instead of 5). The late events (1%) go to a side output that is counted and written to the lake: for the real-time dashboard it makes no difference to lose one position in a hundred, but the day's distance covered that the nightly batch computes must include them, and for that the batch reads all the positions in the lake, late ones included (a justified de facto Lambda). The phone that resends after 22 minutes delivers positions whose window closed 20 minutes ago: all late, all to the side output; the dashboard ignores them and the batch incorporates them. An alternative if the business asked for it: an allowed lateness of 5 minutes to re-emit recent windows, not of 22 (that would keep 27 minutes of windows in state per courier).

Exercise 2.

With TOLERATED_DELAY = 0 the watermark is the maximum event time seen. When t=6.1 arrives (arrival 6.2) the watermark is 6.1 ≥ 5 and the 0–5 window is emitted with the events seen up to then: 0.5, 1.2, 3.8 and... the t=4.9 one arrived at 7.0, afterwards, so the window is emitted with min=12, n=3: no alert, incorrect. The t=4.9 (arrival 7.0) and t=4.2 (arrival 9.5) events are both late: 2 late events. Earliest emission, wrong result.

With TOLERATED_DELAY = 5, the 0–5 window is emitted when the watermark reaches 5, that is, on seeing an event with t ≥ 10: the t=10.2 one (arrival 10.3). By then 0.5, 1.2, 3.8, 4.9 and 4.2 (which arrived at 9.5 with the window still open) have gone in: min=8, n=5, correct and complete, 0 late events. But the alert goes out at 10:10:18, five minutes later than with a delay of 2 (10:07:24, min=8, n=4).

The 2-minute configuration gives the correct alert (the minimum of 8 was there in both) soonest; the 5-minute one also gives the exact count; the 0 one fails. It is the completeness/latency trade-off of section 2, and the reason the tolerated delay is measured and not guessed.

Exercise 3.

Flink detects the TaskManager's death (heartbeat) and restarts the whole job (or the affected region, with fine-grained recovery) from the 10:07:30 checkpoint: it restores the state of all the operators as it was then (the 10:05–10:10 window with the events before 10:07:30 already applied; the 10:00–10:05 one, which had not yet been emitted at that instant, also restored with its state) and repositions the Kafka consumer at the offsets saved in that checkpoint. The events between 10:07:30 and 10:07:50 are read again and applied on top of that state: none is counted twice, because the state that contained them was thrown away. The watermark advances again and the 10:00–10:05 window is emitted again, with the same content. Since the sink is upsert-kafka with the key (window_start, product, market), the topic receives a second message with the same key and the same value; the consumer of 08-02 (or Kafka's compaction) treats it as an update with no changes. Exactly-once in the state, and effectively a single visible alert.

With an INSERT into PostgreSQL with no key, the row for the 10:00–10:05 alert would exist twice. Fixes, from least to most effort: a primary key (window_start, product, market) with INSERT ... ON CONFLICT DO UPDATE (an idempotent sink; Flink's JDBC sink does it with upsert); or the JDBC sink in exactly-once mode with XA (two phases, committing with the checkpoint), which adds the checkpoint latency to every alert. For alerts, the first option is the right one.

Processing a stream means permanently running the same DAG of operators as a batch, over an input that never ends, and that forces you to answer questions batch never had: when a window is complete (the watermark, derived from event time and from a tolerated delay that is measured), what to do with what arrives afterwards (discard, re-emit or divert the late events), how to group (tumbling, sliding and session windows by key), how to make durable a state that lives for ever (checkpoints consistent with the offsets, savepoints for deploying) and how not to count twice after a failure (exactly-once in the state through checkpoints, and in the sink through idempotency or a transaction). Backpressure protects the engine by letting Kafka absorb the spikes, and lag is the signal. Lambda and Kappa are two ways of living alongside batch: the first duplicates the logic, the second unifies it, and modern engines make the choice one of deployment. At Kilometre Zero, simulations/tumbling_window.py made the mechanism visible, PyFlink solved the low-stock alert with three SQL statements and an upsert-kafka sink, the session window detects couriers with no signal, and Spark Structured Streaming showed that the code of 05-03 works almost unchanged with withWatermark and window.

With this lesson, analytics has its two halves: the batch jobs of daily_sales.py and ALS, and the streams for alerts and delivery. But batch jobs do not launch themselves. Somebody has to wait for the day's file to be complete in HDFS, validate it, launch spark-submit, load the result into the analytics database, raise the alarm if something fails and retry, and do it every day, and for the seven days of Grape Harvest Week when the wine price was corrected. Until now that was a cron and a chain of scripts. The last lesson of the module deals with job scheduling and data pipelines: how to express those dependencies as a DAG of tasks in Airflow, with sensors, retries, backfill and alerts, so that Kilometre Zero's data platform runs without anybody launching it by hand.