Apache Beam Programming Guide: PCollections, Windowing, Runners (Dataflow / Flink)

apache beam is the write-once-run-anywhere programming model that most senior data engineers meet the day they need one pipeline definition to run on Dataflow in production and on Flink in the on-prem region — and it is the single framework whose "unified batch and streaming" pitch actually survives an interview because the model was designed for it from day one, not retrofitted. A beam pipeline is a directed acyclic graph of beam ptransform operations that accept and emit beam pcollection datasets; the runner — DataflowRunner, FlinkRunner, SparkRunner, or the DirectRunner for local tests — is a swappable execution back-end that reads the same graph and translates it to the target engine's primitives. The engineering trade-off lives in which runner you pick and how you configure windowing + triggers — not in whether the model works.

This guide is the senior-DE walkthrough you wished existed the first time an interviewer asked "explain the difference between a bounded and unbounded PCollection" or "what breaks when you move a Beam pipeline from DataflowRunner to beam flink runner?" or "walk me through a 5-minute tumbling window with 1-minute allowed lateness and a watermark that's 30 seconds behind." It walks through why Beam still matters despite the runner-lock-in worry, the PCollection / PTransform / pipeline atoms, the four window types (fixed, sliding, session, global) with watermarks and lateness, the runner trade-offs across dataflow, Flink, and Spark, beam sql on Calcite, and the production patterns — autoscale flags, checkpoint state, saturation monitoring — that senior engineers ship into every unified batch streaming deployment. Each section pairs a teaching block with a Solution-Tail interview answer — code, a step-by-step trace, an output table, then a concept-by-concept breakdown of why it works.

When you want hands-on reps immediately after reading, drill the streaming practice library →, rehearse on the ETL practice library →, and sharpen the tuning axis with the optimization practice library →.

---

On this page

• Why Beam still matters despite runner lock-in fears
• PCollection + PTransform + pipeline
• Windowing — fixed, sliding, session, global
• Runners — Dataflow / Flink / Spark
• Beam SQL + production patterns
• Cheat sheet — Beam recipes
• Frequently asked questions
• Practice on PipeCode

---

1. Why Beam still matters despite runner lock-in fears

Beam's write-once-run-anywhere promise is still the only unified batch + streaming model that survives an interview

The one-sentence invariant: Apache Beam is a programming model and a set of SDKs (Java, Python, Go) whose graph is executed by a runner — DataflowRunner, FlinkRunner, SparkRunner, or the DirectRunner — so the same 20-line pipeline compiles and runs on any of them, and the model itself makes no distinction between a bounded batch source and an unbounded streaming source. Every other stream processor (Flink, Spark Structured Streaming, Kafka Streams) either bolts a streaming API onto a batch engine or picks a favourite engine and calls it a day. Beam decouples the model from the runtime — that is the entire point, and the reason it still matters despite fifteen years of "batch + streaming unification" narratives from every other vendor.

The four axes interviewers actually probe.

