Part 8 · Derived history

Stream Processing: Logs, CDC, Event Time, Windows, and Effectively-Once

Lesson 18 derived one dataset from another by scanning a bounded input — a snapshot with a beginning and an end-of-file — and re-running deterministically from that fixed history. But the real world never reaches end-of-file: orders, clicks, sensor readings, and model-input events keep arriving forever. A stream is exactly batch processing without the end-of-file. That one change breaks every assumption batch leaned on. You can no longer wait for all the input before producing output, so you must decide when a result is "done enough" to emit; the input arrives in a different order than it happened, so you must separate when an event occurred from when you saw it; and a crash mid-stream can replay or duplicate work, so "ran exactly once" stops being free. This lesson builds the machinery — the durable log, change data capture, event time and windows, watermarks, and effectively-once delivery — that lets unbounded data stay as correct and rebuildable as a batch job.

1 · The log is the abstraction

Start with the unit. An event is a small, immutable, self-contained record that something happened at a point in time — "user 42 clicked product 91 at 10:00:03". A stream is an unbounded sequence of such events. The question is what data structure carries them, and the answer that makes everything else work is a log: an append-only, totally-ordered sequence of records. This is the same primitive you met in lesson 03 as the write-ahead log — writes only ever go on the end, and the order they land in is the order of record. Here we hand that primitive to producers and consumers across the network.

A producer appends events to the end of the log. A consumer reads forward from a position it controls, called its offset — an integer index into the log saying "I have processed everything up to here." Crucially the broker does not remember consumers; each consumer owns its own offset and advances it at its own pace. Because the log is retained (not deleted as it is read), a consumer can rewind its offset to zero and replay the entire history, or a brand-new consumer can join and catch up from the beginning. Replay is the whole point: it is what lets a derived view be rebuilt from the source of truth, exactly as batch could re-scan a file (lesson 18).

A single log on one machine cannot absorb an unbounded write rate, so the log is partitioned — split into independent sub-logs by a partition key, exactly the mechanism from lesson 11. Order is guaranteed only within a partition; events in different partitions have no relative order. The standard trick is to key by an entity id (user, account, device) so that all events for one entity land in one partition and stay strictly ordered, while the system scales out by adding partitions. This is the same trade as a shard key: throughput scales with partitions, but a hot key (one celebrity user) overloads its partition (lesson 11).

Message queue vs log-based broker

Two delivery models look superficially similar and are deeply different. A message queue (the classic JMS/AMQP/SQS model) treats a message as a task: the broker hands it to one consumer, waits for an acknowledgement, and then deletes it. The message is gone; there is no history; load is spread across consumers because each message goes to exactly one of them. That is ideal for distributing work — a job that should be done once by some worker. A log-based broker (Kafka-style) instead retains every message in the ordered log and lets each consumer track its own offset. Nothing is deleted on read; messages age out only by a time- or size-based retention policy.

Stream processing needs the log-based model precisely because consumers fail, code changes, and derived state must be rebuilt. When you fix a bug in your feature-extraction code, you do not want the events to be gone — you want to reset the offset to zero, replay the full history through the corrected code, and rebuild the derived state from scratch. A queue that deleted on ack threw that ability away. The log-based broker is, as people say, the filesystem of the streaming world: append, retain, replay.

2 · Change data capture: the database is a stream

Here is the problem that makes streams unavoidable in practice. Your system of record is a database. But you also keep derived copies of that data for fast reads: a search index, a cache, a feature store, a vector index for retrieval. Every one of those must reflect what the database says. The naive approach is the dual write: in application code, write to the database and write to the search index. This is broken — the two writes are not atomic, so a crash between them, or two concurrent updates applied in different orders to the two systems, leaves them permanently disagreeing (this is the cross-system version of the race conditions from lesson 13). You cannot make a dual write correct by trying harder.

Change data capture (CDC) fixes this with one observation: the database already produces a perfect ordered stream of every change it makes — its replication log, the same write-ahead/replication log from lessons 03 and 08 that followers tail to stay in sync. CDC taps that log and publishes each committed row change (insert/update/delete) as an event onto a log-based broker. Now there is exactly one write — the database commit — and it is the single source of truth. Every derived system becomes a consumer that tails the change stream and applies the changes in commit order. The search index, cache, and feature store can never disagree with the database for long, because they are all replaying the same ordered log of facts.

