Stream processing patterns

⏱ 33 min

Hook

Your streaming job restarts after a deploy. When it comes back, the running per-user totals it had been accumulating are gone, so it recounts — and every event the source re-sends during the reconnect gets counted a second time. Totals that were right an hour ago are now inflated, and no single line of code is wrong. The problem is that the job holds state, and state plus redelivery plus restarts is where streaming stops being about windows and starts being about guarantees.

Concept

Three patterns turn windowing into a working processor. Each carries an operational cost you should name before you adopt it.

Stateful aggregation. A processor that computes running counts, sums, or top-N per key must remember something between events — the partial aggregate for every open window and key. That memory is state, and unlike a batch job that starts empty and exits, a streaming job holds state for as long as it runs. The cost is operational weight: state must be checkpointed so a restart can resume it, and it must be bounded so it does not grow forever. Watermarks pull double duty here — closing a window is also what lets the processor delete that window's state. A stateful job without a mechanism to expire state is a memory leak with a schedule.

Handling late and out-of-order events. Records do not arrive in event-time order, and some arrive after the watermark has already closed their window. A processor must decide, on purpose, what to do with such a late event. Three policies are common: drop it (simplest, and correct when the window is truly final), send it to a side output for separate handling (an audit trail, a correction job), or reopen and update the window (most correct, most expensive). The wrong move is to have no policy — then late events get silently folded into whatever window happens to be open, corrupting an unrelated result.

Delivery semantics. When a source or sink can fail and retry, how many times does a record get processed? Two guarantees frame the answer.

Guarantee	Promise	Cost
At-least-once	Every record is processed, maybe more	Duplicates are possible; you must tolerate them
Exactly-once	Every record's effect lands exactly once	Coordinated checkpoints and idempotent or transactional sinks

At-least-once is the cheap, common default: on failure the source replays recent records, so nothing is lost, but a record processed just before a crash can be processed again after recovery. Exactly-once does not mean a record is literally handled once; it means its effect on the result appears once, no matter how many times it is delivered. Systems achieve it by combining checkpointed progress with sinks that are idempotent (applying the same write twice equals applying it once) or transactional (writes commit atomically with the progress marker).

There is a pragmatic middle path worth its own name: effectively-once, where an at-least-once source feeds a processor that deduplicates by a unique event id. You accept duplicate delivery and neutralize it with a dedup key, getting exactly-once results without full transactional machinery. It is often the best cost-to-correctness ratio a team can buy, which is why you will build it in the worked example.

🧠 Knowledge check

1. What does "exactly-once" actually guarantee?

a) A record is physically transmitted only one timeb) Each record’s effect on the result appears once, regardless of how many times it is deliveredc) The system never fails or restartsd) Late events are always dropped

2. Why must a long-running stateful streaming job expire window state?

a) To make results less accurateb) Because unbounded, never-released state grows until the job runs out of memoryc) Because state cannot be checkpointedd) To force every event to be reprocessed

Worked example

Let me build the effectively-once pattern, because it ties the three ideas together in a few lines. Imagine an at-least-once source that, during a reconnect, re-delivers one event — evt-102 arrives twice. A naive running count would tally it twice. The fix is a dedup key store: remember which event ids you have already folded in, and skip repeats. That store is state, and it is what makes the aggregation idempotent under redelivery.

from collections import defaultdict

delivered = [
    ("evt-100", "checkout"),
    ("evt-101", "checkout"),
    ("evt-102", "signup"),
    ("evt-102", "signup"),   # duplicate redelivery from an at-least-once source
    ("evt-103", "checkout"),
]

counts = defaultdict(int)
processed = set()  # the dedup key store — remembering ids makes the fold idempotent
for event_id, action in delivered:
    if event_id in processed:
        continue
    processed.add(event_id)
    counts[action] += 1

print(dict(counts))

{'checkout': 3, 'signup': 1}

The source delivered five records, but only four distinct events happened, and the result reflects the four. Drop the processed guard and signup jumps to 2 — the duplicate silently inflates the total, exactly the restart bug from the hook. The processed set gave us exactly-once results over an at-least-once stream, with no transactional sink required.

Notice the cost the pattern quietly incurs, and connect it back to stateful aggregation: processed grows by one entry per event, forever, unless you bound it. In a real job you would expire dedup keys once the watermark guarantees no further duplicate for them can arrive — the same watermark-drives-state-cleanup principle from windowing, now protecting your dedup store instead of your window counts. Every streaming guarantee is paid for in state, and every piece of state needs an expiry rule.

Hands-on

Your turn, with a different stream and a different guarantee decision. The exercise below hands you an at-least-once feed of payment events containing duplicate deliveries and one late event that arrives after its window has closed. You will implement the dedup-by-id pattern for effectively-once counts and apply a stated late-event policy, then report the per-key totals and how many late events your policy dropped.

▶ Python Arenade205-effectively-once-with-late-policy

Open in Python Arena ↗

Success criteria: duplicates are counted once, the late event is handled per the stated policy, and your totals match the reference. Python Arena grades the totals and the late-event count automatically.

Recap

• You can describe stateful aggregation and explain why a long-running job must bound and expire its state, with the watermark authorizing cleanup.
• You can name three policies for late events — drop, side output, or update — and explain why having no policy corrupts results.
• You can distinguish at-least-once from exactly-once, define exactly-once as a guarantee about effects, and build effectively-once with a dedup key.

Next up, in the lab, you assemble these ideas into a running processor: a generator emits out-of-order and late events, and you implement a tumbling-window aggregation whose watermark decides when each window closes and what happens to the events that arrive too late.

🧠 Knowledge check

1. An at-least-once source occasionally re-delivers events. How can you get exactly-once results without a transactional sink?

a) By switching to processing timeb) By deduplicating on a unique event id so repeated deliveries fold in only oncec) By dropping every event that arrives out of orderd) By disabling checkpoints

2. A record arrives after the watermark has already closed its window. Which is NOT one of the standard late-event policies?

a) Drop the recordb) Route it to a side output for separate handlingc) Reopen and update the windowd) Silently add it to whichever window is currently open

📌 Key takeaways

• Stateful aggregation holds partial results between events; that state must be checkpointed and expired, and the watermark authorizes the cleanup. - Late events need a deliberate policy — drop, side output, or update — never a silent fallthrough. - At-least-once permits duplicates; exactly-once guarantees effects land once; effectively-once buys exactly-once results with a dedup key over an at-least-once stream.