• PCollection. The atom of Beam — an immutable, distributed dataset that is either bounded (finite; batch) or unbounded (infinite; streaming). Every PTransform reads PCollections and emits new ones; the pipeline never mutates a PCollection in place. Naming which properties change between bounded and unbounded is the fastest senior signal.
• PTransform. The operation — ParDo, GroupByKey, Combine, Flatten, Window, Read, Write. PTransforms compose; a pipeline is a DAG of them. The apply method (Java) or | operator (Python) is the wire between two PTransforms. Senior engineers can name the difference between a ParDo (element-wise) and a Combine (associative reduction) without hesitation.
• Windowing. The rule that partitions an unbounded PCollection into finite windows — fixed, sliding, session, or global. Combined with watermarks (the runner's belief about event-time progress) and triggers (when to emit a pane), windowing is the actual streaming semantics. Half of every Beam interview lives here.
• Runner. The execution back-end. DataflowRunner runs on Google Cloud Dataflow with autoscale + streaming engine; FlinkRunner runs on Flink with savepoints + state backends; SparkRunner runs on Spark (batch-first, limited streaming); the DirectRunner runs in-process for tests. The runner-specific quirks are the questions that separate mid-level from senior candidates.

Why "write once, run anywhere" is a real promise, not marketing.

• Portability by design. The Beam graph is a language-neutral protocol-buffer definition. The runner reads that definition and translates it to its native primitives — Dataflow Fn API workers, Flink KeyedProcessFunctions, or Spark RDDs. The SDK does not embed engine-specific concepts.
• Fn API portability layer. The Fn API is the wire protocol between the Beam runner and the SDK workers. The SDK worker can run in a container in a different language from the runner — Python SDK workers alongside a Java-based Flink runner, for example. This is what makes Beam truly polyglot.
• The "runner-independent" contract. A pipeline that compiles under DirectRunner and passes the standard Beam validation tests will run on DataflowRunner or FlinkRunner without source changes. The runtime characteristics (throughput, latency, autoscale behaviour) differ; the semantics do not.
• Where lock-in still creeps in. Runner-specific I/Os (BigQueryIO, PubsubIO, and to some extent KafkaIO on Flink) are the sneak-in points. Advanced state APIs (StateSpec, TimerSpec) work on all major runners but have subtle behaviour differences. Autoscale and streaming-engine features are Dataflow-only. A senior candidate names these boundaries without prompting.

Why interviewers still probe Beam in 2026 despite the "runner wars."

• Unified batch + streaming. Beam is the only major framework where the same pipeline runs bounded and unbounded workloads with identical semantics. Flink has bounded execution mode; Spark has structured streaming — but neither erases the seam. Beam does.
• Portable Python. Beam Python + FlinkRunner is one of the few production-grade paths to Python-native streaming without a bespoke framework. The Fn API + portable containers make it practical.
• Dataflow is a top-3 managed streaming service. GCP shops default to Dataflow; Dataflow speaks Beam. Any GCP-flavoured data engineering role probes Beam knowledge.
• The vocabulary is portable. Even if you never ship a Beam pipeline, the PCollection / PTransform / windowing vocabulary is the industry lingua franca for streaming semantics. Learning Beam pays back on Flink, Dataflow, and even Spark Structured Streaming interviews.

What interviewers listen for.

• Do you say "the runner is swappable; the model is the constant" in the first sentence when asked about Beam's value? — senior signal.
• Do you separate bounded vs unbounded PCollections from batch vs streaming runners? — senior signal.
• Do you push back on "just use Flink directly" with the portability + unified-model argument? — required answer.
• Do you name the Fn API as the mechanism that makes cross-language SDKs work? — senior signal.

Worked example — the same 20-line pipeline on Dataflow and Flink

Detailed explanation. The canonical "why Beam" demo: write a word-count pipeline once, compile it against DirectRunner for a local test, then swap the runner flag to DataflowRunner and FlinkRunner and watch the same code produce the same results on both. The demo is not just cute — it is the concrete proof that the abstraction holds under production load.

• The pipeline. Read a text source, split into words, count occurrences, write to a sink.
• The swap. One CLI flag (--runner=DataflowRunner, --runner=FlinkRunner, --runner=DirectRunner) picks the execution back-end.
• The observable. Output rows are bit-for-bit identical across runners for a bounded input.

Question. Write the 20-line Python pipeline that counts words in a text file, and show how to run it on DirectRunner, DataflowRunner, and FlinkRunner without source changes. Explain what changes at runtime.

Input.

Component	Value
Input source	gs://example/input.txt (or local path)
Sink	gs://example/output (or local path)
SDK	Python 3.11 + apache-beam[gcp]
Runners tested	DirectRunner, DataflowRunner, FlinkRunner

Code.

# wordcount.py — the canonical 20-line Beam pipeline
import argparse
import re
import apache_beam as beam
from apache_beam.options.pipeline_options import PipelineOptions


def run():
    parser = argparse.ArgumentParser()
    parser.add_argument("--input",  required=True)
    parser.add_argument("--output", required=True)
    known, pipeline_args = parser.parse_known_args()
    options = PipelineOptions(pipeline_args)

    with beam.Pipeline(options=options) as p:
        (p
         | "Read"    >> beam.io.ReadFromText(known.input)
         | "Split"   >> beam.FlatMap(lambda line: re.findall(r"\w+", line.lower()))
         | "Pair"    >> beam.Map(lambda w: (w, 1))
         | "Count"   >> beam.CombinePerKey(sum)
         | "Format"  >> beam.MapTuple(lambda word, n: f"{word}\t{n}")
         | "Write"   >> beam.io.WriteToText(known.output))


if __name__ == "__main__":
    run()

# DirectRunner — local, single-process, for tests
python wordcount.py \
    --input=./input.txt \
    --output=./out \
    --runner=DirectRunner

# DataflowRunner — GCP managed
python wordcount.py \
    --input=gs://example/input.txt \
    --output=gs://example/output \
    --runner=DataflowRunner \
    --project=my-gcp-project \
    --region=us-central1 \
    --temp_location=gs://example/tmp \
    --job_name=wc-2026-07-06

# FlinkRunner — portable, against an existing Flink cluster
python wordcount.py \
    --input=hdfs:///example/input.txt \
    --output=hdfs:///example/output \
    --runner=FlinkRunner \
    --flink_master=flink-jobmanager:8081 \
    --environment_type=DOCKER

Step-by-step explanation.

1. The pipeline is defined once in the with beam.Pipeline(options=options) as p: block. The | operator chains PTransforms — ReadFromText → FlatMap → Map → CombinePerKey → MapTuple → WriteToText. This is the DAG the runner will execute.
2. PipelineOptions(pipeline_args) parses the runner-specific flags (--runner, --project, --region, --flink_master, etc.). The pipeline body knows nothing about which runner it will execute on — this is the write-once-run-anywhere contract in action.
3. On DirectRunner, the graph runs in-process. Splittable DoFns, windowing, and coders all work; parallelism is limited to the local CPU count. Perfect for unit tests and CI.
4. On DataflowRunner, the SDK serialises the graph and submits it to the Dataflow control plane. Dataflow autoscales workers up and down based on queue depth; Streaming Engine (separate compute) handles the shuffle. The --temp_location is where staging artefacts live.
5. On FlinkRunner (portable mode), the SDK submits the graph to Flink via the Fn API. --environment_type=DOCKER tells Flink to launch SDK worker containers; the Python code executes inside those containers while Flink handles the runtime. The three runs produce identical output for a bounded input.

Output.

Runner	Where it runs	Autoscale	State backend	Local latency
DirectRunner	in-process	no	in-memory	~ms
DataflowRunner	GCP managed	yes	Dataflow Streaming Engine	~seconds startup
FlinkRunner	Flink cluster	via Flink	RocksDB / filesystem	~seconds startup

Rule of thumb. For any new Beam pipeline, always run the DirectRunner smoke test first (fast feedback), then promote to the target runner via a CLI flag swap. Never hard-code the runner in application code — put it in PipelineOptions.

Worked example — bounded vs unbounded PCollection semantics

Detailed explanation. The pivot that every Beam interview probes: what changes when the input transitions from a finite file to an infinite stream. The SDK types are the same; the runtime semantics diverge sharply. Bounded PCollections terminate when the source is exhausted; unbounded PCollections never terminate and require windowing to produce any output.

• Bounded PCollection. Backed by a source with a known end (a file, a BigQuery table snapshot, a bounded PubSub subscription). Aggregations produce a final result when the source drains.
• Unbounded PCollection. Backed by a streaming source (PubSub, Kafka, unbounded PubSub). Aggregations require windowing to produce panes; without windowing, GroupByKey never emits.

Question. Show the same word-count pipeline body running against a bounded file source and an unbounded PubSub source. Explain why the unbounded version needs a window and the bounded version does not.

Input.

Component	Bounded version	Unbounded version
Source	beam.io.ReadFromText("gs://.../input.txt")	beam.io.ReadFromPubSub("projects/p/subscriptions/s")
Aggregation	beam.CombinePerKey(sum)	beam.CombinePerKey(sum) after WindowInto(FixedWindows(60))
Termination	when file drained	never (streaming)
Sink	text file	BigQuery streaming insert

Code.

# Bounded — no windowing needed
import apache_beam as beam
from apache_beam.transforms.window import FixedWindows

def bounded_pipeline(p, input_path, output_path):
    (p
     | "Read"   >> beam.io.ReadFromText(input_path)
     | "Split"  >> beam.FlatMap(lambda l: l.split())
     | "Pair"   >> beam.Map(lambda w: (w, 1))
     | "Count"  >> beam.CombinePerKey(sum)
     | "Write"  >> beam.io.WriteToText(output_path))

# Unbounded — MUST have a window before the aggregation
def unbounded_pipeline(p, subscription, table):
    (p
     | "Read"   >> beam.io.ReadFromPubSub(subscription=subscription)
     | "Decode" >> beam.Map(lambda b: b.decode("utf-8"))
     | "Split"  >> beam.FlatMap(lambda l: l.split())
     | "Pair"   >> beam.Map(lambda w: (w, 1))
     | "Window" >> beam.WindowInto(FixedWindows(60))          # 60-second fixed windows
     | "Count"  >> beam.CombinePerKey(sum)
     | "Format" >> beam.MapTuple(lambda w, n: {"word": w, "n": n})
     | "Write"  >> beam.io.WriteToBigQuery(
                    table,
                    schema="word:STRING,n:INTEGER",
                    write_disposition=beam.io.BigQueryDisposition.WRITE_APPEND))

Step-by-step explanation.

1. The bounded pipeline reads the entire file, splits into words, groups by key, sums, and writes. Because the source terminates, CombinePerKey knows when it has seen all the elements for a key — it emits the final count and the pipeline ends.
2. The unbounded pipeline reads from PubSub — a source that never terminates. CombinePerKey cannot know when it has seen "all" the elements because "all" is never reached. Without a window, the pipeline would deadlock — no output would ever be produced.
3. beam.WindowInto(FixedWindows(60)) partitions the unbounded stream into 60-second buckets aligned on event time. Each bucket is treated by downstream transforms as a bounded PCollection. CombinePerKey now emits one row per (word, window) pair, exactly like a bounded run.
4. The WriteToBigQuery transform is streaming-aware — it uses the BigQuery streaming insert API for unbounded pipelines and the batch load API for bounded pipelines. The single API surface hides the difference.
5. The lesson: the pipeline body differs by a single WindowInto step; the runtime semantics differ enormously. Every interview question about "batch vs streaming Beam" reduces to this one line of code.

Output.

Property	Bounded	Unbounded
PCollection type	bounded	unbounded
Source terminates?	yes	no
Windowing required?	no (implicit GlobalWindow)	yes
Aggregation emits when?	source drained	window closes
Sink writes	in bulk	incrementally

Rule of thumb. The moment a pipeline reads from a streaming source, add a WindowInto before the first aggregation. Without it, GroupByKey and CombinePerKey will silently buffer forever.

Worked example — DirectRunner unit tests vs runner-specific integration tests

Detailed explanation. Testing a Beam pipeline has two layers: DirectRunner unit tests (fast, run in CI) and runner-specific integration tests (slower, run in a nightly job). Senior engineers wire both. The Beam TestPipeline helper plus assert_that matchers make DirectRunner unit tests almost as ergonomic as pytest.

• DirectRunner unit tests. Sub-second; run in CI on every commit; catch logic bugs, coder errors, windowing mistakes.
• Runner-specific integration tests. Minute-scale; run nightly; catch runner-specific quirks — Dataflow autoscale behaviour, Flink checkpointing, portable-runner container startup.
• The TestPipeline + assert_that idiom. The Beam SDK's built-in test helpers.

Question. Write the DirectRunner unit test for the word-count pipeline body and describe the corresponding integration test on DataflowRunner.

Input.

Test type	Runner	Runtime
Unit	DirectRunner	< 1 s
Integration	DataflowRunner	3–5 min startup + runtime
Framework	pytest + apache_beam.testing	—

Code.

# tests/test_wordcount.py — DirectRunner unit test
import pytest
import apache_beam as beam
from apache_beam.testing.test_pipeline import TestPipeline
from apache_beam.testing.util import assert_that, equal_to
from wordcount import build_pipeline_body   # your pipeline builder


def test_wordcount_counts_correctly():
    with TestPipeline() as p:
        input_data = ["apple banana apple", "banana cherry apple"]
        output = (p
                  | beam.Create(input_data)
                  | build_pipeline_body())
        assert_that(output, equal_to([("apple", 3), ("banana", 2), ("cherry", 1)]))


def test_wordcount_handles_empty_lines():
    with TestPipeline() as p:
        input_data = ["", "   ", "apple"]
        output = (p
                  | beam.Create(input_data)
                  | build_pipeline_body())
        assert_that(output, equal_to([("apple", 1)]))

# Integration test — nightly Dataflow job
# scripts/integration_test.sh
python wordcount.py \
    --input=gs://example-test/input-nightly.txt \
    --output=gs://example-test/output-$(date +%Y%m%d) \
    --runner=DataflowRunner \
    --project=my-test-project \
    --region=us-central1 \
    --temp_location=gs://example-test/tmp \
    --job_name=wc-nightly-$(date +%Y%m%d) \
    --experiments=use_runner_v2 \
    --num_workers=2 \
    --max_num_workers=4

# After the job completes, verify output
gsutil cat gs://example-test/output-$(date +%Y%m%d)/* | \
    python scripts/verify_output.py --expected=tests/fixtures/expected.txt

Step-by-step explanation.

1. TestPipeline() is a Beam context manager that runs the pipeline on the DirectRunner when the with block exits. Inside the block, you build the graph as normal — beam.Create(input_data) produces a bounded PCollection from an in-memory list.
2. assert_that(output, equal_to([...])) schedules a validation transform onto the pipeline. The validation runs after the pipeline finishes; a failure raises an AssertionError.
3. The unit test takes about 200 ms end-to-end. It runs in CI on every commit and catches 90% of logic bugs — miscounted words, missing splits, wrong coders.
4. The integration test invokes the actual DataflowRunner against a fixed input in a test project. --num_workers=2 --max_num_workers=4 bounds the cost; --experiments=use_runner_v2 enables the modern runner-v2 execution mode. The job takes 3–5 minutes.
5. Post-run verification compares the output artefacts against a golden fixture. Any divergence flags a runner-specific regression — the kind of bug that unit tests cannot catch.

Output.

Test layer	Runtime	Catches	Cost
DirectRunner unit	< 1 s	logic bugs, coder errors	~free (CI minutes)
DataflowRunner integration	3–5 min	runner quirks, autoscale, I/O	~$0.10/run

Rule of thumb. Every Beam pipeline ships with both layers. DirectRunner unit tests for CI; runner-specific integration tests nightly. Skipping the integration layer is how "the pipeline works on my laptop, breaks in prod" happens.

Senior interview question on when Beam is the right framework

A senior interviewer often opens with: "We have a Kafka input, a BigQuery sink, and a business rule that we want the same aggregation to run both as a nightly batch backfill and as a real-time streaming pipeline. Walk me through why you'd pick Apache Beam over writing two separate pipelines in Flink and Spark, and what the runner story looks like."

Solution Using Beam with DataflowRunner in prod and DirectRunner for local dev

# unified_agg.py — one pipeline body, two execution modes
import argparse
import apache_beam as beam
from apache_beam.options.pipeline_options import PipelineOptions
from apache_beam.transforms.window import FixedWindows


def build_pipeline(p, opts):
    if opts.mode == "streaming":
        source = beam.io.ReadFromPubSub(topic=opts.pubsub_topic).with_output_types(bytes)
    else:  # batch backfill
        source = beam.io.ReadFromBigQuery(query=opts.backfill_query,
                                          use_standard_sql=True)

    (p
     | "Source" >> source
     | "Parse"  >> beam.Map(parse_event)
     | "Key"    >> beam.Map(lambda e: (e["tenant_id"], e["amount"]))
     | "Window" >> beam.WindowInto(FixedWindows(300))       # 5-min windows
     | "Sum"    >> beam.CombinePerKey(sum)
     | "Format" >> beam.MapTuple(lambda k, v: {"tenant_id": k, "total": v})
     | "Sink"   >> beam.io.WriteToBigQuery(
                    opts.output_table,
                    schema="tenant_id:STRING,total:FLOAT",
                    write_disposition=beam.io.BigQueryDisposition.WRITE_APPEND))


def run():
    parser = argparse.ArgumentParser()
    parser.add_argument("--mode", choices=["batch", "streaming"], required=True)
    parser.add_argument("--pubsub_topic")
    parser.add_argument("--backfill_query")
    parser.add_argument("--output_table", required=True)
    known, pipeline_args = parser.parse_known_args()
    options = PipelineOptions(pipeline_args, streaming=(known.mode == "streaming"))

    with beam.Pipeline(options=options) as p:
        build_pipeline(p, known)


if __name__ == "__main__":
    run()

# Prod streaming — DataflowRunner
python unified_agg.py \
    --mode=streaming \
    --pubsub_topic=projects/p/topics/events \
    --output_table=p:analytics.tenant_totals \
    --runner=DataflowRunner \
    --project=p \
    --region=us-central1 \
    --temp_location=gs://p-dataflow/tmp \
    --num_workers=2 \
    --max_num_workers=20 \
    --autoscaling_algorithm=THROUGHPUT_BASED \
    --enable_streaming_engine

# Nightly backfill — DataflowRunner, batch mode
python unified_agg.py \
    --mode=batch \
    --backfill_query='SELECT * FROM `p.raw.events` WHERE DATE(ts) = "2026-07-05"' \
    --output_table=p:analytics.tenant_totals \
    --runner=DataflowRunner \
    --project=p \
    --region=us-central1 \
    --temp_location=gs://p-dataflow/tmp

Step-by-step trace.

Step	Streaming path	Batch backfill path
Source	PubSub subscription	BigQuery query
Bounded?	unbounded	bounded
Windowing	5-min fixed windows	5-min fixed windows
Aggregation	CombinePerKey(sum)	CombinePerKey(sum)
Sink	BigQuery streaming insert	BigQuery batch load
Runner	DataflowRunner + Streaming Engine	DataflowRunner (batch)
Autoscale	THROUGHPUT_BASED, 2–20 workers	dynamic work rebalancing

The single pipeline body handles both modes. In streaming mode, Dataflow provisions long-running workers, wires Streaming Engine, and emits per-window results as they close. In batch mode, Dataflow runs a bounded job to completion and shuts down. The output table receives the same aggregation semantics in both modes — the backfill is byte-for-byte compatible with the streaming output for the same time range.

Output:

Metric	Streaming mode	Batch backfill
Job duration	continuous	~10 min per day
Workers	2–20 (autoscale)	dynamic
Result freshness	~1 min lag	end-of-day
Code paths	one pipeline body	one pipeline body
Test surface	DirectRunner + Dataflow integration	DirectRunner + Dataflow integration

Why this works — concept by concept:

• One pipeline body — the aggregation logic (Parse → Key → Window → Sum → Format → Sink) is identical in both modes. Only the source toggles. The write-once-run-anywhere contract eliminates duplicate code and its associated drift.
• Windowing as the streaming primitive — the 5-minute fixed window is the natural unit for both modes. In streaming, panes fire as time advances; in batch, panes are computed once at the end. The window definition never changes.
• DataflowRunner autoscale — THROUGHPUT_BASED scaling reacts to queue depth; workers grow and shrink between 2 and 20 based on actual load. Streaming Engine offloads shuffle from workers, so autoscale is smoother and startup is faster.
• Same sink, two write paths — WriteToBigQuery internally picks streaming insert vs batch load based on the pipeline's bounded/unbounded property. The application never knows which path was used.
• Cost — Dataflow streaming Engine adds ~15% cost over classic Dataflow but pays back with faster autoscale and better shuffle performance. Batch backfill runs are pay-per-vCPU-hour; the pipeline shuts down when the query drains, so cost is bounded by data volume, not window count.

Streaming
Topic — streaming
Streaming problems on unified batch and streaming pipelines

Practice →

ETL Topic — etl ETL problems on Dataflow / Flink backfill patterns

Practice →

---

2. PCollection + PTransform + pipeline

PCollections are immutable, PTransforms produce new ones, and the pipeline is the DAG — these three atoms are the entire Beam model

The mental model in one line: a beam pcollection is an immutable, distributed dataset (bounded or unbounded); a beam ptransform is an operation that consumes one or more PCollections and emits new ones; and a pipeline is the directed acyclic graph of PTransforms — every Beam program is a graph of these three atoms, no matter which SDK or runner you pick. The rest of Beam — windowing, triggers, state, timers, side inputs — is a set of features layered on top of these three primitives.

PCollection — the atom.

• Immutability. Once created, a PCollection cannot be modified. Every "transformation" produces a new PCollection. This eliminates a whole class of aliasing bugs and lets the runner freely reorder, retry, and parallelise operations.
• Distributed. The elements of a PCollection are partitioned across worker nodes. The SDK provides no direct index-based access; you can only apply PTransforms to it.
• Bounded vs unbounded. A bounded PCollection has a known finite size (a file, a table snapshot). An unbounded PCollection is infinite (a Kafka topic, a PubSub subscription). Every source constructor produces one or the other.
• Coder. Every PCollection has a coder — a serialisation contract for its element type. The SDK infers coders for common types (str, int, bytes, tuple) and you provide custom coders for domain types. Coder mismatches are the second-most-common Beam error.

PTransform — the operation.

• ParDo. Element-wise transformation. The Java DoFn or Python DoFn subclass defines process(element) that emits zero or more output elements. The most general PTransform.
• Map / FlatMap. Sugar for ParDo when the transformation is a simple function. Map emits exactly one output per input; FlatMap emits zero or more.
• GroupByKey. Group elements of a KV<K, V> PCollection by key. Requires a windowing context if the PCollection is unbounded.
• Combine / CombinePerKey. Reduce elements to a summary using an associative + commutative combining function. More efficient than GroupByKey + reduce because the runner can do partial combining at map side.
• Flatten. Union multiple PCollections of the same element type into one.
• Window. Assign each element to one or more windows based on event time.
• Read / Write. I/O transforms — the source and sink of every pipeline.

Pipeline — the DAG.

• DAG structure. Every pipeline is a directed acyclic graph of PTransforms. Multiple output PCollections can be consumed by multiple downstream PTransforms; the same input PCollection can be fanned out. Cycles are not allowed.
• Deferred execution. Building the pipeline (p | "Read" >> beam.io.ReadFromText(...)) does not execute it. The runner reads the completed graph and plans execution.
• run() semantics. pipeline.run() submits the graph to the runner. For DirectRunner, execution happens in-process; for DataflowRunner, the graph is uploaded and executed on the cloud.
• with Pipeline() as p: idiom. The context manager calls run() on exit and blocks until the pipeline terminates (bounded) or is cancelled (unbounded).

I/O transforms — where PCollections come from and go.

• BigQueryIO. Read from and write to BigQuery. Supports both batch load and streaming insert. Handles schemas, dead-letter routing, and partition-aware writes.
• KafkaIO. Read from and write to Kafka. Java-native; Python support via the portable Fn API. Handles offset checkpointing and exactly-once semantics on supported runners.
• PubsubIO. Read from and write to Google PubSub. The default source for streaming pipelines on Dataflow.
• GcsIO / FileIO. Read from and write to filesystems (GCS, S3, HDFS, local). The default source for batch pipelines.
• JdbcIO. Read from and write to any JDBC-compatible database. Used for backfill from legacy Postgres / MySQL.

Common interview probes on PCollection + PTransform.

• "What's the difference between ParDo and Combine?" — element-wise vs associative reduction with partial combining.
• "Why is a PCollection immutable?" — safety under retry and reordering; no aliasing bugs.
• "How do you union two PCollections?" — Flatten, provided they share an element type.
• "What's a coder and why does it matter?" — serialisation contract; wrong coder = runtime error at shuffle boundary.

Worked example — the 20-line word-count in the Python SDK

Detailed explanation. The Beam "Hello, World" — a word-count pipeline in 20 lines of Python. The exercise is not the counting itself but the graph shape: five composed PTransforms, one output PCollection at each step, one sink. Every senior candidate should be able to write this from memory.

• The shape. Read → Split → Pair → Count → Write — five PTransforms in series.
• The primitives. ReadFromText, FlatMap, Map, CombinePerKey, WriteToText.
• The output. One line per unique word: "apple\t3".

Question. Write the word-count pipeline in the Python SDK, then annotate every PTransform with the type of PCollection it produces.

Input.

Component	Value
Input file	input.txt — plain text
Output file	out.txt — tab-separated
SDK	apache-beam[gcp] 2.55+
Runner	DirectRunner

Code.

# wordcount_annotated.py
import re
import apache_beam as beam

with beam.Pipeline() as p:
    lines    = (p
                | "Read" >> beam.io.ReadFromText("input.txt"))
                # PCollection[str]  — one element per line, bounded

    words    = (lines
                | "Split" >> beam.FlatMap(lambda l: re.findall(r"\w+", l.lower())))
                # PCollection[str]  — one element per word, bounded

    pairs    = (words
                | "Pair" >> beam.Map(lambda w: (w, 1)))
                # PCollection[Tuple[str, int]]

    counts   = (pairs
                | "Count" >> beam.CombinePerKey(sum))
                # PCollection[Tuple[str, int]] — aggregated, one element per unique word

    formatted = (counts
                 | "Format" >> beam.MapTuple(lambda w, n: f"{w}\t{n}"))
                # PCollection[str] — one formatted line per unique word

    (formatted
     | "Write" >> beam.io.WriteToText("out"))
                # sink; returns PCollection[None]

Step-by-step explanation.

1. beam.io.ReadFromText("input.txt") produces PCollection[str] — one element per line. The PCollection is bounded because the file has a known end. The runner reads the file in parallel and distributes lines across workers.
2. beam.FlatMap(lambda l: re.findall(r"\w+", l.lower())) splits each line into zero or more words. FlatMap differs from Map in that the function returns an iterable — each word becomes a separate element in the output PCollection.
3. beam.Map(lambda w: (w, 1)) converts each word to a (word, 1) tuple. The KV-shaped PCollection is the input contract for any per-key aggregation.
4. beam.CombinePerKey(sum) groups by key and applies the associative combiner sum. Because sum is commutative and associative, the runner can partially combine at the map side, then finalise at the reduce side — the combiner is the same function in both phases.
5. beam.MapTuple(lambda w, n: f"{w}\t{n}") formats each (word, count) tuple into a tab-separated string. beam.io.WriteToText("out") writes the formatted PCollection to the sink, producing one output file per worker shard.

Output.

Line in out-00000-of-00001	Content
1	apple\t3
2	banana\t2
3	cherry\t1

Rule of thumb. The five-step pattern (Read → transform → key → aggregate → Write) is the shape of 80% of Beam pipelines. Learn to write it from memory; the interviewer will ask.

Worked example — ParDo vs Combine for the same aggregation

Detailed explanation. Both ParDo (in a GroupByKey-then-reduce shape) and CombinePerKey produce the same output for an aggregation — but the runtime characteristics differ enormously. CombinePerKey is combining aware, letting the runner do partial reduction at the map side and reduce shuffle volume by orders of magnitude. Interviewers probe this constantly because it separates the "I know Beam syntax" candidates from the "I understand Beam performance" candidates.

• GroupByKey + ParDo. Group all values for a key, then reduce. Every value crosses the shuffle boundary.
• CombinePerKey. Reduce partially at the map side, then finalise at the reduce side. Only the partial results cross the shuffle boundary.
• The saving. For a sum over billions of elements, CombinePerKey can reduce shuffle volume by 100–1000×.

Question. Rewrite a word-count aggregation using both GroupByKey + ParDo and CombinePerKey. Explain which one you'd ship to production and why.

Input.

Component	Value
Input	PCollection[Tuple[str, int]] — (word, 1) tuples
Aggregation	sum
Cardinality	~10M elements, ~100K unique keys

Code.

# Version A — GroupByKey + ParDo (naive)
import apache_beam as beam

def gbk_sum():
    return (
        beam.GroupByKey()
        | beam.MapTuple(lambda k, vs: (k, sum(vs)))
    )

# Version B — CombinePerKey (idiomatic)
def combine_sum():
    return beam.CombinePerKey(sum)

# Usage
with beam.Pipeline() as p:
    words = (p
             | beam.Create([("apple", 1), ("banana", 1), ("apple", 1), ("apple", 1)]))

    (words | "A" >> gbk_sum()      | "PrintA" >> beam.Map(print))
    (words | "B" >> combine_sum()  | "PrintB" >> beam.Map(print))

Step-by-step explanation.

1. Version A uses GroupByKey to gather every value for each key on one worker, then applies a MapTuple that runs sum(vs) on the local list. This is semantically correct but expensive — every input value crosses the network to reach the reducer.
2. Version B uses CombinePerKey(sum) — the runner recognises that sum is associative and commutative and can therefore be applied incrementally. Each worker sums its local apple values (say, 3M partial sums into 300 partial totals), and only the partial totals are shuffled.
3. The performance gap: for 10M elements with 100K unique keys, Version A shuffles all 10M elements. Version B shuffles at most num_workers × num_keys partial totals — typically <1M elements. The 10× reduction in shuffle volume is the difference between "job takes 30 minutes" and "job takes 3 minutes."
4. CombinePerKey also plays better with autoscale: partial combining reduces the per-key working set, so worker RAM stays bounded even when a single key has billions of values (the "hot key" problem is dampened, though not eliminated).
5. The rule: whenever the reduce function is associative + commutative, use CombinePerKey. Use GroupByKey + ParDo only when the reduce requires the full list (e.g. median, top-K by custom comparator).

Output.

Version	Shuffle volume	Runtime (10M elements)	Hot-key resilience
A — GroupByKey + ParDo	10M elements	~30 min	poor (all values on one worker)
B — CombinePerKey	~100K partial totals	~3 min	good (partial combining)

Rule of thumb. CombinePerKey for associative reductions (sum, count, min, max, average via CombineFn). GroupByKey only when you genuinely need the full grouped list. Ship CombinePerKey; the shuffle savings pay back on day one.

Worked example — side inputs and multiple outputs

Detailed explanation. Not every pipeline is a linear chain. Real pipelines join a main input against a small lookup table (side input) and emit multiple output types (multiple outputs). Both patterns are first-class in Beam and unlock most non-trivial transformations.

• Side input. A PCollection made available to every element of another PCollection as a read-only lookup. Typically small — a config table, a currency conversion table.
• Multiple outputs. A DoFn that emits into multiple named output PCollections — e.g. a validation DoFn that emits valid records into "good" and invalid records into "bad" for dead-letter routing.
• Interview probe. "How would you dead-letter invalid records without dropping them?" — multiple outputs from a DoFn.

Question. Enrich a stream of events with a small country-code → currency lookup table, and route invalid events (missing country code) into a dead-letter output.

Input.

PCollection	Elements
events	[{"id": 1, "amount": 100, "country": "US"}, {"id": 2, "amount": 200, "country": "XX"}]
currency_lookup	{"US": "USD", "GB": "GBP", "EU": "EUR"}

Code.

import apache_beam as beam
from apache_beam.pvalue import AsDict, TaggedOutput

class EnrichWithCurrency(beam.DoFn):
    def process(self, event, lookup):
        country = event.get("country")
        if country in lookup:
            yield {**event, "currency": lookup[country]}
        else:
            yield TaggedOutput("dead_letter", {**event, "reason": "unknown_country"})

with beam.Pipeline() as p:
    events = (p | "Events" >> beam.Create([
        {"id": 1, "amount": 100, "country": "US"},
        {"id": 2, "amount": 200, "country": "XX"},
        {"id": 3, "amount": 300, "country": "GB"},
    ]))

    lookup = (p | "Lookup" >> beam.Create([("US", "USD"), ("GB", "GBP"), ("EU", "EUR")]))

    outputs = (events
               | "Enrich" >> beam.ParDo(EnrichWithCurrency(), lookup=AsDict(lookup))
                              .with_outputs("dead_letter", main="enriched"))

    (outputs.enriched   | "PrintOK"   >> beam.Map(lambda e: print("OK",   e)))
    (outputs.dead_letter | "PrintBad" >> beam.Map(lambda e: print("BAD",  e)))

Step-by-step explanation.

1. The EnrichWithCurrency DoFn takes the main input event and a lookup side input passed as AsDict(lookup). AsDict broadcasts the lookup PCollection as a Python dict available to every worker.
2. If the event's country code is in the lookup, the DoFn yields the enriched event into the main output. If not, it yields a TaggedOutput("dead_letter", ...) — the tagged output goes to a named secondary output PCollection.
3. .with_outputs("dead_letter", main="enriched") tells Beam that this DoFn has two outputs: dead_letter and a main output named enriched. The result is a container with .enriched and .dead_letter PCollection attributes.
4. The downstream pipeline can route each output independently: enriched events go to the analytics warehouse; dead-letter events go to a monitoring topic and a retry queue.
5. The side input is broadcast to every worker; the runner handles the fan-out. For large side inputs (>100 MB), consider AsSingleton on a lightweight summary or a stateful DoFn — broadcasting a 10 GB side input to 100 workers is 1 TB of network traffic.

Output.

Output	Content
enriched	{"id": 1, "amount": 100, "country": "US", "currency": "USD"}
enriched	{"id": 3, "amount": 300, "country": "GB", "currency": "GBP"}
dead_letter	{"id": 2, "amount": 200, "country": "XX", "reason": "unknown_country"}

Rule of thumb. Side inputs for small lookups (< 100 MB). Multiple outputs (TaggedOutput) for dead-letter routing. Never drop invalid records silently — always tag them into a dead-letter output that a monitoring job can process.

Senior interview question on composing PCollection + PTransform patterns

A senior interviewer might ask: "Design a pipeline that reads clickstream events from PubSub, enriches each event with a small user-profile lookup, filters out bots, and writes valid records to BigQuery and invalid records to a dead-letter GCS bucket. Walk me through the PCollection graph and the PTransforms you'd compose."

Solution Using ReadFromPubSub + side input + tagged outputs + WriteToBigQuery + WriteToGCS

# clickstream.py — production-shaped Beam pipeline
import apache_beam as beam
from apache_beam.pvalue import AsDict, TaggedOutput
from apache_beam.options.pipeline_options import PipelineOptions
from apache_beam.transforms.window import FixedWindows


class EnrichAndValidate(beam.DoFn):
    def process(self, event, user_profiles):
        user_id = event.get("user_id")
        if not user_id or event.get("is_bot"):
            yield TaggedOutput("bad", {**event, "reason": "bot_or_missing_uid"})
            return

        profile = user_profiles.get(user_id)
        if profile is None:
            yield TaggedOutput("bad", {**event, "reason": "unknown_user"})
            return

        yield {**event, "user_country": profile["country"], "user_plan": profile["plan"]}


def build(p, opts):
    events = (p | "Read" >> beam.io.ReadFromPubSub(
                    subscription=opts.subscription).with_output_types(bytes)
                | "Decode" >> beam.Map(lambda b: __import__("json").loads(b.decode())))

    profiles = (p | "Profiles" >> beam.io.ReadFromBigQuery(
                    table=opts.profiles_table)
                  | "KV" >> beam.Map(lambda r: (r["user_id"],
                                                 {"country": r["country"], "plan": r["plan"]})))

    outputs = (events
               | "Window" >> beam.WindowInto(FixedWindows(60))
               | "Enrich" >> beam.ParDo(EnrichAndValidate(), user_profiles=AsDict(profiles))
                              .with_outputs("bad", main="good"))

    (outputs.good
     | "ToBQ" >> beam.io.WriteToBigQuery(
                    opts.output_table,
                    schema=("event_id:STRING,user_id:STRING,user_country:STRING,"
                            "user_plan:STRING,ts:TIMESTAMP"),
                    write_disposition=beam.io.BigQueryDisposition.WRITE_APPEND))

    (outputs.bad
     | "ToJSON" >> beam.Map(lambda r: __import__("json").dumps(r))
     | "ToGCS"  >> beam.io.WriteToText(opts.dead_letter_prefix,
                                        file_name_suffix=".jsonl.gz",
                                        compression_type=beam.io.filesystems.CompressionTypes.GZIP))

Step-by-step trace.

Stage	PCollection type	Bounded?	Transforms
Read PubSub	PCollection[bytes]	unbounded	ReadFromPubSub, Decode
Read profiles	PCollection[Tuple[str, dict]]	bounded (BQ snapshot)	ReadFromBigQuery, Map
Window events	PCollection[dict]	unbounded, windowed	WindowInto(FixedWindows(60))
Enrich + validate	main: PCollection[dict] / bad: PCollection[dict]	unbounded	ParDo + .with_outputs
Write BQ	sink	—	WriteToBigQuery (streaming insert)
Write GCS	sink	—	WriteToText (windowed file per 60s)

The graph shows two independent branches merging into the enrichment stage, then splitting again into two sinks. The side input (profiles) is a small lookup broadcast to every worker; the main input (clickstream) is a high-volume unbounded stream. Dead-letter routing is inside the DoFn — no invalid record is silently dropped.

Output:

Sink	Format	Frequency
BigQuery streaming	rows	continuous
GCS dead-letter	gzipped jsonl, one file per 60s window	per window
DirectRunner test result	assertions pass	on CI

Why this works — concept by concept:

• PCollection immutability — the main stream is never mutated; each stage produces a new PCollection. Retries under transient errors are safe because the source stream state (PubSub subscription cursor) is the only thing that advances.
• Side input broadcast — AsDict(profiles) materialises the small lookup on every worker. For a 100 MB profile table and 20 workers, this is 2 GB of broadcast traffic — acceptable. For a 10 GB table, you'd swap to a stateful DoFn or a lookup service.
• Tagged outputs for dead-letter — the TaggedOutput("bad", ...) pattern keeps the invalid records in the pipeline so downstream monitoring can count and retry them. Silent drops are the antipattern.
• Windowing on the enrichment side — the 60-second fixed window is applied before enrichment so the dead-letter files are naturally rotated per window, and the BigQuery streaming insert is bounded by the same window.
• Cost — DataflowRunner autoscales workers based on PubSub queue depth; the streaming insert cost is O(events); the dead-letter GCS cost is O(bad events × log(bad events)) due to file overhead. Total cost is dominated by BigQuery streaming insert (~$0.05 per GB) and PubSub egress.

Streaming
Topic — streaming
Streaming problems on PCollection composition and DoFn design

Practice →

ETL Topic — etl ETL problems on dead-letter routing and side-input enrichment

Practice →

---

3. Windowing — fixed, sliding, session, global

Four window types, three time domains, one watermark — mastering these three axes is the entire Beam streaming semantics

The mental model in one line: beam windowing partitions an unbounded PCollection into finite windows by event time (the four canonical window types are fixed, sliding, session, and global); the watermark is the runner's belief about how far event time has advanced; and triggers decide when to emit a pane — combining these three axes correctly is the entire streaming semantics story. Every windowing bug is a mismatch between one of these three axes and the interviewer's expectations.

The four canonical window types.

• Fixed (tumbling) windows. Non-overlapping, equal-length windows aligned on epoch — e.g. [00:00, 00:05), [00:05, 00:10), .... Every element belongs to exactly one window. The default for "give me the count per 5 minutes."
• Sliding (hopping) windows. Overlapping windows with a fixed length and a slide period — e.g. 5-minute windows sliding every 1 minute produces [00:00, 00:05), [00:01, 00:06), [00:02, 00:07), .... Every element belongs to multiple windows. Used for smoothed moving averages.
• Session windows. Variable-length windows separated by gaps of inactivity — e.g. all events from a user with no gap larger than 30 minutes form one session. Every element belongs to exactly one session. Used for user-session analysis.
• Global window. One window that spans all time. The default for bounded PCollections. For unbounded PCollections, requires a custom trigger to fire panes.

The three time domains.

• Event time. The timestamp intrinsic to the event — when the click happened, when the sensor read, when the order was placed. This is what humans mean when they say "time."
• Processing time. Wall-clock time on the worker when the element is processed. Diverges from event time due to network delay, batching, and reprocessing.
• Ingest time. The timestamp assigned by the source when it received the event. A middle ground — deterministic but often close to processing time.

The watermark — the runner's belief about event-time progress.

• Definition. The watermark at time t is the runner's assertion that "no more elements with event time ≤ t will arrive." A watermark of 10:00:00 means the runner believes 09:59:59 and earlier are drained.
• Estimation. For PubSub with reliable timestamps, the watermark tracks the oldest unacknowledged message. For Kafka, KafkaIO uses a heuristic based on max seen event time minus a lag budget.
• Consequences. A window fires (by default) when the watermark passes the window's end. Late data arrives after the watermark; the pipeline must decide whether to drop it, discard the previous output and re-emit, or accumulate.

Allowed lateness — the tolerance for late data.

• .with_allowed_lateness(Duration.standardMinutes(1)). Extends the window's "open" duration by 1 minute past the watermark. Elements arriving within this budget re-fire the window.
• .accumulating_fired_panes(). Each pane emits the accumulated total (subsequent panes are supersets of earlier ones).
• .discarding_fired_panes(). Each pane emits only the new elements since the last pane (subsequent panes are deltas).
• .with_late_data(DROP). Elements later than the allowed lateness are dropped (default).

Triggers — when to emit a pane.

• Default trigger. AfterWatermark() — fires once when the watermark passes the window's end. The "wait for completeness" default.
• Early triggers. AfterProcessingTime(1 min), AfterCount(1000) — fire panes early, before the watermark passes. Used when latency matters more than completeness.
• Late triggers. Fire additional panes for late data after the watermark has passed but within allowed lateness.
• Composite triggers. AfterWatermark(early=..., late=...) combines early, on-time, and late triggers into one specification.

Common interview probes on windowing.

• "What's the difference between fixed and sliding windows?" — non-overlapping vs overlapping; one window per element vs many.
• "How does a session window decide its end?" — when the gap between consecutive elements exceeds the session gap.
• "What is the watermark and what happens when it passes a window's end?" — runner's belief about event-time completeness; window fires (by default).
• "What is allowed lateness?" — the tolerance for late data before the window is retired.

Worked example — 5-minute tumbling window with 1-minute allowed lateness

Detailed explanation. The canonical interview question — count events per 5-minute window, with 1 minute of allowed lateness. Walk the interviewer through the trigger, the pane firings, and the accumulating-vs-discarding semantics.

• The window. 5-minute fixed windows aligned on epoch — [00:00, 00:05), [00:05, 00:10), ....
• The trigger. Default AfterWatermark() — fires once when the watermark reaches the window's end.
• The lateness. 1 minute — late-arriving data within 1 minute past the watermark re-fires the window.

Question. Configure a Beam pipeline that counts events per 5-minute window with 1-minute allowed lateness and accumulating fired panes. Trace what happens for an event that arrives at event time 10:03 when the watermark is at 10:07.

Input.

Parameter	Value
Window type	Fixed
Window size	5 minutes
Allowed lateness	1 minute
Accumulation	accumulating
Sink	BigQuery

Code.

import apache_beam as beam
from apache_beam.transforms.window import FixedWindows
from apache_beam.transforms.trigger import AfterWatermark, AccumulationMode
from apache_beam.utils.timestamp import Duration

with beam.Pipeline() as p:
    (p
     | "Read"   >> beam.io.ReadFromPubSub(subscription="projects/p/subscriptions/s")
     | "Parse"  >> beam.Map(parse_event)
     | "Key"    >> beam.Map(lambda e: (e["tenant"], 1))
     | "Window" >> beam.WindowInto(
                     FixedWindows(5 * 60),                        # 5 minutes
                     trigger=AfterWatermark(),                    # fire when watermark passes
                     allowed_lateness=Duration(seconds=60),        # 1 minute of lateness
                     accumulation_mode=AccumulationMode.ACCUMULATING)
     | "Count"  >> beam.CombinePerKey(sum)
     | "Format" >> beam.MapTuple(lambda k, v: {"tenant": k, "count": v})
     | "Write"  >> beam.io.WriteToBigQuery("p:analytics.tenant_counts",
                                            schema="tenant:STRING,count:INTEGER"))

Step-by-step explanation.

1. FixedWindows(5 * 60) creates non-overlapping 5-minute windows aligned on epoch. An event with event time 10:03 falls into window [10:00, 10:05).
2. AfterWatermark() is the default trigger — the window fires when the watermark passes the window's end (10:05 in our example). Because the watermark lags event time (network delay, source lag), the pane typically fires shortly after 10:05.
3. allowed_lateness=Duration(seconds=60) extends the window's "open" duration by 1 minute past the watermark. If a late event arrives at event time 10:03 while the watermark is at 10:07, the event is within allowed lateness (10:07 - 10:05 = 2 min? No — wait: watermark 10:07 means we're only 2 min past the window end; allowed lateness = 1 min; so 10:07 is 2 min > 1 min past 10:05 → beyond lateness).
4. Let's redo the trace properly. Window [10:00, 10:05) closes at watermark = 10:05. Allowed lateness = 1 minute, so late events accepted until watermark = 10:06. An event arriving with event time 10:03 when watermark is at 10:07 is 1 minute past the lateness budget — dropped. If the same event had arrived when watermark was at 10:05:30, it would have been accepted and re-fired the window.
5. AccumulationMode.ACCUMULATING means each pane emits the accumulated total (initial pane + late pane = final total). If instead we chose DISCARDING, each late-pane fire would emit only the new elements since the last fire — the downstream sink would have to sum them itself.