Event sourcing (briefly) pushes this idea up into the application's own design. Instead of storing only current state and emitting changes as an afterthought, you make the log of events the primary record — "cart_item_added", "order_placed", "order_cancelled" — and derive current state by replaying those events. The difference from CDC is the level: CDC extracts a change stream from a database's low-level row mutations, whereas event sourcing records high-level, domain-meaningful events from the start. Both rest on the same foundation: an ordered, immutable history of facts from which any number of read-optimized views can be built and rebuilt. The cost event sourcing adds is that you now own event schema design, versioning (lesson 05's schema evolution applies directly), and periodic snapshotting so you need not replay from the dawn of time on every restart.

3 · Time, windows, and watermarks

Batch jobs had it easy: the input was a fixed snapshot, and one timestamp column was as good as another. Streams force a distinction. Event time is when the event actually occurred, stamped by the source ("the click happened at 10:00:03"). Processing time is when your stream processor observed it. In a perfect network they would be nearly equal — but they diverge, sometimes wildly. A mobile client goes into a tunnel, queues its events for an hour offline, and uploads them at 11:05; the events happened at 10:00 but you see them at 11:05. A backfill replays last week's log today. A consumer falls behind and catches up in a burst. If you bucket by processing time you will attribute that 10:00 click to the 11:05 minute — your hourly metrics, your real-time features, your monitoring all become wrong in a way that correlates exactly with network trouble. You almost always want to group by event time.

Grouping unbounded data requires cutting it into finite windows. The four standard shapes:

Now the core difficulty. A 1-minute tumbling window for event time 10:00:00–10:00:59 should contain every event that happened in that minute. But because event time and processing time diverge, events for that window can keep dribbling in long after 10:01 wall-clock. When do you declare the window finished and emit its result? Wait forever and you are always perfectly correct but never produce output. Emit the instant 10:01 wall-clock arrives and you will miss every straggler. You need an explicit policy, and that policy is a watermark.

A watermark is the processor's assertion: "I believe I have now seen all events with event time ≤ T." It is a moving low-water mark on event time, trailing behind the latest event time seen by some delay you choose. When the watermark passes the end of a window, the processor decides the window is complete enough and emits its result. The watermark delay is the knob: a larger delay waits longer, so it captures more late events (more correct), but holds the result back longer (higher latency) and keeps window state in memory longer (more cost). A smaller delay emits sooner but drops or mis-handles more stragglers. This is the freshness-versus-correctness trade, made into a single number.

Stream joins: three kinds, all subtle

Once you can window a stream by event time, you can join two streams — combine records that belong together, the streaming cousin of the SQL join. But a join assumes both sides are present at the moment you match them, and on an unbounded input nothing is ever fully present. That single fact splits stream joins into three kinds, and the time semantics from §3 are what make each one work or quietly break.

The trickiest of the three is the stream-table join, precisely because of event time versus processing time. Say a click stream is enriched with a "user tier" table that the user can change. A click happened at 10:00 when the user was tier gold; the user upgrades to platinum at 10:05; you finally process that 10:00 click at 10:07. If you read "whatever the table says now" you enrich the old click with platinum — and worse, a replay months later would read a different table state and produce a different answer, so the job is no longer deterministic. The correct join reads the table value as of event time 10:00 (tier gold), which means the table side must retain its history as a changelog you can index by time — exactly the CDC change-event log from §2. This is why "join a stream to a table" is really "join two streams, one of which you fold into current state."

Two consequences fall out. First, event-time vs processing-time choice governs join correctness, not just window counts: a stream-stream join keyed on processing time would pair a click with whatever impression happened to arrive near it on the wire, which (as in §3) skews exactly when the network is bad. Second, join state is bounded by the window, not by the stream — the 30-minute window means you only buffer 30 minutes of one side, and watermarks (§3) tell you when it is safe to evict an entry because no further match can arrive within its window. Pick the window too wide and join state (and cost) balloons; too narrow and you silently miss valid late matches — the same completeness-versus-cost knob the watermark widget below lets you feel.

4 · Feel the knob: the watermark widget

Each dot below is an event, plotted by its event time. Most arrive close to on-time, but a tail arrives late (a slice of clients were briefly offline). The watermark delay sets how long after a window's end you keep waiting before you emit. Drag it and watch two numbers move in opposite directions: the fraction of late events captured (completeness) climbs, while the emit latency you add to every window grows with it.

// each event has an arrival lag = (processing time - event time), in seconds.
// the watermark waits `delay` seconds past a window's end before emitting.
// an event is captured iff it arrives before the watermark passes -> lag <= delay.
for (const e of events) {
  if (e.lag <= delay) captured++;   // made it in before the window emitted
  else                 dropped++;   // late: drop or route to a side output
}
const completeness = captured / events.length;  // climbs with delay
const addedLatency = delay;                      // every window result waits this long

5 · Fault tolerance: there is no exactly-once, only effectively-once

A stream processor runs forever, so it will crash mid-flight, and when it restarts it must not lose events nor double-count them. The marketing phrase is "exactly-once," and taken literally it is impossible: a processor can read an event, update state, and write output, then crash before recording that it finished — on restart it cannot tell whether the output was written, so it must redo the work, and the work happens twice at the level of physical execution. You cannot make the physical execution happen exactly once across crashes. What you can guarantee is that the observable effect is as if each event were processed once. That is effectively-once (sometimes "exactly-once semantics"), and it is built from three ingredients.

Put together: checkpoint state and offset atomically, replay from the checkpoint on crash, and make the writes idempotent so the replay is invisible. The output looks as though every event was processed once — effectively-once — even though, physically, the replayed slice ran twice. Note the dependency on earlier lessons: the atomic "advance offset with output" step is exactly a transaction (lesson 13), and where output spans systems it needs the cross-system coordination that consensus underpins (lesson 17). Effectively-once is not magic; it is idempotence plus a transactional checkpoint plus replay.

Backpressure: what happens when the consumer can't keep up

Crashes are the dramatic failure; the quiet one is simply that the producer is faster than the consumer, for longer than a burst. Events arrive at 80,000/s; your processor drains 50,000/s. Where do the extra 30,000/s go? There are only three answers, and every streaming system is some combination of them.

The deep reason a log-based broker gives backpressure almost for free is the pull model from §1. The broker never pushes; each consumer pulls at its own pace and the unread suffix of the log simply sits on the broker's disk. A slow consumer's offset falls further behind the head — its lag grows — but nothing upstream has to block and nothing is dropped, because the durable log is the buffer, sized in hours of disk rather than seconds of RAM. Lag is the pressure gauge: you watch it, alert on it, and add consumer parallelism when it trends up. A push-based pipeline (a synchronous RPC chain, or in-memory reactive streams) has no such free buffer and must implement backpressure as an explicit protocol — credit/permit windows, or TCP-style flow control — or it falls back to dropping or OOMing.

Stream-table duality and the batch/stream convergence

One more idea pays off in the next lesson. A table is a snapshot of current state; a stream is the log of changes to that state. They are two views of the same thing — the stream-table duality. Replay the change stream from the beginning and fold it up and you reconstruct the table (this is exactly what CDC consumers and event sourcing do, §2); conversely, watch a table and emit every change and you get a stream. Once you see that a table is just a materialized fold over a log, the wall between batch and stream starts to dissolve: a batch job is a stream processor over a bounded prefix of the log, and a stream processor is a batch job that never finishes. That convergence — and the question of whether to run one unified pipeline or a fast-but-approximate stream layer next to a slow-but-correct batch layer — is the lambda/kappa debate the next lesson (20, derived-data correctness) takes up.

Checkpoint exercise

Where this points next

We now have both halves of derived history: lesson 18's batch layer, which is slow but reprocesses the whole history deterministically, and this lesson's stream layer, which is fast but must reason about time, lateness, and replay to stay correct. The next lesson (20) opens Part 9 — derived views — by putting both halves in one frame. It asks the question every real system faces: caches, search indexes, feature stores, and RAG vector indexes are all derived data that must be kept consistent with a source of truth — so do you run a fast approximate stream layer beside a slow exact batch layer and reconcile them (the lambda architecture), or treat the durable log as the single pipeline and reprocess by replay (the kappa architecture)? The stream-table duality you just met is the hinge of that argument, and keeping derived views correct end to end is where Part 9 begins.

Interview prompts

• Contrast a message queue with a log-based broker, and say when each wins. (§1 — a queue deletes a message on ack and load-balances it to one consumer (good for one-time work dispatch); a log-based broker retains every message in an ordered log and lets each consumer track its own offset, enabling replay, many independent consumers, and rebuilding derived state — the model stream processing needs.)
• What is an offset, and why does the broker not track it for the consumer? (§1 — the offset is the consumer's position "processed up to here"; the consumer owns it so it can advance at its own pace, replay by rewinding to zero, and let many independent consumers read the same log without interfering.)
• Why is a dual write broken, and how does CDC fix it? (§2 — writing to the DB and the index in app code is non-atomic, so a crash or reordering between them leaves them permanently inconsistent; CDC publishes the database's replication log as one ordered change stream, so there is a single source-of-truth write and every derived system is a consumer replaying the same changes.)
• Event time vs processing time — why do they diverge and which should you window by? (§3 — event time is when it happened, processing time when you saw it; an offline mobile client, a backfill, or a lagging consumer makes them differ; group by event time or your metrics/features skew exactly when the network is bad.)
• Define a watermark and the trade it controls; work a number. (§3 — a watermark is "I believe I've seen all events with event time ≤ T," trailing the latest event time by a chosen delay; when it passes a window's end the window emits. A larger delay captures more late events but raises latency and state cost. Worked: a 1-min window with a 10-s delay emits ~10s after its end, and a 30-s-late event is dropped or sent to a late-data path.)
• Why can't you process exactly once, and what gives you effectively-once instead? (§5 — a crash between doing work and recording that it finished forces a replay, so physical execution can repeat; effectively-once makes the observable effect as-if-once via idempotent writes + offsets checkpointed atomically with state + replay from the last checkpoint.)
• Explain the stream-table duality and why it matters for batch/stream convergence. (§5 — a table is a snapshot of state, a stream is its log of changes; folding the stream rebuilds the table and watching the table emits the stream, so a batch job is a stream over a bounded prefix and a stream is a never-ending batch — the hinge of the lambda/kappa choice in lesson 20.)
• A consumer is permanently slower than its producer. What are your options, and why does a log-based broker handle this better than a push pipeline? (§5 — three responses: drop (lose data, only with a counter), buffer (absorbs bursts but a sustained overload OOMs an unbounded queue), or backpressure (propagate "I'm full" upstream until the source slows). A pull-based log gives backpressure almost free: the consumer pulls at its own rate, the unread suffix sits on disk, and lag is a bounded, observable buffer — safe as long as lag < retention, after which unread events age out and are lost silently.)