Lab 02 — A Mini Beam/Dataflow Processing Engine

Phase 27 · Lab 02 · Phase README · Warmup

The problem

Every large-scale data system eventually faces the same fork: you have a batch pipeline that recomputes features nightly over the warehouse, and a streaming pipeline that updates them per event — and they are two codebases that drift apart until the model trains on one definition and serves on another. Apache Beam's founding idea (the Dataflow model, from Google's MillWheel/FlumeJava lineage) is that this fork is false: batch is just streaming over a bounded source. One transform algebra — Map, Filter, FlatMap, GroupByKey, CombinePerKey — plus event-time windowing and a watermark that declares "event time has advanced to here," and the same pipeline runs on a file or an infinite stream.

The concepts you must be able to explain on a whiteboard:

1. Event time vs processing time. An element carries the timestamp of when the thing happened, not when the pipeline saw it. All windowing is over event time.
2. Windowing. Fixed/tumbling windows partition the timeline (each event in exactly one window). Sliding windows overlap (size 10, period 5 → every event lands in 2 windows) — how you compute "the last 10 minutes, updated every 5."
3. Watermark + trigger. In a stream you can never be sure a window is complete. The watermark is the runner's moving estimate of event-time progress; the default trigger fires a window's aggregate when the watermark passes the window's end.
4. The shuffle. GroupByKey is the expensive, network-moving heart of every distributed engine (Beam, Spark, MapReduce). CombinePerKey is the optimizable version — associative aggregation that can pre-combine before the shuffle.

What you build

File map

Run it

pytest test_lab.py -v                       # your lab.py (red until you implement)
LAB_MODULE=solution pytest test_lab.py -v   # the reference (green)
python solution.py                          # worked example

Success criteria

• map/filter/flat_map transform values while preserving timestamps and windows, and chain functionally.
• group_by_key groups (key, value) tuples per (window, key) — the same key in two windows produces two groups — and raises KeyedError on non-tuple input.
• combine_per_key(sum) / custom combiners aggregate deterministically.
• FixedWindows(10) puts t=9 in [0,10) and t=10 in [10,20) — exactly one window per event.
• SlidingWindows(size=10, period=5) puts t=7 in both [0,10) and [5,15).
• window_into raises WindowError on an element with no event timestamp.
• The StreamingRunner does not fire a window at watermark=9 for [0,10), does fire at watermark=10, never re-fires, holds later windows open, and rejects a backwards watermark.
• The same events produce identical results through the batch pipeline and the streaming runner, for both fixed and sliding windows.
• All 30 tests pass under both lab and solution.

How this maps to the real stack

• PCollection is Beam's PCollection, literally: an immutable collection whose elements carry event timestamps and window assignments. Our map/filter/flat_map are the ParDo / Map / FlatMap PTransforms; group_by_key/combine_per_key are GroupByKey / Combine.perKey. Real Beam fuses element-wise stages and distributes the GroupByKey shuffle across workers; ours runs in one list so the semantics are inspectable.
• FixedWindows / SlidingWindows are Beam's classes of the same names (beam.window.FixedWindows(60), SlidingWindows(600, 60)); Beam adds session windows (gap-based, per-key) which make a good extension below.
• The watermark is the real mechanism Google Dataflow tracks per stage: an estimate of the oldest unprocessed event time. Our advance_watermark is the test-friendly injection of what a runner derives from its sources. The default Beam trigger — AfterWatermark() — is exactly "fire when the watermark passes the end of the window," which is what our runner implements.
• What we deliberately left out is the hard 20%: late data (an element arriving after the watermark passed its window) with allowed-lateness and pane accumulation (discarding/accumulating), exactly-once effects via idempotent sinks and checkpointed state, and autoscaling the shuffle. Real Dataflow's value is running this model at scale with those guarantees; the model itself is what you just built. Our runner also fires a (window, key) exactly once and drops nothing — real Beam lets you trade completeness vs latency with early/late triggers.
• Google Dataflow vs Spark Structured Streaming: Spark expresses the same ideas as micro-batch (or continuous) queries over unbounded tables with watermarks for state eviction; Beam is the portable API (Dataflow, Flink, Spark runners). The interview line: "windowing + watermark + trigger is the vocabulary; Beam/Dataflow, Flink, and Spark are dialects."

Limits of the miniature. Single-process, in-memory, integer event times; no late data, no state backend, no checkpointing, no parallelism. combine_per_key receives the full value list rather than an incremental accumulator (real Beam's CombineFn has create_accumulator/add_input/merge_accumulators precisely so partial aggregates can pre-combine before the shuffle) — the deterministic result is the same, the scalability trick is not shown.

Extensions (your own machine)

• Add session windows: per-key windows that extend while events keep arriving within a gap, and merge when they touch — the windowing type that requires window merging, which fixed and sliding never do.
• Add late-data handling: keep a fired window's state for allowed_lateness ticks and re-fire an updated pane (accumulating mode) when a late element arrives.
• Implement an incremental combiner (create_accumulator/add_input/merge) and show that merging per-partition accumulators equals combining the whole list — the pre-shuffle "combiner lifting" optimization.
• Run the real thing: pip install apache-beam, port build_word_count to Beam's Python SDK with beam.WindowInto(beam.window.FixedWindows(10)), and run it with the DirectRunner.

Interview / resume signal

"Built a miniature Beam/Dataflow engine: PCollections with event-time elements, the Map/Filter/FlatMap/GroupByKey/CombinePerKey transform algebra keyed per (window, key), fixed and sliding window assignment, and a watermark-driven streaming runner that fires each window exactly once when the watermark passes its end — with a test proving batch and streaming produce identical results over the same events, which is the Dataflow model's central claim."