Output.

Event time	Watermark at ingest	Pane fired?	Value emitted
10:03	10:05	initial	count = 47
10:03	10:05:30 (late but within 1 min)	late (accumulating)	count = 48
10:03	10:07 (beyond lateness)	dropped	none
10:04	10:05:59	late	count = 49

Rule of thumb. For most analytics workloads, 5-minute fixed windows with 1–5 minute allowed lateness and accumulating mode is the safe default. Late data within the budget re-fires the window; downstream sees the final total after the lateness horizon.

Worked example — sliding windows for a 5-minute moving average

Detailed explanation. Sliding windows are the tool for moving averages and rolling metrics. Every event belongs to multiple windows because the windows overlap. Interview probes here are usually "why do sliding windows use more compute than fixed?" and "when would you pick sliding over fixed?"

• The shape. 5-minute windows sliding every 1 minute — window k covers [k min, k min + 5 min).
• The overlap. Every event belongs to 5 sliding windows (one per minute in the last 5 minutes).
• The cost. State size and compute are 5× the fixed-window equivalent.

Question. Configure a sliding-window pipeline that computes a 5-minute moving average of request latency, emitted every 1 minute. Explain the state cost.

Input.

Parameter	Value
Window size	5 minutes
Slide period	1 minute
Metric	mean latency in ms
Sink	Prometheus pushgateway

Code.

import apache_beam as beam
from apache_beam.transforms.window import SlidingWindows
from apache_beam.transforms.trigger import AfterWatermark

class MeanFn(beam.CombineFn):
    def create_accumulator(self):        return (0.0, 0)
    def add_input(self, acc, x):          return (acc[0] + x, acc[1] + 1)
    def merge_accumulators(self, accs):
        s, n = 0.0, 0
        for a in accs:
            s += a[0]; n += a[1]
        return (s, n)
    def extract_output(self, acc):
        return acc[0] / acc[1] if acc[1] else 0.0

with beam.Pipeline() as p:
    (p
     | "Read"   >> beam.io.ReadFromPubSub(subscription="projects/p/subscriptions/latencies")
     | "Parse"  >> beam.Map(lambda b: (b.decode(), float(parse_latency(b))))
     | "Window" >> beam.WindowInto(
                     SlidingWindows(size=5 * 60, period=60),      # 5-min windows, sliding every 1 min
                     trigger=AfterWatermark())
     | "MovingAvg" >> beam.CombineGlobally(MeanFn()).without_defaults()
     | "Push"   >> beam.Map(push_to_pushgateway))

Step-by-step explanation.

1. SlidingWindows(size=5*60, period=60) creates 5-minute windows that slide every 60 seconds. Each event with event time t belongs to 5 overlapping windows: [t-5:t), [t-4:t+1), ..., [t:t+5) (rounded to slide boundaries).
2. The state cost: for each unique key, the runner keeps state for every open window. With 5-min windows sliding every 1 min, up to 5 windows per key are open at any moment. State size is 5× the fixed-window baseline.
3. The compute cost: every incoming element updates state for 5 windows instead of 1. For high-throughput pipelines, this can be the dominant cost. Choose the slide period carefully — 30-second slides double the cost again.
4. CombineGlobally(MeanFn()).without_defaults() uses a custom CombineFn that maintains a (sum, count) accumulator and extracts the mean at emit time. The .without_defaults() avoids the default "emit zero on empty window" behaviour.
5. The moving average is emitted every 1 minute (as each new window closes), with a value that summarises the last 5 minutes. Downstream Prometheus sees smooth time-series data with 1-minute granularity — the sliding-window sweet spot.

Output.

Emit time	Window	Mean latency (ms)
10:05	[10:00, 10:05)	42.1
10:06	[10:01, 10:06)	43.5
10:07	[10:02, 10:07)	44.2
10:08	[10:03, 10:08)	41.8

Rule of thumb. Sliding windows for moving-average style metrics where the smoothing matters. Fixed windows for per-window aggregates where the discreteness matters. If both cost and smoothness matter, consider a fixed window with a shorter period and downstream smoothing — cheaper than a wide sliding window.

Worked example — session windows for user-behaviour analysis

Detailed explanation. Session windows are the natural fit for "sessions" of related activity separated by gaps of inactivity. Every element belongs to exactly one session; the session's boundaries are dynamic — determined by the elements themselves. Interviewers love this because it's the one window type that cannot be pre-computed at ingestion.

• The shape. Session windows are variable-length; each session starts with an element and grows until a gap larger than gap_duration occurs.
• The key. Sessions are per-key — user A's session and user B's session are independent.
• The merge. When a new element arrives that would bridge two open sessions, the runner merges them into one.

Question. Compute per-user session durations with a 30-minute gap threshold, and emit (user_id, session_start, session_end, event_count) for each closed session.

Input.

Parameter	Value
Gap duration	30 minutes
Key	user_id
Event	click, pageview, etc.
Sink	BigQuery

Code.

import apache_beam as beam
from apache_beam.transforms.window import Sessions
from apache_beam.transforms.trigger import AfterWatermark

class SessionSummary(beam.CombineFn):
    def create_accumulator(self):
        return {"start": None, "end": None, "count": 0}
    def add_input(self, acc, event):
        t = event["ts"]
        return {"start": min(acc["start"], t) if acc["start"] else t,
                "end":   max(acc["end"],   t) if acc["end"]   else t,
                "count": acc["count"] + 1}
    def merge_accumulators(self, accs):
        out = {"start": None, "end": None, "count": 0}
        for a in accs:
            out = self.add_input(out, {"ts": a["start"]})
            out = self.add_input(out, {"ts": a["end"]})
            out["count"] += a["count"] - 2
        return out
    def extract_output(self, acc):
        return acc

with beam.Pipeline() as p:
    (p
     | "Read"   >> beam.io.ReadFromPubSub(subscription="projects/p/subscriptions/clicks")
     | "Parse"  >> beam.Map(parse_click)                      # emit {"user_id":..., "ts":...}
     | "Key"    >> beam.Map(lambda e: (e["user_id"], e))
     | "Window" >> beam.WindowInto(Sessions(gap_size=30 * 60), trigger=AfterWatermark())
     | "Summary" >> beam.CombinePerKey(SessionSummary())
     | "Format" >> beam.MapTuple(lambda uid, s: {
                        "user_id":    uid,
                        "start":      s["start"],
                        "end":        s["end"],
                        "duration_s": s["end"] - s["start"],
                        "events":     s["count"]})
     | "Write"  >> beam.io.WriteToBigQuery("p:analytics.sessions",
                                            schema="user_id:STRING,start:TIMESTAMP,end:TIMESTAMP,"
                                                   "duration_s:INTEGER,events:INTEGER"))

Step-by-step explanation.

1. Sessions(gap_size=30*60) creates a session window that grows as new elements arrive with the same key and a timestamp within 30 minutes of the last seen element. When the gap exceeds 30 minutes, the session closes and downstream aggregation fires.
2. Sessions are per-key. beam.Map(lambda e: (e["user_id"], e)) establishes user_id as the key. Each user's timeline of events forms their own session sequence.
3. When a new element arrives that would bridge two open sessions (e.g. filled the gap between a paused session and a new one), the runner merges the two sessions into one. The SessionSummary CombineFn's merge_accumulators is called to combine the accumulated state.
4. AfterWatermark() fires the pane when the watermark passes the session's end + the gap threshold — i.e., when the runner believes no more elements will arrive within the gap. Late data can re-fire under allowed-lateness rules, potentially extending the session.
5. Downstream sinks receive one row per closed session with the start, end, duration, and event count — the natural analytics unit for user-behaviour work.

Output.

User	Session start	Session end	Duration (s)	Events
user_1	10:00	10:22	1320	45
user_2	10:15	10:41	1560	12
user_1	11:05	11:20	900	8

Rule of thumb. Session windows are for user-level or entity-level analytics where the "unit of interest" is a bout of activity. Pick the gap size to match the domain — 30 min for web sessions, 5 min for IoT sensor bouts, 24 h for daily-visitor unification.

Senior interview question on windowing + triggers + lateness

A senior interviewer might ask: "Design a pipeline that computes hourly ad-impression counts per advertiser, with early panes fired every 5 minutes for freshness, on-time panes when the watermark passes, and late panes for up to 24 hours of allowed lateness. Explain the trigger, the accumulation mode, and how downstream reads the multi-pane emissions."

Solution Using composite triggers with early/on-time/late panes and accumulating mode

# hourly_ad_counts.py
import apache_beam as beam
from apache_beam.transforms.window import FixedWindows
from apache_beam.transforms.trigger import (
    AfterWatermark, AfterProcessingTime, AccumulationMode, Repeatedly, AfterCount)
from apache_beam.utils.timestamp import Duration


with beam.Pipeline() as p:
    (p
     | "Read"   >> beam.io.ReadFromPubSub(subscription="projects/p/subscriptions/impressions")
     | "Parse"  >> beam.Map(parse_impression)              # {"advertiser":..., "ts":..., "cost":...}
     | "Key"    >> beam.Map(lambda e: (e["advertiser"], 1))
     | "Window" >> beam.WindowInto(
                     FixedWindows(60 * 60),                # 1-hour fixed windows
                     trigger=AfterWatermark(
                         early=Repeatedly(AfterProcessingTime(5 * 60)),   # every 5 min before watermark
                         late=Repeatedly(AfterCount(1))                    # every late element
                     ),
                     allowed_lateness=Duration(seconds=24 * 3600),         # 24-h lateness
                     accumulation_mode=AccumulationMode.ACCUMULATING)
     | "Count"  >> beam.CombinePerKey(sum)
     | "AddPaneInfo" >> beam.ParDo(TagWithPaneInfo())
     | "Write"  >> beam.io.WriteToBigQuery(
                     "p:analytics.hourly_ad_counts",
                     schema="advertiser:STRING,window_start:TIMESTAMP,window_end:TIMESTAMP,"
                            "count:INTEGER,pane_kind:STRING,is_final:BOOLEAN",
                     write_disposition=beam.io.BigQueryDisposition.WRITE_APPEND))


class TagWithPaneInfo(beam.DoFn):
    def process(self, kv, window=beam.DoFn.WindowParam, pane_info=beam.DoFn.PaneInfoParam):
        advertiser, count = kv
        kind = "EARLY" if pane_info.timing.name == "EARLY" \
            else "ON_TIME" if pane_info.timing.name == "ON_TIME" \
            else "LATE"
        yield {
            "advertiser":   advertiser,
            "window_start": window.start.to_utc_datetime(),
            "window_end":   window.end.to_utc_datetime(),
            "count":        count,
            "pane_kind":    kind,
            "is_final":     pane_info.is_last,
        }

Step-by-step trace.

Pane	When it fires	Value emitted	is_final
Early pane #1	5 min into window (processing time)	partial count	false
Early pane #2	10 min into window	larger partial count	false
...	...	...	...
Early pane #12	60 min (right at watermark)	near-full count	false
On-time pane	watermark passes window end	full count at watermark	false
Late pane	each late-arriving element	slightly larger count	false
Final late pane	24 h after window end	final total	true

Downstream reads the BigQuery table and uses the is_final column to distinguish "still updating" rows from "guaranteed final" rows. Business dashboards can either wait for is_final = true (accurate but delayed) or read the latest pane by (advertiser, window_start) and accept some volatility.

Output:

Metric	Value
Panes per window	~12 early + 1 on-time + N late
BigQuery rows per window	~15–50
Freshness (early pane)	5 min
Final accuracy horizon	24 h
Downstream deduplication	on (advertiser, window_start) picking latest pane

Why this works — concept by concept:

• Composite trigger — AfterWatermark(early=..., late=...) combines the three phases into one specification. Early panes give freshness; on-time panes give the initial "watermark-complete" value; late panes catch stragglers.
• Repeatedly(AfterProcessingTime(5 min)) — the early trigger fires every 5 minutes of processing time regardless of event-time progress. This is what makes the freshness deterministic even when the watermark is stalled.
• 24-hour allowed lateness — accepts late data for a full day. The trade-off is memory: the runner keeps the window's state alive for 24 h after the on-time pane. For high-cardinality workloads, this is expensive — measure your state size.
• Accumulating mode — every pane emits the accumulated total, not the delta. Downstream deduplicates on (advertiser, window_start) and picks the latest pane. Discarding mode would require downstream to sum the deltas — more error-prone.
• Cost — state size scales with (open windows × keys × pane retention). For hourly windows × 24-h lateness × 1000 advertisers, the runner holds ~24,000 window-key state entries. Dataflow Streaming Engine handles this transparently; on Flink, verify RocksDB state backend has enough disk.

Streaming
Topic — streaming
Streaming problems on windowing and trigger design

Practice →

Optimization Topic — optimization Optimization problems on watermark + lateness tuning

Practice →

---

4. Runners — Dataflow / Flink / Spark

DataflowRunner is the flagship, FlinkRunner is the OSS alternative, SparkRunner is batch-first — pick per your infrastructure, not per your model

The mental model in one line: the same Beam pipeline runs on multiple runners, but each runner has a distinct sweet spot — dataflow is the flagship for GCP-native streaming with autoscale, beam flink runner is the strongest OSS streaming option with savepoints and rich state, SparkRunner is batch-first with limited streaming support, and the DirectRunner is for local tests. The pipeline code doesn't change; the operational story and cost model do.

DataflowRunner — the GCP-native flagship.

• Managed service. Google Cloud Dataflow is a fully managed runner — no cluster to provision, no state backend to configure. Job submission is a single API call.
• Autoscale. THROUGHPUT_BASED scaling grows and shrinks workers based on queue depth. Streaming and batch jobs both autoscale, but with different heuristics.
• Streaming Engine. Offloads shuffle from workers to a separate managed service. Enables faster autoscale, smaller worker VMs, and better checkpoint performance. ~15% price premium.
• Runner v2. The modern execution mode using the portable Fn API. Enables cross-language pipelines and better observability. Default on new jobs since 2023.
• Dataflow SQL. A separate service (Dataflow SQL editor) that lets you write Beam SQL queries via the Dataflow UI. Uses Beam SQL under the hood.
• Weak spots. GCP-only. Cost model can surprise (streaming Engine + persistent workers). Job graph updates are limited (drain + resubmit for most changes).

FlinkRunner — the strongest OSS streaming option.

• Two modes. Classic (JVM-only, Java SDK only) and portable (via Fn API, any SDK including Python). Portable mode is the default choice.
• State backends. RocksDB (recommended for large state), Filesystem (small state, low latency), or Heap (dev only). The state backend affects checkpoint cost and restart latency.
• Savepoints. Manually triggered snapshots that let you upgrade the pipeline gracefully — take savepoint, deploy new code, restore from savepoint. The Flink-specific "graceful upgrade" story.
• Checkpointing. Automatic periodic snapshots for fault tolerance. Configurable interval (typically 30 s to 5 min). Restart from the latest checkpoint on failure.
• Weak spots. OSS operational burden — you run the JobManager and TaskManagers. Portable-runner container startup adds latency. Some Beam features (splittable DoFns) have subtle behavioural differences vs Dataflow.

SparkRunner — batch-first, limited streaming.

• Batch-native. SparkRunner translates Beam PCollections into Spark RDDs. For bounded pipelines, execution is fine and scales with Spark's mature batch engine.
• Streaming support. Historically weak — the Spark micro-batch model doesn't fit the Beam continuous streaming model well. Structured Streaming support has improved, but most teams still avoid Beam+Spark for streaming.
• When to pick. Existing Spark infrastructure, batch-heavy workloads, or teams that already run Spark and want to try Beam without adopting a new cluster. Never for greenfield streaming.
• Weak spots. Streaming semantics are best-effort. Some windowing features are not supported. The Fn API story on Spark is less mature than Flink.

DirectRunner — the local dev runtime.

• In-process. Runs the pipeline in the current Python or Java process. Perfect for unit tests and quick iteration.
• Feature parity. Supports all Beam features — windowing, triggers, side inputs, stateful DoFns.
• Limits. Single-process; no autoscale; no fault tolerance. Never use in production.
• The PortableRunner on local. For cross-language tests, PortableRunner with local Docker containers runs the Fn API against a local job server. Useful for validating portable pipelines before promoting to Dataflow or Flink.

The Fn API portability layer.

• What it is. A gRPC-based protocol between the runner and language-specific SDK workers. Runners speak Fn API; SDK workers execute the Fn API commands in their language.
• Why it matters. Enables cross-language pipelines — a Java Beam pipeline can call Python DoFns via the Fn API. Enables portable runners — the same runner infrastructure serves Python, Java, and Go SDKs.
• The container model. SDK workers run in Docker containers. The runner starts a container per SDK-worker task; the Fn API is the wire.
• Where you see it in config. --environment_type=DOCKER (start containers), --environment_config=<image> (pick the image), --experiments=use_runner_v2 (Dataflow enable flag).

Common interview probes on runners.

• "Which runner do you pick for a new streaming pipeline?" — Dataflow if GCP; Flink if OSS/on-prem; not Spark.
• "What breaks when moving from DataflowRunner to FlinkRunner?" — I/O connectors (some are Dataflow-native); autoscale behaviour; certain state APIs.
• "What is the Fn API?" — gRPC protocol enabling cross-language pipelines and portable runners.
• "When would you use SparkRunner?" — legacy Spark infrastructure, batch-only workloads, and only reluctantly.

Worked example — the same pipeline on DataflowRunner and FlinkRunner

Detailed explanation. Take a Beam pipeline body and demonstrate the exact flag changes needed to move it between Dataflow and Flink. The pipeline logic is identical; only the runner-configuration flags change.

• The pipeline body. Any Beam pipeline — say, the earlier hourly ad-impression counter.
• Dataflow flags. --runner, --project, --region, --temp_location, plus autoscale + Streaming Engine flags.
• Flink flags. --runner, --flink_master, --environment_type, plus state-backend + checkpoint flags.

Question. Show the CLI invocation for the same hourly_ad_counts.py on DataflowRunner and FlinkRunner. Explain each runner-specific flag.

Input.

Component	Value
Pipeline file	hourly_ad_counts.py (unchanged)
Input	PubSub subscription
Output	BigQuery
Region	us-central1 (Dataflow) / on-prem (Flink)

Code.

# DataflowRunner — GCP managed
python hourly_ad_counts.py \
    --runner=DataflowRunner \
    --project=my-project \
    --region=us-central1 \
    --temp_location=gs://my-bucket/dataflow-tmp \
    --job_name=hourly-ad-counts-$(date +%Y%m%d) \
    --streaming \
    --num_workers=2 \
    --max_num_workers=20 \
    --autoscaling_algorithm=THROUGHPUT_BASED \
    --enable_streaming_engine \
    --experiments=use_runner_v2 \
    --sdk_container_image=us.gcr.io/my-project/beam-sdk:2.55.0

# FlinkRunner — portable, against an existing Flink cluster
python hourly_ad_counts.py \
    --runner=FlinkRunner \
    --flink_master=flink-jobmanager.internal:8081 \
    --environment_type=DOCKER \
    --environment_config=apache/beam_python3.11_sdk:2.55.0 \
    --streaming \
    --parallelism=8 \
    --checkpointing_interval=60000 \
    --state_backend=rocksdb \
    --state_backend_storage_path=file:///opt/flink/state \
    --savepoint_path=file:///opt/flink/savepoints

Step-by-step explanation.

1. On Dataflow, --project, --region, and --temp_location identify the GCP project, region, and staging bucket. --job_name names the job for the Dataflow UI. These four flags are always required.
2. --num_workers=2 --max_num_workers=20 --autoscaling_algorithm=THROUGHPUT_BASED enables autoscale between 2 and 20 workers. Dataflow measures queue depth and CPU utilisation to decide when to scale. --enable_streaming_engine offloads shuffle to the managed service (~15% cost, better performance).
3. --experiments=use_runner_v2 enables the modern portable execution mode. Combined with --sdk_container_image, this ships a custom SDK image with any per-team dependencies pre-baked.
4. On Flink, --flink_master points at the Flink JobManager. --environment_type=DOCKER tells Beam to spin up Python SDK worker containers (portable mode). --environment_config picks the SDK image. --parallelism sets the initial task-slot count.
5. --checkpointing_interval=60000 triggers a Flink checkpoint every 60 s. --state_backend=rocksdb uses RocksDB for large state (recommended for windowed pipelines). --savepoint_path is where manually-triggered savepoints land. Flink's operational story is savepoint-based upgrades; Dataflow's is drain-and-restart.

Output.

Concern	Dataflow flag	Flink flag
Pick runner	--runner=DataflowRunner	--runner=FlinkRunner
Autoscale	--max_num_workers + --autoscaling_algorithm	Flink cluster autoscale (external)
State	Streaming Engine (managed)	--state_backend=rocksdb
Fault tolerance	automatic	--checkpointing_interval
Graceful upgrade	drain + restart	savepoint + restore

Rule of thumb. For most GCP shops, Dataflow's operational simplicity + Streaming Engine + autoscale outweighs the 15% premium. For OSS / hybrid deployments, Flink's savepoint upgrade story is the operational win.

Worked example — Flink checkpointing and savepoint upgrade

Detailed explanation. Flink's checkpoint + savepoint model is a distinct operational advantage over Dataflow when you need graceful pipeline upgrades. A savepoint captures the entire pipeline state; you can then deploy new code and resume from the savepoint, preserving all in-flight state (windows, timers, buffered elements).

• Checkpoints. Automatic, periodic. For fault recovery.
• Savepoints. Manual, on-demand. For graceful upgrades. Named, versioned, and long-lived.
• The upgrade flow. (1) trigger savepoint on old job; (2) cancel old job; (3) submit new job with --savepoint_path=...; (4) new job resumes with old state.

Question. Walk through the CLI steps to gracefully upgrade a Beam-on-Flink pipeline from version N to version N+1 without losing any in-flight windowed state.

Input.

Component	Value
Cluster	Flink 1.18+
Job ID	abc123
Savepoint dir	hdfs:///beam/savepoints/

Code.

# Step 1 — trigger a savepoint on the running job (via Flink CLI)
flink savepoint abc123 hdfs:///beam/savepoints/ --allowNonRestoredState

# Output:
# Savepoint completed. Path: hdfs:///beam/savepoints/savepoint-abc123-a1b2c3

# Step 2 — cancel the old job (savepoint is separate; state is preserved)
flink cancel abc123

# Step 3 — deploy new pipeline code, submit as new job with savepoint path
python hourly_ad_counts_v2.py \
    --runner=FlinkRunner \
    --flink_master=flink-jobmanager:8081 \
    --environment_type=DOCKER \
    --environment_config=apache/beam_python3.11_sdk:2.55.0 \
    --streaming \
    --savepoint_path=hdfs:///beam/savepoints/savepoint-abc123-a1b2c3 \
    --state_backend=rocksdb

# Output: new job ID, e.g. "def456"

# Step 4 — verify the new job is running and processing
flink list --running

Step-by-step explanation.

1. flink savepoint abc123 hdfs:///beam/savepoints/ triggers a savepoint on the running job. The savepoint captures all operator state — windowed buffers, timer state, source offsets. The --allowNonRestoredState flag lets you drop state for operators that no longer exist in the new pipeline (useful when the graph has changed).
2. Once the savepoint completes and returns a path, cancel the old job. Cancelling clears the running job from the cluster but does not touch the savepoint — the state lives on in the savepoint directory.
3. Deploy the new pipeline code (hourly_ad_counts_v2.py) and submit it with --savepoint_path=<path>. Flink reads the savepoint, matches operators by their name/UID to the new job graph, and restores state for the matched operators.
4. Critical rule: named PTransforms ("Read", "Window", etc.) must keep the same names across versions for state to restore. Renaming a PTransform is a state-losing change. This is why senior Beam engineers set stable operator names at pipeline construction and treat them like schema.
5. After the new job starts, it resumes processing from where the old job left off. In-flight windows continue accumulating; late data still arrives into the correct window; sinks continue writing without duplicates (assuming exactly-once semantics on the sink).

Output.

Step	Command	Effect
1	flink savepoint abc123 ...	savepoint file created; job still running
2	flink cancel abc123	old job stopped; state preserved in savepoint
3	python ... --savepoint_path=...	new job started with restored state
4	flink list --running	verify new job ID present

Rule of thumb. Every production Beam-on-Flink pipeline uses savepoint-based upgrades. The operational cost of naming operators stably (a few extra lines of code) pays back the first time you need to deploy a bug fix without losing an hour of windowed state.

Worked example — DataflowRunner autoscale and cost tuning

Detailed explanation. DataflowRunner's autoscale story is one of its main selling points. Understanding the knobs — num_workers, max_num_workers, autoscaling_algorithm, enable_streaming_engine — is the difference between a $500/month pipeline and a $5000/month pipeline.

• num_workers. Starting worker count. Jobs boot with this many workers.
• max_num_workers. Hard cap. Autoscale never exceeds this.
• autoscaling_algorithm=THROUGHPUT_BASED. Reacts to queue depth and CPU utilisation. The default for streaming.
• enable_streaming_engine. Offloads shuffle to a managed service. Enables smaller worker VMs and faster autoscale.

Question. Tune a Dataflow streaming pipeline that peaks at 100k events/sec and idles at 1k events/sec, minimising cost while keeping latency below 30 s at peak.

Input.

Metric	Value
Peak throughput	100k events/sec
Idle throughput	1k events/sec
Peak duration	4 h/day
Idle duration	20 h/day
Target p99 latency	< 30 s

Code.

# Naive — flat sizing
python pipeline.py \
    --runner=DataflowRunner \
    --num_workers=50 \
    --max_num_workers=50 \
    --machine_type=n1-standard-4
# Cost: 50 workers × 24 h × 30 days = 36000 worker-hours × $0.20 = $7200/month

# Tuned — autoscale + Streaming Engine
python pipeline.py \
    --runner=DataflowRunner \
    --num_workers=2 \
    --max_num_workers=50 \
    --autoscaling_algorithm=THROUGHPUT_BASED \
    --enable_streaming_engine \
    --machine_type=n1-standard-2 \
    --experiments=use_runner_v2
# Cost estimate: 5 workers avg × 24 h × 30 d = 3600 worker-hours × $0.10 (n1-standard-2) = $360
#              + Streaming Engine overhead (~15% of compute) = $54
#              Total ~$414/month — 17× cheaper

Step-by-step explanation.

1. The naive version provisions 50 fixed workers permanently. Cost is workers × hours × price — flat, high, and does not track load. At peak, 50 workers are appropriate; at idle, they are 50× oversized.
2. The tuned version uses --num_workers=2 --max_num_workers=50 --autoscaling_algorithm=THROUGHPUT_BASED. Dataflow starts with 2 workers, scales up during peaks, and shrinks during idle periods. Average worker count over the day tracks average load.
3. --enable_streaming_engine moves shuffle out of the workers to a managed service. This lets Dataflow use smaller worker VMs (n1-standard-2 instead of n1-standard-4) and scale up/down faster because worker restart doesn't need to rehydrate shuffle state.
4. --experiments=use_runner_v2 enables the modern execution mode. Better observability, more efficient shuffle, and smoother autoscale. Should be on for all new jobs.
5. The cost model shifts from flat 50-worker rental to (small workers × autoscale) + streaming Engine overhead. Total cost drops ~17× while p99 latency stays inside the 30 s SLO because peak-time worker count still climbs to 50.

Output.

Metric	Naive	Tuned	Change
Avg workers	50	~5	-90%
Machine type	n1-standard-4	n1-standard-2	-50%
Monthly cost	$7200	~$414	-94%
p99 latency at peak	< 30 s	< 30 s	unchanged
Startup time on scale-up	~2 min (worker boot)	~30 s (Streaming Engine)	4× faster

Rule of thumb. For streaming Dataflow jobs, always enable Streaming Engine + autoscale + Runner v2. Never provision num_workers == max_num_workers unless you have a specific latency-vs-cost requirement that justifies the flat sizing.

Senior interview question on runner selection

A senior interviewer might ask: "You have a Beam pipeline that today runs on Dataflow. The company acquires a subsidiary that runs on-prem with a Flink cluster and no GCP. You need the same pipeline running in both places. Walk me through the migration story, what changes in code, what changes in config, and what operational patterns move with the pipeline."

Solution Using portable Beam pipeline with runner-agnostic code + per-runner deployment config

# unified_pipeline.py — runner-agnostic pipeline body
import argparse
import apache_beam as beam
from apache_beam.options.pipeline_options import PipelineOptions
from apache_beam.transforms.window import FixedWindows


def build(p, opts):
    (p
     | "Read"    >> beam.io.ReadFromKafka(
                     consumer_config={"bootstrap.servers": opts.bootstrap_servers,
                                      "group.id":          opts.consumer_group},
                     topics=[opts.input_topic],
                     with_metadata=False)
     | "Decode"  >> beam.Map(lambda kv: kv[1].decode())
     | "Parse"   >> beam.Map(parse_event)
     | "Key"     >> beam.Map(lambda e: (e["tenant"], 1))
     | "Window"  >> beam.WindowInto(FixedWindows(300))
     | "Count"   >> beam.CombinePerKey(sum)
     | "Format"  >> beam.MapTuple(lambda k, v: f"{k}\t{v}")
     | "Write"   >> beam.io.WriteToText(opts.output_prefix, file_name_suffix=".txt"))


def run():
    parser = argparse.ArgumentParser()
    parser.add_argument("--bootstrap_servers", required=True)
    parser.add_argument("--consumer_group",    required=True)
    parser.add_argument("--input_topic",       required=True)
    parser.add_argument("--output_prefix",     required=True)
    known, pipeline_args = parser.parse_known_args()
    options = PipelineOptions(pipeline_args, streaming=True)

    with beam.Pipeline(options=options) as p:
        build(p, known)


if __name__ == "__main__":
    run()

# Deployment on GCP (Dataflow)
python unified_pipeline.py \
    --bootstrap_servers=kafka.gcp.internal:9092 \
    --consumer_group=analytics-gcp \
    --input_topic=events \
    --output_prefix=gs://my-bucket/analytics/out \
    --runner=DataflowRunner \
    --project=my-gcp-project \
    --region=us-central1 \
    --temp_location=gs://my-bucket/tmp \
    --max_num_workers=20 \
    --autoscaling_algorithm=THROUGHPUT_BASED \
    --enable_streaming_engine

# Deployment on-prem (Flink)
python unified_pipeline.py \
    --bootstrap_servers=kafka.onprem.internal:9092 \
    --consumer_group=analytics-onprem \
    --input_topic=events \
    --output_prefix=file:///data/analytics/out \
    --runner=FlinkRunner \
    --flink_master=flink-jm.onprem.internal:8081 \
    --environment_type=DOCKER \
    --environment_config=apache/beam_python3.11_sdk:2.55.0 \
    --parallelism=8 \
    --checkpointing_interval=60000 \
    --state_backend=rocksdb

Step-by-step trace.

Concern	Dataflow deployment	Flink deployment
Pipeline code	unified_pipeline.py	unified_pipeline.py (same)
Source	KafkaIO against GCP Kafka	KafkaIO against on-prem Kafka
Sink	GCS	local filesystem (or HDFS)
Runner flag	--runner=DataflowRunner	--runner=FlinkRunner
Autoscale	Dataflow autoscale	Flink cluster autoscale (external)
State backend	Streaming Engine	RocksDB
Fault tolerance	automatic	--checkpointing_interval
Graceful upgrade	drain + restart	savepoint + restore
Monitoring	Dataflow monitoring UI	Flink UI + Prometheus

The pipeline body is bit-for-bit identical across deployments. Only the deployment configuration differs. The source (Kafka) is the same connector; the sink (GCS on Dataflow, local FS on Flink) requires slightly different write paths that the WriteToText transform handles transparently based on the URL scheme.

Output:

Deployment	Result
Dataflow, GCP	streaming, autoscaled, GCS sink
Flink, on-prem	streaming, savepoint-upgradeable, local sink
Code branch	one file, same logic
CI/CD	two deployment scripts, one build

Why this works — concept by concept:

• Runner-agnostic pipeline body — no code path in the pipeline depends on which runner it's on. All runner selection lives in PipelineOptions (parsed from CLI flags). The graph the SDK builds is identical.
• Portable KafkaIO — the Kafka source works on both runners via the Fn API portability layer. On Flink portable, KafkaIO runs in a Java container spawned by the portable runner; the pipeline sees the same connector API.
• Runner-specific config — Dataflow flags (--project, --region, --enable_streaming_engine) never appear in the Flink invocation; Flink flags (--flink_master, --state_backend) never appear on Dataflow. Config is per-runner; code is not.
• Sink scheme handling — WriteToText picks the filesystem based on the URL scheme (gs://, file://, hdfs://). No pipeline-code change to swap sinks between clouds.
• Cost — one codebase, two deployment paths, N runners of ops effort. The Beam abstraction pays for itself the first time you need to move a pipeline across clouds without rewriting it in Flink or Spark natively.

Streaming
Topic — streaming
Streaming problems on runner selection and portability

Practice →

Optimization Topic — optimization Optimization problems on Dataflow autoscale + Flink state tuning

Practice →

---

5. Beam SQL + production patterns

beam sql runs SQL queries on any PCollection via Calcite — combine with autoscale flags, checkpoint tuning, and Prometheus for a production-hardened deployment

The mental model in one line: beam sql is a Calcite-powered SQL layer that lets you express any Beam pipeline stage as a SQL query against PCollections; combined with runner-specific autoscale flags, checkpoint tuning, and native metrics, it is the production-ready face of Beam for teams that already speak SQL and want to keep the write-once-run-anywhere contract. Beam SQL is not a separate system — it is a PTransform that accepts SQL and produces PCollections.

Beam SQL — the Calcite-powered SQL layer.

• What it is. A PTransform (SqlTransform) that accepts a SQL string and executes it against named PCollections. Under the hood, Apache Calcite parses the SQL, plans the query, and generates a Beam graph.
• Supported SQL. SELECT, WHERE, JOIN (inner, left, right, full outer), GROUP BY, HAVING, window functions, aggregate functions, UNNEST, CTEs. Common table expressions work.
• Streaming windows in SQL. TUMBLE(ts, INTERVAL '5' MINUTE), HOP(ts, INTERVAL '1' MINUTE, INTERVAL '5' MINUTE), SESSION(ts, INTERVAL '30' MINUTE) — SQL-native window functions mapping to Beam's FixedWindows, SlidingWindows, Sessions.
• User-defined functions. Register Python or Java UDFs and call them from SQL. Enables domain logic that doesn't fit vanilla SQL.
• Trade-off. SQL is terse and readable for aggregations and joins; verbose for element-wise DoFns. Use SQL where SQL shines; use the SDK for complex stateful logic.

Portable pipelines + Docker containers.

• The container model. SDK workers run in Docker containers spawned by the portable runner. The runner sends Fn API commands to the container; the container executes them.
• Custom containers. Build a custom SDK image with your team's dependencies (pip install ...) baked in. Pass via --sdk_container_image=... on Dataflow or --environment_config=... on portable runners.
• Version pinning. The SDK version in the container must match the SDK the pipeline was built with. Mismatches cause runtime errors.

Metrics + monitoring.

• Beam built-in metrics. Counters, distributions, and gauges are first-class in the SDK. beam.metrics.Metrics.counter("ns", "name").inc() from inside a DoFn creates a metric visible in the runner's monitoring UI.
• Runner-specific integration. Dataflow exposes Beam metrics in the Dataflow UI and Cloud Monitoring. Flink exposes them via the Flink metrics reporter (Prometheus, statsd). All runners forward the same Beam metric definitions.
• Custom metrics for production. Success counts, retry counts, dead-letter counts, per-tenant volume — all instrumented as Beam metrics from inside DoFns. The runner surfaces them without extra work.

Failure modes + retries.

• Element retries. A DoFn that raises an exception on a specific element is retried by the runner. After the retry limit (runner-configurable), the element is either dropped or dead-lettered.
• Worker retries. A failed worker (OOM, disk-full) is replaced by the runner. State is restored from the last checkpoint. Data in flight during the failure is reprocessed.
• Retry backoff. beam.io.WriteToBigQuery and other production I/O sinks include internal retry with exponential backoff. External API calls in a DoFn should implement retry explicitly.
• Dead-letter routing. For elements that will never succeed (schema mismatch, bad JSON), route them into a dead-letter output via tagged outputs. Downstream jobs process the dead letter and re-emit valid records.

Senior interview signals.

• Watermark semantics. Naming what a watermark is and how it interacts with allowed lateness without prompting. The single most probed topic.
• Runner-specific quirks. "What does --enable_streaming_engine do?" "When would you pick Flink savepoints over Dataflow drain?" The infrastructure-side knowledge that separates seniors.
• SQL vs SDK trade-off. Knowing when to reach for SqlTransform versus building a custom DoFn. "SQL for aggregations and joins; SDK for stateful logic and per-element custom code."
• Cost model awareness. Explaining why Streaming Engine costs 15% more but enables 4× faster autoscale. The candidate who understands the runtime cost model beats the one who only knows the syntax.

Worked example — Beam SQL streaming aggregation

Detailed explanation. Replace a hand-written Beam pipeline that groups events by tenant and computes a 5-minute tumbling count with a Beam SQL query. Same semantics; 5 lines of SQL vs 15 lines of Python.

• The query. SELECT tenant, TUMBLE_START(ts, INTERVAL '5' MINUTE), COUNT(*) ... GROUP BY tenant, TUMBLE(ts, INTERVAL '5' MINUTE).
• The input. A Row-shaped PCollection with a tenant and ts column.
• The output. One row per (tenant, window) with the count.

Question. Write the Beam SQL query that counts events per tenant per 5-minute tumbling window, and show how to wire it as a PTransform in a Python pipeline.

Input.

Component	Value
Input	PCollection[Row(tenant, ts, event_id)] from Kafka
Window	5-minute tumbling
Aggregation	COUNT(*)
Sink	BigQuery

Code.

import apache_beam as beam
from apache_beam import Row
from apache_beam.transforms.sql import SqlTransform

SCHEMA = "tenant:STRING, window_start:TIMESTAMP, window_end:TIMESTAMP, cnt:INTEGER"

QUERY = """
SELECT
  tenant,
  TUMBLE_START(ts, INTERVAL '5' MINUTE) AS window_start,
  TUMBLE_END(ts, INTERVAL '5' MINUTE)   AS window_end,
  COUNT(*)                              AS cnt
FROM   PCOLLECTION
GROUP BY tenant, TUMBLE(ts, INTERVAL '5' MINUTE)
"""

with beam.Pipeline() as p:
    (p
     | "Read"   >> beam.io.ReadFromKafka(
                     consumer_config={"bootstrap.servers": "kafka:9092"},
                     topics=["events"])
     | "Decode" >> beam.Map(lambda kv: parse_row(kv[1]))
                     .with_output_types(Row(tenant=str, ts=beam.utils.timestamp.Timestamp,
                                             event_id=str))
     | "SQL"    >> SqlTransform(QUERY)
     | "Format" >> beam.Map(lambda r: {"tenant": r.tenant,
                                        "window_start": r.window_start.to_utc_datetime(),
                                        "window_end":   r.window_end.to_utc_datetime(),
                                        "cnt":          r.cnt})
     | "Write"  >> beam.io.WriteToBigQuery("p:analytics.tenant_counts", schema=SCHEMA))

Step-by-step explanation.

1. The pipeline reads a Kafka topic and decodes each message into a Row with typed fields (tenant, ts, event_id). The Row type is what SqlTransform expects — SQL needs typed columns.
2. SqlTransform(QUERY) accepts the SQL string. Calcite parses the query, resolves PCOLLECTION to the upstream PCollection, and plans the equivalent Beam graph. The output is a new PCollection of Row matching the SELECT's projection.
3. TUMBLE(ts, INTERVAL '5' MINUTE) is the SQL-native tumbling window function. It maps to Beam's FixedWindows(300) under the hood. TUMBLE_START and TUMBLE_END return the window boundaries — the same values you'd get from window.start / window.end in a DoFn.
4. The GROUP BY on tenant, TUMBLE(...) triggers a GroupByKey-shaped operation partitioned by the composite key. Aggregate functions (COUNT(*), SUM, AVG) compile to Beam's CombinePerKey.
5. Downstream, the SqlTransform output is a normal PCollection — you can pipe it through any other PTransform. Here we format each row as a dict and write to BigQuery. The whole pipeline is 20 lines; the SQL is 5 of them.

Output.

tenant	window_start	window_end	cnt
acme	2026-07-06 10:00	2026-07-06 10:05	1247
beta	2026-07-06 10:00	2026-07-06 10:05	892
acme	2026-07-06 10:05	2026-07-06 10:10	1301

Rule of thumb. Beam SQL for aggregations, joins, and windowed rollups. The syntax is terser than the SDK; the semantics compile to the same Beam primitives. Escape to the SDK for stateful DoFns and custom per-element logic that SQL can't express.

Worked example — production monitoring with Beam metrics + Prometheus

Detailed explanation. Beam's built-in metrics API (Metrics.counter, Metrics.distribution, Metrics.gauge) surfaces custom metrics in the runner's monitoring UI. On Dataflow, they appear in the Dataflow monitoring UI and Cloud Monitoring. On Flink, they go through the Flink metrics reporter to Prometheus.

• Metrics.counter. Monotonically increasing counter — success count, retry count, dead-letter count.
• Metrics.distribution. Distribution (min, max, mean, count) — per-element processing latency, message size.
• Metrics.gauge. Point-in-time value — queue depth, watermark lag.
• Namespace + name. Every metric has a namespace and a name. Namespace is usually the pipeline or transform; name is the specific metric.

Question. Instrument a Beam pipeline with counters for processed, dead_letter, and retry counts, and expose the metrics via Prometheus for on-call alerting.

Input.

Component	Value
Runner	FlinkRunner
Metrics reporter	Prometheus
Prometheus scrape endpoint	http://taskmanager:9250/metrics

Code.

import apache_beam as beam
from apache_beam.metrics import Metrics


class ProcessEvents(beam.DoFn):
    NAMESPACE = "pipeline.events"

    def __init__(self):
        self.processed   = Metrics.counter(self.NAMESPACE, "processed")
        self.dead_letter = Metrics.counter(self.NAMESPACE, "dead_letter")
        self.retry       = Metrics.counter(self.NAMESPACE, "retry")
        self.latency     = Metrics.distribution(self.NAMESPACE, "latency_ms")

    def process(self, event):
        import time
        t0 = time.time()
        try:
            result = expensive_transform(event)   # domain logic
            self.processed.inc()
            self.latency.update(int((time.time() - t0) * 1000))
            yield result
        except SchemaError:
            self.dead_letter.inc()
            yield beam.pvalue.TaggedOutput("bad", event)
        except TransientError:
            self.retry.inc()
            raise   # let the runner retry


with beam.Pipeline() as p:
    outputs = (p
               | "Read" >> beam.io.ReadFromPubSub(subscription="projects/p/subscriptions/s")
               | "Parse" >> beam.Map(parse_event)
               | "Process" >> beam.ParDo(ProcessEvents()).with_outputs("bad", main="good"))

    (outputs.good | "SinkOK"  >> beam.io.WriteToBigQuery("p:analytics.events", schema=SCHEMA))
    (outputs.bad  | "SinkBad" >> beam.io.WriteToText("gs://p/dead-letter/"))

# flink-conf.yaml — Prometheus reporter
metrics.reporters: prom
metrics.reporter.prom.factory.class: org.apache.flink.metrics.prometheus.PrometheusReporterFactory
metrics.reporter.prom.port: 9250

# prometheus.yml — scrape config
scrape_configs:
  - job_name: flink-beam
    static_configs:
      - targets: ['flink-taskmanager-1:9250', 'flink-taskmanager-2:9250']

Step-by-step explanation.

1. The ProcessEvents DoFn declares four metrics at init: three counters (processed, dead_letter, retry) and one distribution (latency_ms). Namespaces let you group related metrics; names identify each one.
2. Inside process(), the DoFn instruments each code path: self.processed.inc() on success, self.dead_letter.inc() on schema errors (routed to a tagged output), self.retry.inc() on transient errors (rethrown for the runner to retry).
3. self.latency.update(ms) records the per-element processing latency into a distribution. The runner rolls up (min, max, mean, count) and exposes each as a separate metric.
4. On Flink, the Prometheus reporter is configured via flink-conf.yaml. Every metric — Flink's own operator metrics plus Beam's user metrics — is exposed on port 9250 in Prometheus exposition format.
5. Prometheus scrapes every 15 s; Grafana dashboards read from Prometheus. Alerts on rate(pipeline_events_dead_letter[5m]) > 10 catch schema regressions; alerts on rate(pipeline_events_retry[5m]) > 100 catch transient-error storms.

Output.

Metric	Prometheus exposition
processed count	pipeline_events_processed_total{namespace="pipeline.events",...} 152341
dead_letter count	pipeline_events_dead_letter_total{...} 42
retry count	pipeline_events_retry_total{...} 178
latency p99	pipeline_events_latency_ms_bucket{le="500"} distribution

Rule of thumb. Every production Beam pipeline instruments at minimum: processed count, dead-letter count, retry count, and per-element latency. The Beam metrics API works on all runners; wire the runner-specific metrics reporter once and forget.

Worked example — dead-letter routing + retry with exponential backoff

Detailed explanation. External API calls in a DoFn fail transiently — network glitches, rate limits, 5xx errors. The correct pattern is (1) retry with exponential backoff inside the DoFn for transient errors, (2) dead-letter after N retries for stubborn failures, (3) instrument counters for both paths.

• Retry inside the DoFn. Bounded — usually 3–5 attempts with exponential backoff.
• Dead-letter after retry exhaustion. Tagged output emits the failed record for downstream processing.
• Instrumentation. Counters for retry, success, and dead-letter, plus a distribution for retry latency.

Question. Implement a DoFn that calls an external HTTP API with exponential-backoff retry, dead-letters after 3 failed attempts, and instruments the counts.

Input.

Parameter	Value
Max retries	3
Base backoff	100 ms
Backoff jitter	±25%
Retry-eligible errors	5xx, ConnectionError, TimeoutError
Dead-letter output tag	"dead_letter"

Code.

import time, random, logging
import apache_beam as beam
import requests
from apache_beam.metrics import Metrics
from apache_beam.pvalue import TaggedOutput

log = logging.getLogger("enrich")


class EnrichViaAPI(beam.DoFn):
    NAMESPACE = "enrich.api"
    MAX_RETRIES = 3

    def __init__(self):
        self.success  = Metrics.counter(self.NAMESPACE, "success")
        self.retries  = Metrics.counter(self.NAMESPACE, "retries")
        self.dead     = Metrics.counter(self.NAMESPACE, "dead_letter")
        self.duration = Metrics.distribution(self.NAMESPACE, "duration_ms")

    def process(self, event):
        for attempt in range(self.MAX_RETRIES + 1):
            t0 = time.time()
            try:
                r = requests.get(f"https://api.example.com/enrich/{event['id']}", timeout=5)
                r.raise_for_status()
                self.duration.update(int((time.time() - t0) * 1000))
                self.success.inc()
                yield {**event, "enrichment": r.json()}
                return
            except (requests.ConnectionError, requests.Timeout,
                    requests.HTTPError) as e:
                if isinstance(e, requests.HTTPError) and e.response.status_code < 500:
                    # 4xx — non-retriable
                    self.dead.inc()
                    yield TaggedOutput("dead_letter",
                                       {**event, "reason": f"http_{e.response.status_code}"})
                    return
                if attempt < self.MAX_RETRIES:
                    self.retries.inc()
                    base = 0.1 * (2 ** attempt)                    # 0.1, 0.2, 0.4
                    sleep = base * (1 + random.uniform(-0.25, 0.25))
                    log.warning("api retry %d/%d for %s: %s",
                                attempt + 1, self.MAX_RETRIES, event["id"], e)
                    time.sleep(sleep)
                else:
                    self.dead.inc()
                    yield TaggedOutput("dead_letter",
                                       {**event, "reason": f"retries_exhausted: {e}"})
                    return


with beam.Pipeline() as p:
    outputs = (p
               | "Read"    >> beam.io.ReadFromPubSub(subscription="projects/p/subscriptions/s")
               | "Parse"   >> beam.Map(parse_event)
               | "Enrich"  >> beam.ParDo(EnrichViaAPI()).with_outputs("dead_letter", main="ok"))

    (outputs.ok         | "SinkBQ" >> beam.io.WriteToBigQuery("p:analytics.enriched"))
    (outputs.dead_letter | "SinkDL" >> beam.io.WriteToText("gs://p/dead-letter/"))

Step-by-step explanation.

1. The DoFn retries up to MAX_RETRIES times inside process(). Each retry sleeps for 0.1 × 2^attempt seconds with ±25% jitter. Total worst-case retry budget is ~0.7 seconds — enough to absorb a transient blip, short enough not to stall the worker.
2. On a 5xx response or ConnectionError/Timeout, the DoFn retries. On a 4xx response (a permanent client error like 404), the DoFn dead-letters immediately — retrying a 404 is wasted work.
3. If all retries fail, the DoFn dead-letters the record with a reason string. The dead-letter output is a tagged output that a downstream sink writes to a GCS bucket for offline inspection.
4. Every code path increments a counter: success on OK, retries on each retry attempt, dead_letter on permanent failure. The distribution tracks the successful-call latency in ms.
5. Downstream, an alerting rule fires on rate(enrich_api_dead_letter_total[5m]) > 10 — a signal that either the upstream API is degraded or the upstream data is malformed. The runbook checks the dead-letter GCS bucket for the reason breakdown.

Output.

Metric	Prometheus name	Alert threshold
success	enrich_api_success_total	—
retries	enrich_api_retries_total	> 100/5m warns
dead_letter	enrich_api_dead_letter_total	> 10/5m pages
duration_ms	enrich_api_duration_ms histogram	p99 > 500ms warns

Rule of thumb. External API calls from a DoFn need exponential-backoff retry + bounded attempts + dead-letter routing + counter instrumentation. All four are non-negotiable in production; skipping any one produces intermittent data loss or worker stalls.

Senior interview question on Beam SQL + production hardening

A senior interviewer might ask: "You inherit a Beam pipeline written in the Python SDK — no metrics, no dead-letter routing, no SQL layer, and running on DataflowRunner with fixed 20 workers. Walk me through the production-hardening plan for the first two weeks, what you'd rewrite as Beam SQL, and what you'd leave in the SDK."

Solution Using a two-week hardening plan with metrics, dead-letter, SQL migration, and autoscale

Two-week hardening plan — Beam on Dataflow, production-ready
============================================================

Week 1 — Instrumentation and reliability
  Day 1  — Add Beam Metrics counters/distributions to every DoFn
           success/retry/dead_letter/duration_ms per stage
  Day 2  — Wire Cloud Monitoring dashboards + alerts
           alert on dead_letter rate, retry rate, watermark lag
  Day 3  — Introduce tagged outputs for dead-letter routing
           any DoFn that can fail permanently routes bad records to /* "bad" */
  Day 4  — Add exponential-backoff retry inside external-API DoFns
           bounded retries; 4xx = dead-letter immediately; 5xx = retry then dead-letter
  Day 5  — DirectRunner unit tests for happy + dead-letter paths
           TestPipeline + assert_that; run in CI on every PR

Week 2 — Performance and SQL migration
  Day 6  — Enable Streaming Engine + autoscale
           num_workers=2, max_num_workers=20, THROUGHPUT_BASED
           expected: 90% cost drop at steady state
  Day 7  — Migrate simple aggregations to Beam SQL (SqlTransform)
           group-by, joins, window-aggs — SQL is terser + auditable
  Day 8  — Keep custom stateful DoFns in the SDK
           per-user state, timers, complex per-element logic
  Day 9  — Set stable operator names for savepoint / job update
           name every PTransform (|"Name" >>) — never rely on default
  Day 10 — Introduce runner-independent tests
           add FlinkRunner integration test in a nightly job
           validates portability without full migration

Ongoing — runbook + weekly review
  - dead_letter GCS bucket triaged weekly
  - retry rate spikes reviewed with upstream owner
  - watermark lag monitored per topic
  - autoscale bounds re-tuned quarterly against peak/idle profile

Step-by-step trace.

Day	Activity	Deliverable
1	Add Metrics	counters + distributions in every DoFn
2	Dashboards	Cloud Monitoring + alerts
3	Dead-letter	TaggedOutput routing
4	Retry	exponential-backoff helper in API DoFns
5	Unit tests	TestPipeline + assert_that
6	Streaming Engine	autoscale enabled
7	SQL migration	SqlTransform for aggregations
8	SDK for state	stateful DoFns stay in SDK
9	Stable names	every PTransform named
10	Runner-indep test	nightly FlinkRunner integration

After two weeks, the pipeline has metrics, alerts, dead-letter routing, retry, unit + integration tests, autoscale + Streaming Engine, and a SQL layer for the aggregation stages. The team can now iterate on business logic without breaking production; the on-call has dashboards and a runbook.

Output:

Concern	Before	After
Metrics	none	Beam Metrics + Cloud Monitoring
Dead-letter	silent drops	TaggedOutput + GCS bucket
Retry	none	exponential backoff + dead-letter
Autoscale	fixed 20 workers	2–20, THROUGHPUT_BASED
Cost	flat 20 workers × 24 h	~5 workers avg (90% saving)
SQL	none	SqlTransform for aggregations
Runner portability	untested	nightly FlinkRunner test

Why this works — concept by concept:

• Metrics-first hardening — instrumentation before optimisation. Without counters you cannot see the dead-letter rate; without dashboards you cannot page on-call; without alerts you learn about incidents from users. Day 1 is metrics, always.
• Dead-letter routing over silent drops — TaggedOutput preserves failed records for offline inspection. The failure count becomes a first-class SLO signal; the failed records become debugging data.
• Beam SQL where SQL shines — group-by, joins, window aggregations are 5 lines of SQL vs 15 lines of Python. Custom stateful DoFns stay in the SDK because SQL can't express arbitrary state transitions. The split is per-stage, not per-pipeline.
• Streaming Engine + autoscale — the single biggest cost lever on Dataflow. Enabling both drops cost 5–20× on peak-vs-idle workloads while keeping p99 latency inside the SLO. The 15% Streaming Engine premium buys back 90% of the flat-worker cost.
• Cost — two senior-engineer weeks; the avoided cost of one production incident + the monthly Dataflow saving pays back within a month. O(events) per query at runtime; O(1) per quarter for autoscale re-tuning.

Streaming
Topic — streaming
Streaming problems on Beam SQL windowing and dead-letter routing

Practice →

Optimization
Topic — optimization
Optimization problems on Dataflow autoscale and metric tuning

Practice →

---

Cheat sheet — Beam recipes

• The 20-line Python pipeline template. with beam.Pipeline(options=PipelineOptions(pipeline_args)) as p: (p | "Read" >> ReadFromX(...) | "Parse" >> beam.Map(...) | "Key" >> beam.Map(...) | "Window" >> WindowInto(FixedWindows(60)) | "Combine" >> beam.CombinePerKey(sum) | "Write" >> WriteToY(...)). Every senior candidate writes this from memory. Named PTransforms ("Read", "Window") are non-negotiable — they anchor state on savepoint restore.
• Bounded vs unbounded PCollection default rule. Bounded PCollections need no windowing (implicit GlobalWindow). Unbounded PCollections must have a WindowInto before the first aggregation, or GroupByKey/CombinePerKey buffer forever. The moment a source is streaming, add the window.
• The four window types. FixedWindows(size) for non-overlapping tumbling windows; SlidingWindows(size, period) for overlapping smoothed windows; Sessions(gap_size) for per-key inactivity-separated sessions; global window (default) for bounded PCollections. Pick fixed unless a domain reason overrides.
• Trigger + lateness combo. AfterWatermark(early=Repeatedly(AfterProcessingTime(N)), late=Repeatedly(AfterCount(1))) + allowed_lateness=Duration(seconds=L) + AccumulationMode.ACCUMULATING. Early panes give freshness; on-time panes give the initial "complete" value; late panes catch stragglers; accumulating mode means downstream reads the latest pane and gets the total.
• DataflowRunner autoscale flags. --runner=DataflowRunner --num_workers=2 --max_num_workers=20 --autoscaling_algorithm=THROUGHPUT_BASED --enable_streaming_engine --experiments=use_runner_v2 --machine_type=n1-standard-2. Streaming Engine + autoscale + Runner v2 = 5–20× cost saving with no latency regression.
• FlinkRunner state + checkpoint flags. --runner=FlinkRunner --flink_master=host:8081 --environment_type=DOCKER --environment_config=<sdk-image> --parallelism=8 --checkpointing_interval=60000 --state_backend=rocksdb --state_backend_storage_path=file:///opt/flink/state --savepoint_path=file:///opt/flink/savepoints. Rocksdb for large state; savepoints for graceful upgrades.
• Beam SQL query example. SELECT tenant, TUMBLE_START(ts, INTERVAL '5' MINUTE) AS ws, TUMBLE_END(ts, INTERVAL '5' MINUTE) AS we, COUNT(*) AS cnt FROM PCOLLECTION GROUP BY tenant, TUMBLE(ts, INTERVAL '5' MINUTE). Wire via SqlTransform(QUERY). SQL for aggregations; SDK for stateful logic.
• CombinePerKey over GroupByKey + reduce. For any associative + commutative reduction (sum, count, min, max, average via CombineFn), use CombinePerKey. Shuffle volume drops 10–1000× because partial combining happens map-side. GroupByKey only when you need the full grouped list.
• Side input for small lookups. AsDict(lookup_pcoll) broadcasts a dict-shaped PCollection to every worker. Use for lookups < 100 MB. For larger lookups, use a stateful DoFn or a per-worker cache with a lookup service.
• TaggedOutput for dead-letter. yield TaggedOutput("bad", record) inside a DoFn, .with_outputs("bad", main="good") on the ParDo. Never silently drop invalid records; always route to a dead-letter output that a downstream sink writes to GCS/S3.
• Metrics namespace + name. Metrics.counter("stage.namespace", "success"); .inc() from inside a DoFn. Counter for monotonic counts; Distribution for latency and size; Gauge for point-in-time values. Runner-independent — works on Dataflow, Flink, Spark, DirectRunner.
• Retry inside DoFn. Exponential backoff 0.1 × 2^attempt with ±25% jitter, max 3–5 attempts. 4xx = dead-letter immediately; 5xx / ConnectionError / Timeout = retry then dead-letter. Never let the runner alone retry an external API — the retry loop belongs inside the DoFn.
• Named PTransforms for savepoint stability. p | "SpecificName" >> transform(...). Beam uses the name as the state key on savepoint restore. Never rely on the default auto-generated name — it changes across SDK versions and breaks upgrades.
• DirectRunner for unit tests, runner-specific for integration. with TestPipeline() as p: ... assert_that(output, equal_to([...])) for unit tests (sub-second in CI). Nightly --runner=DataflowRunner integration test against a fixed input for runner-specific quirks (autoscale, I/O, container startup).
• Portable Fn API for cross-language SDKs. --environment_type=DOCKER --environment_config=apache/beam_python3.11_sdk:2.55.0. The runner spawns Python worker containers alongside a Java-based Flink runtime. The Fn API is the wire; the pipeline never knows.

Frequently asked questions

What is Apache Beam and why does it matter in 2026?

Apache Beam is a unified programming model plus a set of language SDKs (Java, Python, Go) for defining batch and streaming data pipelines that can run on multiple execution engines — Google Cloud Dataflow, Apache Flink, Apache Spark, and the DirectRunner for local tests. In 2026 it matters for three reasons. First, it is the only major framework where the exact same pipeline body runs on both a bounded batch source (a file, a BigQuery table) and an unbounded streaming source (Kafka, PubSub) with identical semantics — the "unified batch and streaming" pitch every vendor makes actually works in Beam because the model was designed for it. Second, GCP's flagship data pipeline service (Dataflow) speaks Beam natively; any GCP-native data engineering role probes Beam knowledge. Third, the PCollection / PTransform / windowing vocabulary is the industry lingua franca for streaming semantics — even engineers who never ship a Beam pipeline benefit from learning the model because Flink, Dataflow, and Structured Streaming interviews use the same concepts.

Beam vs Flink — which one should I learn first?

Learn Beam first if you want a portable model and don't yet know which execution engine your team will land on — Beam's PCollection/PTransform/windowing vocabulary is transferable to Flink and Dataflow, and the same pipeline runs on both. Learn Flink first if you have a specific need for Flink's advanced state APIs (KeyedProcessFunction, ProcessFunction, custom state timers) or you know your team runs on-prem OSS infrastructure. In practice, senior data engineers know both — Beam for the model + Dataflow / FlinkRunner, and Flink native for the deeper state stories that Beam has not (yet) exposed. Beam-on-Flink (the FlinkRunner in portable mode) gives you Beam's write-once-run-anywhere plus Flink's savepoint upgrades and rich state backends — the best of both worlds for a team that doesn't want to commit to GCP.

Which Beam runner should I pick — Dataflow, Flink, or Spark?

Pick DataflowRunner if you're on GCP and want a managed streaming runtime — no cluster to run, autoscale + Streaming Engine included, ~15% price premium over classic Dataflow but 5–20× cost saving vs a fixed cluster. Pick FlinkRunner if you run on-prem or a non-GCP cloud, need savepoint-based graceful upgrades, or want a mature OSS state backend (RocksDB). Pick SparkRunner only for existing Spark shops running bounded workloads — SparkRunner's streaming story is best-effort and not the recommended path for new streaming pipelines. The DirectRunner is for local unit tests only; never in production. The choice is per-infrastructure, not per-workload — the same pipeline code compiles under any of them; only the runner-specific config flags change.

What is a PCollection and how is it different from a Spark RDD or a Flink DataStream?

A beam pcollection is an immutable, distributed dataset that can be either bounded (finite, batch) or unbounded (infinite, streaming). Every PTransform reads PCollections and emits new PCollections — the collection itself is never mutated. Compared to a Spark RDD, PCollections have first-class streaming semantics (windowing, watermarks, triggers are built into the model, not bolted on) and are runner-agnostic (RDDs are Spark-native). Compared to a Flink DataStream, PCollections do not distinguish between batch (DataSet) and streaming (DataStream) at the type level — the same PCollection type carries both bounded and unbounded data. The abstraction is deliberately weaker than either — no direct index-based access, no in-place mutation, no runner-specific operators — which is what enables the write-once-run-anywhere contract.

Do I need Beam if I'm already on GCP and can just use Dataflow?

You're already using Beam if you're using Dataflow — Dataflow is a Beam runner. The question is whether you write your pipeline in the Beam SDK (Python, Java, Go) or use Dataflow's higher-level services (Dataflow Templates, Dataflow SQL). For simple ETL, Dataflow Templates give a no-code path — pick a template, fill in params, deploy. For anything more complex — custom DoFns, dead-letter routing, multi-source enrichment, stateful processing — you write Beam SDK code and submit it to DataflowRunner. The upside of the SDK path is portability: the same pipeline body runs on FlinkRunner if you ever leave GCP, on DirectRunner in CI, and on other Beam runners as they mature. GCP-only teams can safely use Dataflow's convenience layers; multi-cloud or portability-conscious teams write the Beam SDK.

Beam SQL vs Flink SQL — how do they compare?

Beam SQL and Flink SQL are both Calcite-based SQL layers with streaming semantics — window functions (TUMBLE, HOP, SESSION), joins, aggregations, and CTEs. Beam SQL is a PTransform — you embed it inside a larger Beam pipeline via SqlTransform(QUERY), and the SQL result is a PCollection you can pipe into further Beam transforms. Flink SQL is more of a complete environment — it has a Table API, a SQL client, catalogs, and can be the entire application. Beam SQL is better when SQL is one stage of a larger SDK-based pipeline; Flink SQL is better when the entire application is SQL and you're already committed to Flink. The syntax overlaps 90%; the ecosystem and tooling differ. For portability across runners, Beam SQL keeps the write-once-run-anywhere contract; Flink SQL locks you to Flink.

Practice on PipeCode

• Drill the streaming practice library → for the windowing, trigger, and watermark problems senior Beam interviewers love.
• Rehearse on the ETL practice library → for the source-sink, enrichment, and dead-letter patterns that motivate the Beam model in the first place.
• Sharpen the tuning axis with the optimization practice library → for the autoscale, Streaming Engine, and state-backend problems.
• Stack the prerequisites against PipeCode's broader 450+ data-engineering catalogue to anchor the PCollection + windowing + runner intuition against real graded inputs.

Lock in Beam interview muscle memory

Runner docs explain flags. PipeCode drills explain the decision — when transaction pooling breaks prepared statements, when a session window closes, when Beam SQL beats a custom DoFn, when Dataflow autoscale saves 90% of your bill. Pipecode.ai is Leetcode for Data Engineering — pattern-first practice tuned for the production trade-offs senior data engineers actually face.

Practice streaming problems →
Practice optimization problems →