Google Pub/Sub & Dataflow: Streaming Pipelines on GCP

pub/sub and dataflow are the two managed services that carry almost every real-time workload on Google Cloud — Pub/Sub is the durable, planet-scale message bus that decouples the systems producing events from the systems consuming them, and Dataflow is the managed runner that executes an Apache Beam streaming pipeline against those events with windowing, watermarks, and horizontal autoscaling baked in. Between them they answer the question that every event-driven architecture eventually has to answer: how do you move billions of messages a day from thousands of publishers into a warehouse, a feature store, and a set of downstream microservices without losing a message when a consumer crashes, without double-counting a payment when a worker retries, and without falling hours behind when traffic spikes at midnight.

This guide is the walkthrough you wished existed the first time an interviewer asked "explain the difference between a topic and a subscription," or "your consumer keeps re-processing the same message — what's the ack deadline story?", or "how does Dataflow give you exactly-once when Pub/Sub only promises at-least-once?" It opens the two services in layers: the topic / subscription / acknowledgement model that governs how a message is delivered and retried, the delivery guarantees and ordering-key semantics that decide whether events arrive once or many times and in what order, the Apache Beam programming model of PTransforms that Dataflow executes, the windowing and watermark machinery that turns an unbounded stream into finite, correct aggregates even when data arrives late, and finally the exactly-once, autoscaling, and streaming design patterns that keep the whole thing correct and cheap under load. 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 →, sharpen event-time intuition on the real-time analytics practice library →, and rehearse the transforms on the data-processing practice library →.

---

On this page

• The Pub/Sub model — topics, subscriptions, and acknowledgement
• Delivery guarantees and message ordering
• Dataflow and Apache Beam pipelines
• Windowing and watermarks in Beam
• Exactly-once, autoscaling, and streaming patterns
• Cheat sheet — Pub/Sub and Dataflow streaming recipes
• Frequently asked questions
• Practice on PipeCode

---

1. The Pub/Sub model — topics, subscriptions, and acknowledgement

A topic is a named message channel; a subscription is a durable cursor into it, and the ack is the contract that a message was handled

The one-sentence invariant: Pub/Sub separates the act of publishing a message to a topic from the act of consuming it through a subscription, and every delivered message stays outstanding — eligible for redelivery — until the consumer acknowledges it within the subscription's ack deadline, which is the single mechanism that makes the whole bus reliable in the face of consumer crashes. A publisher writes to a topic and never knows or cares who is reading; each subscription attached to that topic receives its own independent copy of every message published after the subscription was created, and tracks its own backlog of unacknowledged messages. Get this decoupling wrong in an interview — say "a subscription is where you publish" or "all consumers share one queue" — and you signal you have never actually operated the service.

The four nouns you must name without prompting.

• Topic. A named resource messages are published to. It has no consumers of its own — it is a fan-out point. Publishing to a topic with zero subscriptions silently drops the message, because there is no subscription to store it. Creating the subscription before the first publish is therefore mandatory.
• Subscription. A durable named cursor attached to exactly one topic. Each subscription accumulates its own backlog and its own ack state. Two subscriptions on the same topic each get every message — that is how you fan a single event stream out to a warehouse loader and a real-time alerter independently.
• Message. The unit of delivery: a data byte payload plus optional string attributes, a server-assigned message_id, a publish_time, and an optional ordering_key. Max payload is 10 MB.
• Acknowledgement (ack). The consumer's signal that a message was processed and must not be redelivered. Until the ack arrives — or the ack deadline expires — Pub/Sub holds the message outstanding and will redeliver it. This is what makes delivery at-least-once by default.

The subscription delivery types — pull, push, and the native sinks.

• Pull subscription. The consumer (your code, or a Dataflow worker) calls Pull/StreamingPull to fetch a batch, processes it, then calls Acknowledge. Highest throughput; you own the flow control. This is what Dataflow uses under the hood.
• Push subscription. Pub/Sub POSTs each message to an HTTPS endpoint you register; a 2xx response is the ack. Good for serverless (Cloud Run, Cloud Functions) where you do not want a long-running puller.
• BigQuery subscription. Pub/Sub writes matching messages straight into a BigQuery table with no code and no Dataflow job — ideal for raw landing when you need no transformation.
• Cloud Storage subscription. Pub/Sub batches messages into files (Avro/text) in a bucket. Both native sinks trade flexibility for zero operational surface.

The acknowledgement lifecycle — the timers that decide redelivery.

• Ack deadline. The window, per delivered message, in which the consumer must ack. Default 10 seconds; configurable 10–600 seconds at the subscription level. Miss it and the message is redelivered.
• modifyAckDeadline / lease extension. A long-running handler extends the deadline before it expires (client libraries do this automatically while a message is being processed). This prevents redelivery of a message that is simply slow, not stuck.
• nack (negative acknowledgement). The consumer explicitly signals failure by setting the ack deadline to zero, triggering immediate redelivery instead of waiting out the deadline. Use it when you know you cannot process a message right now.
• Dead-letter topic. After max_delivery_attempts (5–100) redeliveries, Pub/Sub forwards the poison message to a configured dead-letter topic instead of retrying forever. Without it, one un-processable message blocks nothing but wastes redelivery budget indefinitely.

What interviewers listen for.

• Do you say "publish to a topic, consume through a subscription" as one clause? — required answer.
• Do you name the ack deadline as the reason a crashed consumer's messages get redelivered? — senior signal.
• Do you mention dead-letter topics for poison messages unprompted? — senior signal.
• Do you know that a subscription created after a publish never sees that message (no backlog time-travel)? — senior signal.
• Do you describe message retention as a subscription/topic setting (default 7 days, up to 31) rather than "messages live forever"? — required answer.

Worked example — creating a topic, two subscriptions, and a dead-letter path

Detailed explanation. The canonical first thing you build is a topic with two independent subscribers and a dead-letter safety net. One subscription feeds a Dataflow streaming pipeline; the other feeds a lightweight alerting Cloud Run service. A separate dead-letter topic catches messages that fail repeatedly. Walk through the gcloud commands that stand this up.

• Topic. orders-events — every order mutation is published here.
• Subscription A. orders-to-dataflow — pull, 60 s ack deadline (Dataflow leases longer batches).
• Subscription B. orders-to-alerter — push to a Cloud Run URL, 10 s ack deadline.
• Dead-letter. orders-dead-letter topic; both subscriptions route poison messages there after 5 attempts.

Question. Provision the topic, both subscriptions, and the dead-letter path so a crash in the alerter never loses a message and a poison message never loops forever.

Input.

Resource	Type	Key setting
orders-events	topic	message retention 7 days
orders-to-dataflow	pull subscription	ack deadline 60 s
orders-to-alerter	push subscription	endpoint = Cloud Run URL
orders-dead-letter	topic	max delivery attempts 5

Code.

# 1. The topic all publishers write to
gcloud pubsub topics create orders-events \
  --message-retention-duration=7d

# 2. The dead-letter topic + a subscription so dead letters are inspectable
gcloud pubsub topics create orders-dead-letter
gcloud pubsub subscriptions create orders-dead-letter-sub \
  --topic=orders-dead-letter

# 3. Subscription A — pull, consumed by Dataflow, 60s ack deadline
gcloud pubsub subscriptions create orders-to-dataflow \
  --topic=orders-events \
  --ack-deadline=60 \
  --dead-letter-topic=orders-dead-letter \
  --max-delivery-attempts=5

# 4. Subscription B — push to Cloud Run, 10s ack deadline
gcloud pubsub subscriptions create orders-to-alerter \
  --topic=orders-events \
  --ack-deadline=10 \
  --push-endpoint=https://alerter-xyz.a.run.app/pubsub \
  --dead-letter-topic=orders-dead-letter \
  --max-delivery-attempts=5

# 5. Grant Pub/Sub's service account permission to publish to the DLQ
PROJECT_NUMBER=$(gcloud projects describe "$(gcloud config get-value project)" --format='value(projectNumber)')
gcloud pubsub topics add-iam-policy-binding orders-dead-letter \
  --member="serviceAccount:service-${PROJECT_NUMBER}@gcp-sa-pubsub.iam.gserviceaccount.com" \
  --role="roles/pubsub.publisher"

Step-by-step explanation.

1. Creating the topic first is not optional — a message published to a topic with no subscriptions is dropped, because there is nowhere to store it. Retention of 7 days means a subscription can seek backwards up to a week, which is the replay budget.
2. The dead-letter topic and its own subscription exist so a human (or an audit job) can inspect poison messages. A dead-letter topic with no subscription would itself silently drop the poison messages — the same fan-out rule applies recursively.
3. Subscription A uses a 60-second ack deadline because Dataflow leases messages in larger batches and extends deadlines automatically; a longer base deadline reduces spurious redelivery under load.
4. Subscription B is a push subscription: Pub/Sub POSTs each message to the Cloud Run endpoint and treats a 2xx as the ack. If the Cloud Run service returns 5xx or times out, the message is redelivered — up to five times, after which it lands in the dead-letter topic.
5. The final IAM binding is the step everyone forgets: Pub/Sub's own service agent must be granted publisher on the dead-letter topic, otherwise dead-lettering silently fails and poison messages retry forever.

Output.

Scenario	Behaviour
Alerter (sub B) crashes mid-message	message redelivered after 10 s ack deadline
Dataflow (sub A) worker restarts	leased messages redelivered; no loss
Message fails 5 times on either sub	forwarded to orders-dead-letter
New subscription added tomorrow	sees only messages published from tomorrow on

Rule of thumb. Always create the subscription before the first publish, always attach a dead-letter topic with a real subscription behind it, and always grant Pub/Sub's service agent publisher on the DLQ. These three steps remove the most common "we silently lost messages" incident.

Worked example — a resilient pull consumer with lease extension

Detailed explanation. A hand-rolled pull consumer must extend the ack deadline while it works, ack on success, and nack on transient failure. The Python client library's StreamingPull does lease management for you, but understanding what it does is the point of the interview. Walk through a consumer that processes each message idempotently and lets the library manage the lease.

• Flow control. Cap outstanding messages so one slow batch cannot exhaust memory.
• Lease extension. The library auto-extends the deadline up to max_lease_duration while your callback runs.
• Ack / nack. message.ack() on success; message.nack() on a known-transient error to trigger immediate redelivery.

Question. Write a StreamingPull consumer that processes orders idempotently, extends the lease automatically, and nacks transient failures.

Input.

Parameter	Value
Subscription	orders-to-dataflow
Max outstanding messages	1000
Max lease duration	300 s
Idempotency key	message.message_id

Code.

# resilient_consumer.py — StreamingPull with flow control + lease management
from concurrent.futures import TimeoutError
from google.cloud import pubsub_v1

project_id = "acme-prod"
subscription_id = "orders-to-dataflow"

subscriber = pubsub_v1.SubscriberClient()
sub_path = subscriber.subscription_path(project_id, subscription_id)

# Flow control caps in-flight work; lease is auto-extended up to 300s
flow_control = pubsub_v1.types.FlowControl(
    max_messages=1000,
    max_lease_duration=300,   # library extends ackDeadline up to this
)

processed = set()  # in real life: Redis / Bigtable, not an in-memory set

def callback(message: pubsub_v1.subscriber.message.Message) -> None:
    mid = message.message_id
    try:
        if mid in processed:          # idempotency guard against redelivery
            message.ack()
            return
        handle_order(message.data, dict(message.attributes))
        processed.add(mid)
        message.ack()                 # success — never redelivered
    except TransientError:
        message.nack()                # immediate redelivery
    except PoisonError:
        message.ack()                 # drop locally; DLQ path handles repeats

streaming_pull = subscriber.subscribe(
    sub_path, callback=callback, flow_control=flow_control
)
print(f"Listening on {sub_path} ...")
with subscriber:
    try:
        streaming_pull.result()
    except TimeoutError:
        streaming_pull.cancel()
        streaming_pull.result()

Step-by-step explanation.

1. FlowControl(max_messages=1000) bounds how many messages are leased at once. Without it a burst can lease millions of messages, blow up worker memory, and cause deadline expirations because the callback cannot keep up.
2. max_lease_duration=300 tells the client library to keep calling modifyAckDeadline behind the scenes so a message being actively worked is not redelivered — up to five minutes, after which it gives up and lets the message be redelivered.
3. The processed set is the idempotency guard. Because delivery is at-least-once, the same message_id can arrive twice; checking it before doing work makes the handler safe to re-run. Production code uses a shared store (Redis/Bigtable) with a TTL, not a per-process set.
4. message.ack() on success removes the message from the backlog permanently. message.nack() on a transient error sets the deadline to zero for immediate redelivery — faster recovery than waiting out the full deadline.
5. For a poison message the code acks locally to avoid a hot retry loop; the subscription's max_delivery_attempts dead-letter policy is the real backstop that captures repeatedly-failing messages across all consumers.

Output.

Event	Consumer action	Pub/Sub result
First delivery, success	ack()	removed from backlog
Redelivery of same id	ack() (dedup hit)	removed; no double-processing
Transient error	nack()	redelivered immediately
Slow handler (200 s)	auto lease extension	not redelivered

Rule of thumb. Let the client library manage the lease, always set flow control, and always dedup by message_id — at-least-once means the same message will arrive twice eventually, and the handler must be safe when it does.

Worked example — replaying a backlog with seek and snapshots

Detailed explanation. When a downstream bug corrupts data, you often need to reprocess the last N hours of messages. Pub/Sub supports seek to a timestamp (within the retention window) and snapshots that capture a subscription's ack state so you can restore it later. Walk through both replay mechanisms.

• Seek to time. Rewinds a subscription's cursor to a wall-clock time within retention; every message after that time becomes unacked and is redelivered.
• Snapshot. A named capture of a subscription's exact ack state at a moment; seek to a snapshot restores that state precisely.
• Constraint. You can only seek within the topic/subscription retention window — replay is bounded by how much history you chose to keep.

Question. A bad deploy at 14:00 corrupted the warehouse. Restore the orders-to-dataflow subscription so it reprocesses everything from 13:55 onward.

Input.

Parameter	Value
Subscription	orders-to-dataflow
Retention	7 days
Seek target	2026-09-05T13:55:00Z
Pre-seek safety	snapshot before seeking

Code.

# 1. Snapshot current ack state FIRST (so the replay itself is reversible)
gcloud pubsub snapshots create orders-before-replay \
  --subscription=orders-to-dataflow

# 2. Seek the subscription back to just before the bad deploy
gcloud pubsub subscriptions seek orders-to-dataflow \
  --time=2026-09-05T13:55:00Z

# 3. If the replay makes things worse, restore the exact prior state
gcloud pubsub subscriptions seek orders-to-dataflow \
  --snapshot=orders-before-replay

# 4. Clean up the snapshot once the replay is verified good
gcloud pubsub snapshots delete orders-before-replay

Step-by-step explanation.

1. Snapshotting before seeking is the reversibility contract: the snapshot captures which messages were acked at that instant, so a botched replay can be undone with a second seek to the snapshot.
2. seek --time=13:55 marks every message with a publish_time at or after 13:55 as unacknowledged again, so the consumer redelivers all of them. Messages before 13:55 stay acked and are not resent.
3. Because seek only rewinds unacked state and does not delete data, downstream must be idempotent — the same messages will be reprocessed, and any non-idempotent sink will double-count.
4. seek --snapshot restores the captured ack state exactly, including which messages were and were not acked — more precise than a timestamp seek.
5. Snapshots and time-seek are both bounded by retention; you cannot replay 10 days of history on a 7-day retention window. Retention is a cost/replay-budget trade-off you set at provisioning time.

Output.

Step	Subscription state
After snapshot	ack state captured as orders-before-replay
After seek to 13:55	all messages ≥ 13:55 unacked → redelivered
After seek to snapshot	exact pre-replay ack state restored
After snapshot delete	replay finalised; snapshot removed

Rule of thumb. Snapshot before you seek, keep retention long enough to cover your worst realistic replay window, and never replay into a non-idempotent sink — seek redelivers, it does not de-duplicate.

Data engineering interview question on the Pub/Sub model

A senior interviewer often opens with: "Design the ingestion layer for an e-commerce platform where order events must reach both a Dataflow warehouse pipeline and a fraud-alerting service, a crashed consumer must never lose a message, and a message that cannot be parsed must not loop forever. Walk me through the topics, subscriptions, ack strategy, and the failure modes you would guard against."

Solution Using topics, independent subscriptions, ack deadlines, and a dead-letter topic

# 1. One topic, two independent durable subscriptions (fan-out)
gcloud pubsub topics create orders-events --message-retention-duration=7d
gcloud pubsub topics create orders-dead-letter

# 2. Warehouse path — pull, long ack deadline for batchy Dataflow
gcloud pubsub subscriptions create orders-to-dataflow \
  --topic=orders-events --ack-deadline=60 \
  --dead-letter-topic=orders-dead-letter --max-delivery-attempts=5 \
  --enable-exactly-once-delivery

# 3. Fraud path — push to Cloud Run, short ack deadline for low latency
gcloud pubsub subscriptions create orders-to-fraud \
  --topic=orders-events --ack-deadline=10 \
  --push-endpoint=https://fraud.a.run.app/pubsub \
  --dead-letter-topic=orders-dead-letter --max-delivery-attempts=5

# 4. Publisher — batched, with an ordering key so a customer's events stay ordered
from google.cloud import pubsub_v1

publisher = pubsub_v1.PublisherClient(
    publisher_options=pubsub_v1.types.PublisherOptions(enable_message_ordering=True),
    batch_settings=pubsub_v1.types.BatchSettings(max_messages=100, max_latency=0.05),
)
topic = publisher.topic_path("acme-prod", "orders-events")

def publish_order(order_id: int, customer_id: int, payload: bytes) -> None:
    future = publisher.publish(
        topic, payload,
        ordering_key=str(customer_id),          # per-customer ordering
        event_type="OrderPlaced",               # attribute for filtering
    )
    future.result()  # block for the server-assigned message_id

Step-by-step trace.

Step	Component	What happens
1	Publisher	order event published to orders-events with ordering_key=customer_id
2	Topic fan-out	one copy queued to orders-to-dataflow, one to orders-to-fraud
3	Dataflow sub	pull + 60 s deadline + exactly-once dedup by message_id
4	Fraud sub	push to Cloud Run; 2xx acks, 5xx redelivers
5	Failure	5 failed attempts on either sub → orders-dead-letter
6	Crash	unacked leases expire → redelivered; no message lost

After deployment, one publish reaches two independent consumers; the warehouse subscription enables exactly-once delivery so Dataflow never sees a Pub/Sub-side duplicate, the fraud subscription trades that for lower push latency, and any message failing five times on either path is quarantined in the dead-letter topic instead of blocking or looping.

Output:

Requirement	Mechanism	Result
Two independent consumers	two subscriptions on one topic	full fan-out, independent backlogs
No loss on crash	ack deadline + redelivery	crashed consumer's messages resent
No poison loop	dead-letter topic, 5 attempts	quarantined for inspection
Per-customer order	ordering key = customer_id	ordered within key

Why this works — concept by concept:

• Topic / subscription fan-out — the topic is a publish-only fan-out point; each subscription is an independent durable cursor with its own backlog and ack state, so the warehouse and fraud consumers never interfere with each other's progress.
• Ack deadline + redelivery — a message stays outstanding until acked or the deadline expires; a crashed consumer simply lets the lease lapse, and Pub/Sub redelivers, which is precisely what makes the default guarantee at-least-once.
• Dead-letter topic — after max_delivery_attempts the poison message is forwarded to a separate topic instead of retried forever, converting an unbounded retry loop into a bounded, inspectable quarantine.
• Ordering key — messages sharing an ordering_key published to the same region are delivered in publish order, giving per-customer ordering without forcing a single global order that would kill throughput.
• Cost — Pub/Sub bills by data volume (published + delivered) plus retained backlog; the design adds one extra topic (the DLQ) and one extra subscription copy per consumer. Throughput is effectively O(1) per message with horizontal fan-out; the only unbounded cost is retained backlog, capped by the retention window.

Streaming
Topic — streaming
Streaming ingestion and message-bus problems

Practice →

Data eng Topic — data-processing Data-processing problems on event pipelines

Practice →

---

2. Delivery guarantees and message ordering

At-least-once is the default, exactly-once is a subscription feature, and ordering is an opt-in per-key contract — never assume any of them

The mental model in one line: Pub/Sub delivers at-least-once by default (a message may arrive more than once and in any order), can be upgraded to exactly-once delivery per subscription (no duplicate delivery of an acked message, and a stricter ack contract), and can be made ordered per ordering_key (messages sharing a key, published to one region, arrive in publish order) — and every one of these is an explicit choice with a throughput cost, not a free default. The single most common production bug in GCP streaming is code that assumes ordering or uniqueness it never actually enabled. Senior interviewers probe this precisely because the naive mental model ("it's a queue, messages come out once in order") is wrong on both counts.

The three guarantees, precisely stated.

• At-least-once (default). Every message is delivered one or more times. Duplicates happen on ack-deadline expiry, publisher retries, and network partitions. Order is not guaranteed. This is the cheapest, highest-throughput mode.
• Exactly-once delivery (opt-in per subscription). Once a message is successfully acknowledged, it will not be redelivered; and the ack itself is confirmed so the consumer knows it "took." It does not mean a message is published exactly once — a publisher retry still creates two distinct messages with two message_ids. It reduces, but does not eliminate, the need for idempotency.
• Ordering (opt-in per key). Messages with the same ordering_key, published to the same region, are delivered to a subscription in the order they were published. Different keys are still parallel. Enabling ordering caps per-key throughput because the key's messages must be delivered serially.

Why duplicates and reordering happen — the mechanics.

• Publisher-side retries. A publish that times out but actually succeeded is retried by the client, creating two messages. Only application-level dedup (or an idempotent producer id) removes these — even exactly-once delivery cannot, because they are genuinely two messages.
• Ack-deadline expiry. A slow consumer that has not acked in time gets the message redelivered while still processing the first copy. Lease extension mitigates; it does not eliminate.
• Reordering. Without an ordering key, Pub/Sub optimises for throughput and makes no ordering promise — parallel delivery across many streams naturally interleaves.
• Ordering key throughput cap. A hot key (e.g. one enormous customer) serialises all its traffic; ordering trades parallelism for order, so a single key can become a throughput bottleneck.

The exactly-once delivery contract in detail.

• Guaranteed. No redelivery after a successful ack; ack responses are confirmed (an ackId that fails must be retried by the consumer).
• Not guaranteed. De-duplication of distinct messages created by publisher retries; cross-subscription uniqueness; exactly-once processing end-to-end (that is Dataflow's job, covered later).
• Cost. Lower maximum throughput per subscription and higher latency than at-least-once, because Pub/Sub does more bookkeeping.

What interviewers listen for.

• Do you state "at-least-once by default" as the first fact? — required answer.
• Do you distinguish exactly-once delivery (Pub/Sub) from exactly-once processing (Dataflow)? — senior signal.
• Do you note that publisher retries create genuine duplicates that even exactly-once delivery cannot remove? — senior signal.
• Do you know ordering is per-key and same-region, and that it caps per-key throughput? — senior signal.
• Do you still recommend idempotent consumers even with exactly-once enabled? — required answer.

Worked example — measuring your real duplicate rate

Detailed explanation. Before deciding whether to enable exactly-once delivery (and pay its throughput cost), measure how often duplicates actually happen on an at-least-once subscription. A small Beam or standalone counter that tracks distinct vs total message_ids over a window gives you the real number. Walk through the measurement.

• Metric. duplicate_rate = 1 - distinct_message_ids / total_deliveries over a rolling window.
• Source. The at-least-once subscription in question.
• Decision. If the rate is < 0.1% and the sink is idempotent, at-least-once + dedup is cheaper than exactly-once.

Question. Instrument a consumer to report the observed duplicate delivery rate over one hour.

Input.

Parameter	Value
Subscription	orders-at-least-once
Window	1 hour rolling
Dedup store	Redis set of message_ids
Report metric	duplicate_rate

Code.

# duplicate_meter.py — observe real duplicate rate on an at-least-once sub
import redis
from google.cloud import pubsub_v1

r = redis.Redis()
subscriber = pubsub_v1.SubscriberClient()
sub_path = subscriber.subscription_path("acme-prod", "orders-at-least-once")

def callback(message):
    mid = message.message_id
    # SADD returns 1 if newly added, 0 if the id was already present (duplicate)
    is_new = r.sadd("seen:orders", mid)
    r.expire("seen:orders", 3600)          # 1-hour rolling window
    r.incr("deliveries:total")
    if is_new == 0:
        r.incr("deliveries:duplicate")     # this delivery was a repeat
    process(message.data)
    message.ack()

subscriber.subscribe(sub_path, callback=callback)

def report():
    total = int(r.get("deliveries:total") or 0)
    dup = int(r.get("deliveries:duplicate") or 0)
    rate = (dup / total) if total else 0.0
    print(f"deliveries={total} duplicates={dup} duplicate_rate={rate:.4%}")

Step-by-step explanation.

1. SADD is the trick: it returns 1 when the message_id is new and 0 when it already exists in the set, so a return of 0 is exactly a duplicate delivery. This gives an exact count with no race, because Redis set operations are atomic.
2. The EXPIRE keeps the set to a one-hour window so the memory footprint stays bounded and the rate reflects recent behaviour, not all-time history.
3. Two counters — total deliveries and duplicate deliveries — give the ratio directly. The counter increments are separate from the set membership so the metric survives across the window boundary.
4. The measured rate drives the design decision: a rate under 0.1% with an idempotent sink means at-least-once plus your own dedup is cheaper and faster than enabling exactly-once delivery.
5. Critically, this measures delivery duplicates (same message_id twice). Publisher-retry duplicates (two different message_ids for the same logical event) are invisible here — they require a business key, not message_id, to detect.

Output.

Metric	Value (example)
Total deliveries	4,210,338
Duplicate deliveries	1,684
Duplicate rate	0.0400%
Decision	at-least-once + Redis dedup sufficient

Rule of thumb. Measure your duplicate rate before paying for exactly-once delivery. message_id dedup catches redelivery duplicates; a business key catches publisher-retry duplicates. Most workloads with an idempotent sink never need exactly-once delivery at all.

Worked example — ordering keys for per-entity order

Detailed explanation. An account-balance stream must apply debits and credits in publish order per account, but different accounts can be processed in parallel. Ordering keys give exactly this: serial per key, parallel across keys. Walk through enabling ordering end-to-end.

• Publisher. Enable message ordering; set ordering_key = account_id.
• Subscription. Enable message ordering on the subscription.
• Effect. Same-account events arrive in order; different accounts stay parallel.

Question. Publish and consume account transactions so each account's events are strictly ordered without serialising the whole stream.

Input.

Parameter	Value
Ordering key	account_id
Publisher option	enable_message_ordering=True
Subscription flag	--enable-message-ordering
Region	single region (required for ordering)

Code.

# ordered_publisher.py — per-account ordering
from google.cloud import pubsub_v1

publisher = pubsub_v1.PublisherClient(
    publisher_options=pubsub_v1.types.PublisherOptions(enable_message_ordering=True)
)
topic = publisher.topic_path("acme-prod", "account-txns")

def publish_txn(account_id: str, amount_cents: int, kind: str) -> None:
    data = f'{{"account":"{account_id}","amount":{amount_cents},"kind":"{kind}"}}'.encode()
    future = publisher.publish(topic, data, ordering_key=account_id)
    # On ordered publish, a failure pauses the key; you must resume it
    try:
        future.result()
    except Exception:
        publisher.resume_publish(topic, ordering_key=account_id)
        raise

# The subscription must also opt in to ordering
gcloud pubsub subscriptions create account-txns-sub \
  --topic=account-txns \
  --enable-message-ordering \
  --ack-deadline=30

Step-by-step explanation.

1. enable_message_ordering=True on the publisher makes messages with the same ordering_key publish serially in call order; without it, the ordering_key is ignored and no order is promised.
2. A failed ordered publish pauses that key — no further messages for the key are sent until you call resume_publish. This is a safety feature: it prevents a later message from jumping ahead of a failed earlier one, which would violate order.
3. The subscription must also enable ordering with --enable-message-ordering; enabling it only on the publisher is a classic half-configured bug that yields no ordering guarantee.
4. Ordering is per-key and single-region. Two accounts (two keys) are delivered in parallel, so throughput scales with key cardinality; a stream with millions of accounts still parallelises massively.
5. The cost is a per-key throughput cap: all of one account's events are serial, so a pathologically hot account becomes a bottleneck. Choose a key granularity that keeps any single key's rate modest.

Output.

Account (key)	Publish order	Delivery order
acct-1	credit, debit, credit	credit, debit, credit (preserved)
acct-2	debit, debit	debit, debit (preserved)
acct-1 vs acct-2	interleaved	parallel, no cross-key order

Rule of thumb. Enable ordering on both publisher and subscription, choose a key fine-grained enough that no single key is a hotspot, and remember that ordering caps per-key throughput — never use a single constant ordering key for a whole stream.

Data engineering interview question on delivery guarantees

A senior interviewer might ask: "You have a payments stream on Pub/Sub. The product owner says 'we must never double-charge a customer.' Explain exactly what Pub/Sub does and does not guarantee, whether enabling exactly-once delivery is sufficient, and design the end-to-end idempotency that actually prevents a double charge."

Solution Using exactly-once delivery plus a business-key idempotency ledger

# 1. Consumer with exactly-once delivery AND a business-key ledger
from google.cloud import pubsub_v1
import psycopg2

subscriber = pubsub_v1.SubscriberClient()
sub_path = subscriber.subscription_path("acme-prod", "payments-eos")  # exactly-once sub

def charge_once(conn, payment_id: str, customer_id: str, cents: int) -> None:
    """Idempotent charge keyed on the BUSINESS id, not message_id."""
    with conn:                                   # one transaction
        with conn.cursor() as cur:
            cur.execute(
                "INSERT INTO charge_ledger(payment_id) VALUES (%s) "
                "ON CONFLICT (payment_id) DO NOTHING RETURNING payment_id",
                (payment_id,),
            )
            if cur.fetchone() is None:
                return                            # already charged — skip
            cur.execute(
                "INSERT INTO charges(payment_id, customer_id, cents) VALUES (%s,%s,%s)",
                (payment_id, customer_id, cents),
            )

def callback(message):
    conn = get_conn()
    payment_id = message.attributes["payment_id"]   # business idempotency key
    charge_once(conn, payment_id,
                message.attributes["customer_id"],
                int(message.attributes["cents"]))
    message.ack()

subscriber.subscribe(sub_path, callback=callback)

-- 2. The idempotency ledger — the real double-charge guard
CREATE TABLE charge_ledger (
    payment_id  TEXT PRIMARY KEY,
    charged_at  TIMESTAMPTZ NOT NULL DEFAULT now()
);

CREATE TABLE charges (
    payment_id  TEXT PRIMARY KEY REFERENCES charge_ledger(payment_id),
    customer_id TEXT NOT NULL,
    cents       BIGINT NOT NULL
);

# 3. The subscription is exactly-once to cut redelivery duplicates
gcloud pubsub subscriptions create payments-eos \
  --topic=payments --enable-exactly-once-delivery --ack-deadline=60

Step-by-step trace.

Step	Input	Reasoning
1	payment p-9 delivered	exactly-once sub: no Pub/Sub redelivery after ack
2	publisher retried p-9	two message_ids, same payment_id attribute
3	first delivery	INSERT ledger succeeds → charge applied
4	second (retry) delivery	ON CONFLICT DO NOTHING → no row → skip
5	commit	ledger + charge commit atomically

After deployment, exactly-once delivery removes redelivery duplicates at the Pub/Sub layer, but the load-bearing guard against a double charge is the charge_ledger keyed on the business payment_id: even when a publisher retry manufactures two distinct messages for the same payment, the second insert conflicts and is skipped, and because the ledger insert and the charge insert commit in one transaction, there is no window where the charge exists without the ledger record.

Output:

Delivery	ledger insert	charge applied?
p-9 first time	success	yes
p-9 redelivery (same msg id)	prevented by exactly-once	n/a
p-9 publisher-retry (new msg id)	conflict → skip	no (correct)
p-10 first time	success	yes

Why this works — concept by concept:

• Exactly-once delivery — a subscription feature that stops Pub/Sub from redelivering an already-acked message, removing the redelivery class of duplicates at the source.
• Business-key ledger — the payment_id primary key catches the duplicates exactly-once delivery cannot: two genuinely distinct messages from a publisher retry. This is the guarantee the product owner actually needs.
• Atomic ledger + charge — wrapping the ledger insert and the charge in one transaction means "recorded as charged" and "actually charged" commit together; a crash between them cannot leave them inconsistent.
• ON CONFLICT DO NOTHING — turns the second attempt into a cheap no-op instead of an error, so redelivery and retries are safe by construction.
• Cost — exactly-once delivery lowers per-subscription throughput and adds latency; the ledger adds one indexed insert per payment. Net cost is O(1) per payment with a bounded ledger you TTL periodically — far cheaper than a reconciliation job hunting double charges after the fact.

Streaming
Topic — streaming
Streaming delivery-guarantee and dedup problems

Practice →

Real-time Topic — real-time-analytics Real-time analytics problems on ordered event streams

Practice →

---

3. Dataflow and Apache Beam pipelines

Apache Beam is the unified programming model; Dataflow is the managed runner — you write PTransforms over PCollections and Dataflow executes them elastically

The mental model in one line: Apache Beam is a portable programming model where you build a pipeline as a directed graph of PTransforms (ParDo, GroupByKey, Combine, Flatten) operating on PCollections — bounded for batch or unbounded for streaming — and Dataflow is Google's fully-managed runner that takes that graph, shards it across autoscaling workers, and runs the same code in batch or streaming mode by changing only the source and the windowing. The headline selling point interviewers want you to articulate is unification: one Beam pipeline runs as a nightly batch job over files or as a 24/7 streaming job over Pub/Sub, with no rewrite. Say "Dataflow is just hosted Spark" and you have missed the entire point of the Beam model.

The Beam nouns you must name.

• PCollection. An immutable, distributed dataset flowing between transforms. Bounded (a finite file → batch) or unbounded (a Pub/Sub stream → streaming). Every element carries an implicit timestamp and window.
• PTransform. An operation that consumes one or more PCollections and produces new ones. The core primitives are ParDo (per-element map/flatmap), GroupByKey (shuffle by key), CoGroupByKey (join), Combine (associative aggregation), and Flatten (union).
• ParDo + DoFn. ParDo applies a user DoFn to each element; the DoFn can emit zero, one, or many outputs, hold state, and set timers. It is the Beam equivalent of a mapper/flatmapper.
• Pipeline + Runner. The Pipeline object is the graph; the Runner executes it. DataflowRunner runs it on Google Cloud; DirectRunner runs it locally for tests. The same graph, different runner.

Why the unified model matters.

• One codebase, two modes. Swap ReadFromText (bounded) for ReadFromPubSub (unbounded) and add windowing, and the identical transform chain runs as streaming instead of batch.
• Portability. Beam pipelines can target Dataflow, Flink, or Spark runners. Dataflow is the managed, autoscaling, no-ops option on GCP.
• Correctness primitives built in. Event-time windowing, watermarks, and triggers are first-class in Beam, so late and out-of-order data are handled by the model, not by hand-rolled buffers.

How Dataflow executes the graph.

• Fusion. Adjacent element-wise transforms (a chain of ParDos) are fused into one stage so elements flow through without materialising intermediate PCollections — a major efficiency win.
• Shuffle. GroupByKey/Combine force a shuffle; Dataflow Shuffle (batch) and Streaming Engine (streaming) offload this state and shuffle off the workers to a managed backend, so workers stay stateless and autoscaling is cheap.
• Streaming Engine. Moves window state and shuffle out of worker VMs into the Dataflow service, enabling faster autoscaling and smaller workers. It is the recommended mode for streaming.
• Workers. Managed Compute Engine VMs that Dataflow scales up and down based on backlog and CPU.

What interviewers listen for.

• Do you describe Beam as the model and Dataflow as the runner, not as synonyms? — required answer.
• Do you name ParDo, GroupByKey, Combine as the core transforms? — senior signal.
• Do you explain fusion and why it avoids materialising intermediates? — senior signal.
• Do you mention Streaming Engine / Dataflow Shuffle moving state off workers? — senior signal.
• Do you know the same pipeline runs batch or streaming by changing source + windowing? — required answer.

Worked example — a streaming word-count over Pub/Sub

Detailed explanation. The "hello world" of streaming Beam: read lines from Pub/Sub, split into words, window into fixed one-minute windows, count per word, and write to BigQuery. It exercises ReadFromPubSub, ParDo, windowing, and CombinePerKey. Walk through the pipeline.

• Source. ReadFromPubSub on a subscription (unbounded PCollection).
• Transforms. FlatMap to split words → WindowInto fixed 60 s → CombinePerKey(sum).
• Sink. WriteToBigQuery with per-window counts.

Question. Write the streaming word-count pipeline and explain what makes it streaming rather than batch.

Input.

Parameter	Value
Source	subscription text-lines-sub
Window	fixed 60 s
Aggregation	count per word
Sink	BigQuery analytics.word_counts

Code.

# streaming_wordcount.py — the same shape works batch or streaming
import apache_beam as beam
from apache_beam.options.pipeline_options import PipelineOptions, StandardOptions

opts = PipelineOptions(streaming=True)          # <-- streaming mode
opts.view_as(StandardOptions).runner = "DataflowRunner"

def to_words(line: bytes):
    for w in line.decode("utf-8").split():
        yield (w.lower(), 1)

with beam.Pipeline(options=opts) as p:
    (
        p
        | "Read" >> beam.io.ReadFromPubSub(subscription="projects/acme-prod/subscriptions/text-lines-sub")
        | "Split" >> beam.FlatMap(to_words)
        | "Window" >> beam.WindowInto(beam.window.FixedWindows(60))   # 60s tumbling
        | "Count" >> beam.CombinePerKey(sum)
        | "Format" >> beam.Map(lambda kv: {"word": kv[0], "count": kv[1]})
        | "Write" >> beam.io.WriteToBigQuery(
              table="acme-prod:analytics.word_counts",
              write_disposition=beam.io.BigQueryDisposition.WRITE_APPEND,
          )
    )

Step-by-step explanation.

1. PipelineOptions(streaming=True) plus ReadFromPubSub make the source PCollection unbounded — the pipeline runs forever, processing messages as they arrive, rather than reading a finite file and stopping.
2. FlatMap(to_words) is sugar over ParDo: each input line produces zero-or-more (word, 1) output elements. This is the per-element transform primitive.
3. WindowInto(FixedWindows(60)) is the line that makes an unbounded stream aggregatable: it slices the infinite stream into 60-second event-time windows so CombinePerKey has a finite group to sum. Without windowing, GroupByKey/Combine on an unbounded PCollection would never emit, because the group never "ends."
4. CombinePerKey(sum) shuffles by word and sums the counts within each window. Combine is preferred over GroupByKey + manual sum because it is associative and Dataflow can pre-aggregate (combiner lifting) before the shuffle, cutting shuffle volume dramatically.
5. WriteToBigQuery with WRITE_APPEND streams each window's per-word counts into the table. Switching this pipeline to batch would only require swapping the source to ReadFromText and removing streaming=True — the transform chain is identical.

Output.

Window (event time)	word	count
12:00:00–12:01:00	pipeline	42
12:00:00–12:01:00	dataflow	17
12:01:00–12:02:00	pipeline	39

Rule of thumb. Use CombinePerKey over GroupByKey whenever the aggregation is associative — Dataflow lifts the combiner before the shuffle, so it moves far less data. Windowing is what makes an unbounded stream groupable; without it, aggregations never fire.

Worked example — a stateful ParDo for per-key deduplication

Detailed explanation. Beam's stateful DoFn lets a ParDo keep per-key state and set timers — the primitive behind deduplication, sessionization, and running aggregates. Build a stateful ParDo that drops duplicate order events per key within a time-to-live. Walk through state and timers.

• State. A BagState or ValueState scoped to (key, window).
• Timer. An event-time or processing-time timer to expire state.
• Effect. First occurrence of a key passes; repeats within the TTL are dropped.

Question. Write a stateful ParDo that emits an order only the first time its id is seen within a 10-minute TTL.

Input.

Parameter	Value
Key	order_id
State	ValueState[bool] "seen"
TTL	10 min (processing-time timer)
Effect	drop duplicates within TTL

Code.

# stateful_dedup.py — per-key dedup with a TTL timer
import apache_beam as beam
from apache_beam.transforms.userstate import ReadModifyWriteStateSpec, TimerSpec, on_timer
from apache_beam.transforms.timeutil import TimeDomain
from apache_beam.coders import BooleanCoder

class DedupFn(beam.DoFn):
    SEEN = ReadModifyWriteStateSpec("seen", BooleanCoder())
    EXPIRE = TimerSpec("expire", TimeDomain.REAL_TIME)   # processing-time TTL

    def process(self, element,
                seen=beam.DoFn.StateParam(SEEN),
                expire=beam.DoFn.TimerParam(EXPIRE)):
        order_id, payload = element
        if seen.read():
            return                       # duplicate within TTL — drop
        seen.write(True)
        expire.set(beam.utils.timestamp.Timestamp.now() + 600)  # 10 min
        yield (order_id, payload)

    @on_timer(EXPIRE)
    def expire_state(self, seen=beam.DoFn.StateParam(SEEN)):
        seen.clear()                     # allow the key again after TTL

# usage: keyed input, then the stateful ParDo
# pcoll | beam.Map(lambda o: (o["order_id"], o)) | beam.ParDo(DedupFn())

Step-by-step explanation.

1. Stateful ParDo requires a keyed PCollection — state is scoped per key. The upstream Map turns each order into (order_id, payload) so the runner routes all events for one id to the same state cell.
2. ReadModifyWriteStateSpec("seen") is per-key persistent state managed by Dataflow (in Streaming Engine, off the worker). Reading it tells us whether this id has been seen inside the current TTL.
3. On the first sighting the DoFn writes seen=True, sets a processing-time timer 10 minutes out, and emits the element. On any repeat within 10 minutes, seen.read() is True and the element is dropped.
4. @on_timer(EXPIRE) fires when the timer elapses and clears the state, so the same id seen after the TTL is treated as new — this bounds state growth and makes dedup a rolling window rather than forever.
5. This is the general shape for sessionization and running aggregates too: state holds the running value, a timer flushes or expires it. Dataflow persists this state durably so a worker crash does not lose it.

Output.

Event	State before	Emitted?
order-7 (t=0)	seen=∅	yes
order-7 (t=30s)	seen=True	no (dropped)
order-7 (t=11m)	expired/cleared	yes
order-8 (t=1m)	seen=∅	yes

Rule of thumb. Reach for a stateful ParDo when per-key memory across elements is required — dedup, sessions, running totals. Always pair state with a timer to expire it, or streaming state grows without bound.

Worked example — a CoGroupByKey streaming join

Detailed explanation. Joining two streams — orders and shipments — on a shared key is CoGroupByKey. Each side is keyed, co-grouped within a window, and the DoFn emits the joined result. Walk through a windowed stream-to-stream join.

• Inputs. Two keyed PCollections: orders and shipments, both keyed on order_id.
• Transform. CoGroupByKey groups both sides per key per window.
• Output. One joined record where both sides are present in the window.

Question. Join the order and shipment streams on order_id within a 5-minute window.

Input.

Parameter	Value
Left	orders keyed on order_id
Right	shipments keyed on order_id
Window	fixed 5 min
Join	inner (both present)

Code.

# stream_join.py — CoGroupByKey inner join within a window
import apache_beam as beam

def key_by_order(x):        # both streams -> (order_id, record)
    return (x["order_id"], x)

with beam.Pipeline(options=opts) as p:
    orders = (p | "OrdersIn" >> beam.io.ReadFromPubSub(subscription=ORDERS_SUB)
                | "OParse" >> beam.Map(parse_json)
                | "OKey" >> beam.Map(key_by_order)
                | "OWin" >> beam.WindowInto(beam.window.FixedWindows(300)))
    ships  = (p | "ShipsIn" >> beam.io.ReadFromPubSub(subscription=SHIPS_SUB)
                | "SParse" >> beam.Map(parse_json)
                | "SKey" >> beam.Map(key_by_order)
                | "SWin" >> beam.WindowInto(beam.window.FixedWindows(300)))

    joined = (
        {"order": orders, "ship": ships}
        | "CoGroup" >> beam.CoGroupByKey()
        | "InnerJoin" >> beam.FlatMap(emit_if_both)
    )

def emit_if_both(kv):
    order_id, grouped = kv
    for o in grouped["order"]:
        for s in grouped["ship"]:
            yield {"order_id": order_id, "total": o["total"], "carrier": s["carrier"]}

Step-by-step explanation.

1. Both streams are keyed on order_id and windowed identically (fixed 5 minutes). The matching windows are what let the join find both sides — a join only considers elements that fall in the same event-time window.
2. CoGroupByKey takes a dict of tagged PCollections and produces, per key, a dict of iterables — {"order": [...], "ship": [...]} — grouping both sides together.
3. emit_if_both is the join logic: it emits a joined record only when both the order and shipment iterables are non-empty for that key in that window — an inner join. Swapping to a left join means emitting orders even when shipments are empty.
4. Because the join is windowed, an order in window N and its shipment in window N+1 will not join — a real limitation. For late-arriving matches you widen the window, use session windows, or hold state in a stateful ParDo with a timer.
5. Dataflow implements CoGroupByKey as a shuffle; keeping the two streams' windows aligned and the key cardinality reasonable keeps the shuffle cost bounded.

Output.

order_id	total	carrier	note
ord-1	4200	FedEx	both in window → joined
ord-2	990	—	shipment in next window → not joined
ord-3	1500	UPS	both in window → joined

Rule of thumb. Window both sides of a CoGroupByKey identically, and choose the window wide enough to cover realistic arrival skew between the streams — a too-narrow window silently drops valid matches that arrive one window late.

Data engineering interview question on Dataflow and Beam

A senior interviewer might ask: "You need to enrich a Pub/Sub order stream with customer data from BigQuery, aggregate revenue per region per minute, and write to BigQuery — all in one Dataflow job that must also run as a nightly batch backfill over GCS files. Walk me through the Beam pipeline, how you keep one codebase for batch and streaming, and how Dataflow executes it efficiently."

Solution Using a single Beam pipeline with a side input, windowing, and a swappable source

# unified_pipeline.py — one pipeline, batch or streaming by config
import apache_beam as beam
from apache_beam.options.pipeline_options import PipelineOptions, StandardOptions

def build(opts: PipelineOptions, source, streaming: bool):
    with beam.Pipeline(options=opts) as p:
        # Side input: small customer dim, refreshed per job (batch) or periodically
        customers = (
            p | "DimRead" >> beam.io.ReadFromBigQuery(
                    query="SELECT customer_id, region FROM dim.customers",
                    use_standard_sql=True)
              | "DimKV" >> beam.Map(lambda r: (r["customer_id"], r["region"]))
        )
        cust_map = beam.pvalue.AsDict(customers)

        events = source(p)                      # swappable: PubSub or GCS
        if streaming:
            events = events | "Win" >> beam.WindowInto(beam.window.FixedWindows(60))
        else:
            events = events | "WinAll" >> beam.WindowInto(beam.window.GlobalWindows())

        (
            events
            | "Parse" >> beam.Map(parse_json)
            | "Enrich" >> beam.Map(
                  lambda o, cm: (cm.get(o["customer_id"], "UNKNOWN"), o["total"]),
                  cust_map)                      # side-input join on the dim
            | "SumRegion" >> beam.CombinePerKey(sum)
            | "Fmt" >> beam.Map(lambda kv: {"region": kv[0], "revenue": kv[1]})
            | "Write" >> beam.io.WriteToBigQuery(
                  "acme-prod:analytics.revenue_by_region",
                  write_disposition=beam.io.BigQueryDisposition.WRITE_APPEND)
        )

def pubsub_source(p):
    return p | "Stream" >> beam.io.ReadFromPubSub(subscription=ORDERS_SUB)

def gcs_source(p):
    return p | "Files" >> beam.io.ReadFromText("gs://acme-raw/orders/*.json")

# Streaming run
python unified_pipeline.py --streaming --runner=DataflowRunner \
  --region=us-central1 --enable_streaming_engine

# Batch backfill run — SAME transforms, different source
python unified_pipeline.py --runner=DataflowRunner --region=us-central1

Step-by-step trace.

Step	Transform	Batch vs streaming
1	source	GCS files (bounded) vs Pub/Sub (unbounded)
2	windowing	GlobalWindows vs FixedWindows(60)
3	side input	customer dim as AsDict map, same both modes
4	enrich	Map join on the side-input dict
5	aggregate	CombinePerKey(sum) — combiner-lifted
6	write	WriteToBigQuery append, same both modes

After deployment, the exact same transform chain runs as a 24/7 streaming job (Pub/Sub source, 60-second windows, Streaming Engine) and as a nightly batch backfill (GCS source, global window) — only the source function and the windowing line differ, both driven by a --streaming flag. Dataflow fuses the parse/enrich/format ParDos into one stage, lifts the CombinePerKey combiner before the shuffle, and autoscales workers to the backlog.

Output:

region	revenue (window)	mode
us-west	128,400	streaming (per minute)
us-east	96,750	streaming (per minute)
us-west	4,110,900	batch (full backfill)

Why this works — concept by concept:

• Unified Beam model — the same PTransform chain expresses both batch and streaming; only the source (bounded vs unbounded) and windowing change, so one tested codebase serves the live job and the backfill.
• Side input (AsDict) — the small customer dimension is broadcast to every worker as an in-memory map, giving a cheap stream-to-dimension join without a shuffle-heavy CoGroupByKey.
• CombinePerKey + combiner lifting — associative aggregation lets Dataflow pre-aggregate per bundle before the shuffle, moving far less data across the network than GroupByKey + sum.
• Fusion — adjacent element-wise ParDos (parse → enrich → format) are fused into a single stage, so elements stream through without materialising intermediate PCollections.
• Cost — O(elements) work with a single shuffle at the aggregation; Streaming Engine keeps state off workers so autoscaling is fast and workers are small. The batch and streaming jobs reuse the identical logic, halving maintenance cost versus two separate pipelines.

Data eng
Topic — data-processing
Data-processing problems on Beam-style transforms

Practice →

Streaming Topic — streaming Streaming pipeline and join problems

Practice →

---

4. Windowing and watermarks in Beam

Windows slice the unbounded stream into finite groups; the watermark decides when a window is "done"; triggers and allowed lateness handle the stragglers

The mental model in one line: windowing partitions an unbounded stream by event time into fixed, sliding, session, or global windows so aggregations have a finite group to emit, and the watermark — Beam's continuously-advancing estimate of "event time has progressed to at least T" — is what tells the runner a window is complete enough to fire, while triggers and allowed lateness control early results and how long a window stays open for late data. This is the single hardest cluster of concepts in stream processing, and it is where senior interviews live. If you cannot distinguish event time (when the thing happened) from processing time (when Dataflow saw it), you cannot reason about correctness under lateness at all.

Event time vs processing time — the distinction everything rests on.

• Event time. The timestamp of when the event actually occurred (attached at the source, e.g. event_ts in the payload). Correct aggregations are almost always event-time based.
• Processing time. The wall-clock time when Dataflow processes the element. It drifts arbitrarily behind event time under backlog.
• Skew. The gap between them. A phone that was offline uploads an event with a two-hour-old event time; processing time is now. Windowing by event time puts it in the right (old) window; the watermark and lateness settings decide whether that window is still open.

The four window types.

• Fixed (tumbling). Non-overlapping, equal-length windows (FixedWindows(60) = one per minute). Each element belongs to exactly one window. The default for "metrics per minute."
• Sliding (hopping). Overlapping windows of a given size at a given period (SlidingWindows(size=300, period=60) = a 5-minute window every minute). Each element belongs to multiple windows — good for moving averages.
• Session. Gap-based, data-driven windows (Sessions(gap=600)): a window extends as long as events keep arriving within the gap, then closes. Perfect for user-activity sessions.
• Global. One window for the whole stream; useless for streaming aggregation unless paired with a non-default trigger. It is the default for batch.

Watermarks — how Beam knows a window is complete.

• Definition. The watermark is the runner's estimate that "no more data with event time ≤ W will arrive." When the watermark passes a window's end, the default trigger fires that window.
• Source-derived. For Pub/Sub, Dataflow derives the watermark from message publish/timestamp metadata (or a custom timestamp attribute you configure); it is a heuristic, not a certainty.
• Late data. Anything arriving with event time behind the watermark is late. By default Beam drops late data; allowed_lateness keeps the window open to accept it.
• On-time vs late firings. The default AfterWatermark trigger fires once when the watermark passes the window end (on-time), and can fire again for each late element if you configure late firings.

Triggers and accumulation — controlling when and how results emit.

• Default trigger. AfterWatermark() — emit once when the window closes.
• Early triggers. AfterProcessingTime or AfterCount — emit speculative partial results before the window closes, for low-latency dashboards.
• Late triggers. Fire again when late data arrives within allowed_lateness.
• Accumulation mode. ACCUMULATING re-emits the full updated total on each firing; DISCARDING emits only the delta since the last firing. Choosing wrong double-counts or under-counts downstream.

What interviewers listen for.

• Do you separate event time from processing time in the first sentence? — required answer.
• Do you define the watermark as an estimate of event-time completeness? — senior signal.
• Do you name fixed / sliding / session and when each fits? — required answer.
• Do you explain allowed lateness + late firings for stragglers? — senior signal.
• Do you know accumulating vs discarding and its double-count risk? — senior signal.

Worked example — fixed windows with a custom event-time timestamp

Detailed explanation. To window by event time you must tell Beam each element's timestamp — usually from a field in the payload, not the Pub/Sub publish time. WithTimestamps (or a timestamp attribute on the Pub/Sub read) assigns it. Walk through per-minute revenue with a payload event time.

• Timestamp. Extract event_ts from the JSON payload.
• Assign. beam.window.TimestampedValue or a timestamp attribute on ReadFromPubSub.
• Window. FixedWindows(60) on event time.

Question. Compute per-minute revenue windowed on the payload's event_ts, not on arrival time.

Input.

Parameter	Value
Timestamp field	event_ts (epoch seconds)
Window	fixed 60 s
Aggregation	sum(total)
Source	Pub/Sub with timestamp attribute

Code.

# event_time_windows.py — window on payload event time, not arrival time
import apache_beam as beam
import json

class AddEventTime(beam.DoFn):
    def process(self, raw):
        rec = json.loads(raw.decode("utf-8"))
        ts = rec["event_ts"]                      # epoch seconds from the payload
        yield beam.window.TimestampedValue(rec, ts)

with beam.Pipeline(options=opts) as p:
    (
        p
        | "Read" >> beam.io.ReadFromPubSub(subscription=ORDERS_SUB)
        | "Stamp" >> beam.ParDo(AddEventTime())     # assign event-time timestamp
        | "Win" >> beam.WindowInto(beam.window.FixedWindows(60))
        | "KV" >> beam.Map(lambda r: (None, r["total"]))
        | "Sum" >> beam.CombinePerKey(sum)
        | "Fmt" >> beam.Map(lambda kv: {"revenue": kv[1]})
        | "Write" >> beam.io.WriteToBigQuery("acme-prod:analytics.revenue_min")
    )

Step-by-step explanation.

1. AddEventTime reads event_ts from the payload and wraps the record in a TimestampedValue, which is how Beam learns each element's event time. This is the crucial step: without it, Beam would use the ingestion time and a delayed event would land in the wrong window.
2. Alternatively, configuring ReadFromPubSub(timestamp_attribute="event_ts") lets Pub/Sub-level metadata drive the watermark directly, which produces a tighter watermark than assigning timestamps downstream.
3. FixedWindows(60) now slices by event time, so an event that happened at 12:00:30 but arrived at 12:03:00 still lands in the 12:00–12:01 window — correct by construction.
4. CombinePerKey(sum) aggregates within each event-time window. The single None key aggregates the whole stream per window; a real pipeline would key by region or product.
5. The window fires when the watermark passes 12:01:00. If the delayed event arrives after that, it is late — handled by the allowed-lateness settings in the next example, not here.

Output.

Event-time window	revenue
12:00:00–12:01:00	18,900
12:01:00–12:02:00	22,150
12:02:00–12:03:00	19,880

Rule of thumb. Always window on the payload's event time, not arrival time — set timestamp_attribute on ReadFromPubSub when you can, so the watermark itself is derived from real event times and windows are correct under delay.

Worked example — allowed lateness with early and late triggers

Detailed explanation. A dashboard wants speculative counts every 10 seconds, a final count when the window closes, and corrected counts for data that arrives up to 5 minutes late. That is an early processing-time trigger plus AfterWatermark plus late firings, with a chosen accumulation mode. Walk through the trigger configuration.

• Early. AfterProcessingTime(10) — speculative results every 10 s.
• On-time. AfterWatermark — fire when the window closes.
• Late. Refire on each late element within allowed_lateness=5m.
• Accumulation. ACCUMULATING so each firing carries the full corrected total.

Question. Configure windowing so the dashboard sees early speculative counts, a final count, and corrections for late data up to 5 minutes.

Input.

Parameter	Value
Window	fixed 60 s
Early trigger	every 10 s (processing time)
Late trigger	on each late element
Allowed lateness	5 min
Accumulation	ACCUMULATING

Code.

# triggers.py — early + on-time + late firings with accumulation
import apache_beam as beam
from apache_beam.transforms.trigger import (
    AfterWatermark, AfterProcessingTime, AccumulationMode, Repeatedly
)

windowed = (
    events
    | "Win" >> beam.WindowInto(
        beam.window.FixedWindows(60),
        trigger=AfterWatermark(
            early=Repeatedly(AfterProcessingTime(10)),   # speculative every 10s
            late=Repeatedly(AfterProcessingTime(0)),     # refire on each late elem
        ),
        allowed_lateness=300,                            # accept 5 min of late data
        accumulation_mode=AccumulationMode.ACCUMULATING, # each firing = full total
    )
    | "Count" >> beam.CombinePerKey(sum)
)

Step-by-step explanation.

1. AfterWatermark(...) is the composite trigger: its early sub-trigger fires before the watermark passes the window end, and its late sub-trigger fires after, for late data — the on-time firing at the watermark is implicit.
2. early=Repeatedly(AfterProcessingTime(10)) emits a speculative partial result every 10 seconds of processing time, giving the dashboard low-latency approximate counts before the window is complete.
3. allowed_lateness=300 keeps the window's state alive for 5 minutes after the watermark passes its end, so late elements can still update it; without this, late data is dropped the instant the watermark crosses.
4. late=Repeatedly(AfterProcessingTime(0)) fires again for each late element within that 5-minute grace, so corrections flow to the dashboard as stragglers arrive.
5. ACCUMULATING means each firing emits the full corrected total for the window, so a downstream sink can overwrite the prior value by window key. DISCARDING would emit only the increment since the last firing — appropriate only if the sink sums deltas itself. Choosing wrong here is the classic double-count bug.

Output.

Firing	Type	count
t+10s	early (speculative)	120
t+60s	on-time (watermark)	300
t+90s	late (straggler)	305
t+5m+	window closed	(no more firings)

Rule of thumb. Pick ACCUMULATING when the sink overwrites by window key (each firing is the final answer so far) and DISCARDING only when the sink sums deltas. Set allowed_lateness to your real observed straggler delay — too long wastes state, too short drops valid data.

Worked example — session windows for user activity

Detailed explanation. To measure user sessions — bursts of activity separated by idle gaps — session windows group events per key that are no more than a gap apart, closing the session after the gap of silence. Walk through per-user session duration.

• Window. Sessions(gap=600) — 10-minute inactivity gap closes a session.
• Key. user_id — sessions are per user.
• Output. Session start, end, and event count per user.

Question. Compute each user's session length using a 10-minute inactivity gap.

Input.

Parameter	Value
Window	Sessions(gap=600)
Key	user_id
Metric	session event count + span
Timestamp	event_ts

Code.

# sessions.py — per-user session windows
import apache_beam as beam

with beam.Pipeline(options=opts) as p:
    (
        p
        | "Read" >> beam.io.ReadFromPubSub(subscription=CLICKS_SUB)
        | "Stamp" >> beam.ParDo(AddEventTime())          # from earlier example
        | "KV" >> beam.Map(lambda r: (r["user_id"], 1))
        | "Sessions" >> beam.WindowInto(beam.window.Sessions(600))  # 10 min gap
        | "Count" >> beam.CombinePerKey(sum)
        | "Fmt" >> beam.Map(lambda kv: {"user_id": kv[0], "events": kv[1]})
        | "Write" >> beam.io.WriteToBigQuery("acme-prod:analytics.user_sessions")
    )

Step-by-step explanation.

1. Sessions(600) creates a data-driven window per key: as long as consecutive events for a user_id are within 10 minutes of each other, they merge into one session window; a gap larger than 10 minutes starts a new session.
2. Session windows are per key by nature — each user's activity forms its own set of sessions, and Beam merges overlapping session windows as new events fill gaps.
3. Because sessions merge dynamically, an event arriving between two existing sessions can merge them into one — Beam handles this window-merging automatically, which is why session logic is impossible to hand-roll correctly without the model.
4. CombinePerKey(sum) counts events per session window; you could equally emit min(event_ts) and max(event_ts) to get the session span.
5. The watermark and allowed-lateness rules still apply: a late click can extend or merge a session that already fired, so the same accumulation-mode discipline matters here too.

Output.

user_id	session	events
u-1	09:00–09:07	14
u-1	09:20–09:22	3
u-2	09:01–09:15	41

Rule of thumb. Use session windows for anything defined by "activity separated by idle gaps" — user sessions, device bursts, support conversations. The gap is the only tuning knob, and Beam's automatic window-merging is the reason you never roll this by hand.

Data engineering interview question on windowing and watermarks

A senior interviewer might ask: "You are computing per-minute click-through rate from a mobile app, but phones go offline and upload events hours later. Some events arrive two hours after they happened. Design the windowing, watermark, trigger, and lateness strategy so early dashboards are fast, final numbers are correct, and very-late data does not silently corrupt closed windows."

Solution Using event-time fixed windows, allowed lateness, and accumulating triggers with a late sink

# late_tolerant_ctr.py — correct per-minute CTR under hours-late data
import apache_beam as beam
from apache_beam.transforms.trigger import (
    AfterWatermark, AfterProcessingTime, AccumulationMode, Repeatedly
)

with beam.Pipeline(options=opts) as p:
    events = (
        p
        | "Read" >> beam.io.ReadFromPubSub(
              subscription=EVENTS_SUB, timestamp_attribute="event_ts")  # watermark from event time
        | "Parse" >> beam.Map(parse_json)
        | "Win" >> beam.WindowInto(
              beam.window.FixedWindows(60),
              trigger=AfterWatermark(
                  early=Repeatedly(AfterProcessingTime(15)),   # fast dashboards
                  late=Repeatedly(AfterProcessingTime(0)),     # corrections
              ),
              allowed_lateness=2 * 60 * 60,                    # 2 hours late accepted
              accumulation_mode=AccumulationMode.ACCUMULATING,
          )
    )

    ctr = (
        events
        | "ToRatio" >> beam.Map(lambda e: (e["ad_id"], (e["clicks"], e["impressions"])))
        | "Sum" >> beam.CombinePerKey(
              lambda vs: (sum(c for c, _ in vs), sum(i for _, i in vs)))
        | "CTR" >> beam.Map(lambda kv: {
              "ad_id": kv[0],
              "ctr": kv[1][0] / kv[1][1] if kv[1][1] else 0.0})
        # Sink overwrites by (ad_id, window) so accumulating firings replace prior values
        | "Write" >> beam.io.WriteToBigQuery(
              "acme-prod:analytics.ctr_per_min",
              write_disposition=beam.io.BigQueryDisposition.WRITE_APPEND)
    )

Step-by-step trace.

Step	Config	Effect
1	timestamp_attribute=event_ts	watermark derived from real event time
2	FixedWindows(60)	per-minute event-time windows
3	early AfterProcessingTime(15)	speculative CTR every 15 s
4	allowed_lateness=2h	windows accept data up to 2 h late
5	late firing	each straggler updates the window
6	ACCUMULATING	each firing emits full corrected CTR

After deployment, dashboards get an approximate CTR within 15 seconds, a final on-time CTR when the watermark passes each minute, and corrected values as phones upload hours-old events within the two-hour grace; anything later than two hours is dropped rather than silently corrupting an already-reported window, and because firings accumulate, the sink simply keeps the latest value per (ad_id, window).

Output:

ad_id	window	ctr	firing
ad-7	12:00	0.031	early (15 s)
ad-7	12:00	0.042	on-time
ad-7	12:00	0.045	late (+40 min)
ad-7	12:00	(dropped)	>2 h late

Why this works — concept by concept:

• Event-time windowing — deriving the watermark from event_ts puts each event in the minute it actually happened, so an event uploaded hours late still counts toward the correct window rather than "now."
• Watermark + on-time firing — the watermark's crossing of each window end produces the authoritative on-time result, the number most consumers treat as final.
• Allowed lateness (2 h) — keeps window state alive long enough to absorb the real straggler distribution; the explicit bound is what stops unbounded state growth while still accepting realistic late uploads.
• Accumulating mode — each firing carries the full corrected total, so the sink overwrites by (ad_id, window) and never double-counts, which a discarding-mode delta stream would risk.
• Cost — window state is retained for the allowed-lateness horizon (2 h), the dominant memory cost; Streaming Engine holds it off-worker. Compute is O(events) with one shuffle at the combine. The 2-hour bound is the single dial trading state cost against correctness for late data.

Real-time
Topic — real-time-analytics
Real-time analytics problems on windowing and late data

Practice →

Streaming Topic — streaming Streaming problems on watermarks and triggers

Practice →

---

5. Exactly-once, autoscaling, and streaming patterns

Dataflow gives exactly-once processing on top of Pub/Sub's at-least-once delivery, autoscales to backlog, and rewards a handful of proven design patterns

The mental model in one line: Dataflow provides exactly-once processing — each input element affects the pipeline's aggregated state exactly once — by de-duplicating Pub/Sub messages on their message_id, checkpointing state deterministically, and pairing with exactly-once sinks (BigQuery Storage Write API), while autoscaling continuously resizes the worker pool to the backlog and CPU, and a small catalogue of patterns (dead-letter branches, side-input enrichment, drain-vs-cancel updates) keeps the whole streaming system correct, elastic, and operable. The distinction interviewers hammer is that exactly-once processing (Dataflow) is a different guarantee from exactly-once delivery (a Pub/Sub subscription feature) — Dataflow gives you correct aggregates even over an at-least-once source.

Exactly-once processing — how Dataflow achieves it.

• Source dedup by message_id. Dataflow records the message_id of every Pub/Sub message it has processed and drops redeliveries, so an at-least-once source becomes effectively once inside the pipeline.
• Deterministic shuffle + checkpointing. State and shuffle are checkpointed; on worker failure, Dataflow replays from the last checkpoint deterministically, so retried work does not double-apply to aggregates.
• Exactly-once sinks. The BigQuery Storage Write API supports exactly-once via stream offsets; without an exactly-once sink, the pipeline can still emit a duplicate on the output side even if internal processing was once.
• The caveat. Exactly-once processing covers pipeline state, not non-idempotent external side effects in a DoFn (e.g. calling a payment API). Those still need idempotency keys.

Autoscaling — how Dataflow sizes the job.

• Horizontal autoscaling. Adds or removes workers based on backlog (estimated time to clear the Pub/Sub backlog) and CPU utilisation. Streaming jobs scale continuously; --max_num_workers caps the pool.
• Streaming Engine. Moves window/shuffle state off workers into the service, so scaling is fast and does not require re-distributing large per-worker state.
• Vertical autoscaling (Dataflow Prime). Adjusts worker memory to avoid OOMs without manual machine-type tuning.
• Backlog signal. The single most important autoscaling input for streaming is the Pub/Sub backlog (unacked message age/size); a growing backlog is the trigger to scale up.

The core streaming patterns.

• Dead-letter branch. In the pipeline, route un-parseable elements to a side output and write them to a dead-letter table/topic instead of crashing the job. A streaming job must never die on one bad record.
• Side-input enrichment. Broadcast a small, slowly-changing dimension as a side input (optionally refreshed periodically) to enrich the stream without a shuffle.
• Drain vs cancel. Drain stops ingestion but lets in-flight windows finish and flush — the safe way to update a pipeline. Cancel stops immediately and discards in-flight state — data loss.
• Update in place. Deploy new pipeline code with --update, mapping state from the old job to the new so windows and dedup state carry over.

Operational guardrails.

• Backlog + system lag are the two health metrics; rising system lag means the job cannot keep up.
• Max workers bounds cost; too low starves the job under spikes, too high risks runaway spend.
• Snapshots let you stop and restart a streaming job from a saved state.
• Dead-letter volume is an alerting metric — a spike means an upstream schema change.

What interviewers listen for.

• Do you distinguish exactly-once processing (Dataflow) from exactly-once delivery (Pub/Sub)? — required answer.
• Do you name message_id dedup + checkpointing as the mechanism? — senior signal.
• Do you cite backlog as the autoscaling signal and Streaming Engine's role? — senior signal.
• Do you know drain vs cancel and why drain is the safe update path? — senior signal.
• Do you route bad records to a dead-letter side output rather than crashing? — required answer.

Worked example — a dead-letter side output for bad records

Detailed explanation. A streaming job must survive malformed input. Beam's tagged outputs let a DoFn emit good records to the main output and bad records to a dead-letter output, which is written to a separate BigQuery table. Walk through the pattern.

• Parse DoFn. Try to parse; on success emit main output, on failure emit to the dead_letter tag.
• Split. with_outputs separates the two streams.
• Sinks. Good → main table; dead → dead-letter table with the raw payload and error.

Question. Build a parse step that never crashes the job and quarantines bad records.

Input.

Parameter	Value
Main output	parsed orders
Dead-letter tag	dead
Dead-letter sink	analytics.orders_dlq
Captured on failure	raw bytes + error string

Code.

# dead_letter.py — tagged outputs so one bad record can't kill the job
import apache_beam as beam
import json

class ParseOrders(beam.DoFn):
    def process(self, raw):
        try:
            yield json.loads(raw.decode("utf-8"))              # main output
        except Exception as e:
            yield beam.pvalue.TaggedOutput(                    # dead-letter output
                "dead", {"raw": raw.decode("utf-8", "replace"), "error": str(e)})

parsed = (
    p | "Read" >> beam.io.ReadFromPubSub(subscription=ORDERS_SUB)
      | "Parse" >> beam.ParDo(ParseOrders()).with_outputs("dead", main="good")
)

# Main path continues; dead path is quarantined
parsed.good | "Process" >> beam.Map(process_order) | "OK" >> beam.io.WriteToBigQuery(
    "acme-prod:analytics.orders")
parsed.dead | "DLQ" >> beam.io.WriteToBigQuery(
    "acme-prod:analytics.orders_dlq")

Step-by-step explanation.

1. The ParseOrders DoFn yields a normal element on success and a TaggedOutput("dead", ...) on failure, so a single malformed message becomes a quarantined row instead of an exception that stalls or crashes the job.
2. .with_outputs("dead", main="good") splits the ParDo into two PCollections: parsed.good (main) and parsed.dead (the tagged failures).
3. The good path continues into the real processing and the main sink; the dead path is written to a separate orders_dlq table with the raw payload and the error string, preserving everything needed to debug and replay.
4. This is the difference between a streaming job that survives a bad upstream deploy and one that pages the on-call at 3 a.m.: the job keeps running, the DLQ volume spikes, and an alert on DLQ row count surfaces the problem without an outage.
5. Replaying quarantined records is a separate batch job that reads orders_dlq, fixes or re-parses, and re-publishes — keeping the streaming path clean.

Output.

Input	Destination
valid JSON order	analytics.orders
truncated JSON	analytics.orders_dlq (raw + error)
wrong schema	analytics.orders_dlq (raw + error)
job status	keeps running throughout

Rule of thumb. Every production streaming job needs a dead-letter side output and an alert on its volume. A streaming pipeline must never crash on one bad record — quarantine it, keep running, and replay from the DLQ later.

Worked example — draining a job for a safe update

Detailed explanation. Updating a running streaming pipeline safely means draining it: Dataflow stops pulling new messages but lets in-flight windows complete and flush their results, so no data is lost and no window is left half-computed. Contrast with cancel, which discards in-flight state. Walk through the update procedure.

• Drain. Stop ingestion; finish and flush open windows; then stop.
• Cancel. Stop immediately; discard buffered/in-flight state (data loss).
• Update-in-place. --update carries state to a new job graph.

Question. Deploy new pipeline code without losing in-flight window state.

Input.

Parameter	Value
Running job	orders-streaming
Method	drain, then redeploy (or --update)
Guarantee	in-flight windows flushed
Anti-pattern	cancel (discards state)

Code.

# Option A — drain the old job (flushes in-flight windows), then start the new one
gcloud dataflow jobs drain "$(gcloud dataflow jobs list \
  --filter='name=orders-streaming AND state=Running' \
  --format='value(id)' --region=us-central1)" --region=us-central1

# start the new version after drain completes
python pipeline.py --runner=DataflowRunner --job_name=orders-streaming \
  --region=us-central1 --enable_streaming_engine

# Option B — update in place: carry state from old graph to new (no drain gap)
python pipeline.py --runner=DataflowRunner --job_name=orders-streaming \
  --region=us-central1 --enable_streaming_engine --update

Step-by-step explanation.

1. drain flips the job to stop reading from Pub/Sub while allowing already-ingested elements and open windows to complete and emit — so the per-minute windows currently open are flushed with correct final values before the job stops.
2. Unacked Pub/Sub messages are simply left in the subscription backlog during a drain, so the new job picks them up on start; nothing is lost, though there is a brief processing gap between drain completion and new-job startup.
3. --update avoids even that gap: Dataflow maps the running job's state (windows, dedup, timers) onto the new pipeline graph, provided the transform names and coders are compatible, so the job upgrades in place.
4. cancel is the anti-pattern for a graceful update — it stops immediately and discards in-flight window state, losing any not-yet-emitted aggregates. Reserve cancel for a job that is already broken.
5. The practical rule: use --update for compatible code changes, drain + redeploy when the graph changed incompatibly, and never cancel a healthy streaming job.

Output.

Method	In-flight windows	Data loss	Downtime
drain + redeploy	flushed	none	brief gap
update in place	carried over	none	none
cancel	discarded	yes	immediate

Rule of thumb. Update compatible changes with --update, use drain when the graph changed, and never cancel a healthy job — drain flushes open windows, cancel throws them away.

Worked example — capping cost with max workers and backlog alerts

Detailed explanation. Autoscaling is elastic, but unbounded elasticity is unbounded cost. Cap the worker pool and alert on backlog so the job scales for real spikes but cannot run away. Walk through the settings and the alert.

• Cap. --max_num_workers bounds the pool.
• Signal. Pub/Sub backlog (oldest unacked age) and Dataflow system lag.
• Alert. Backlog age > threshold means the cap is too low for current load.

Question. Configure autoscaling bounds and an alert that fires when the job cannot keep up.

Input.

Parameter	Value
Max workers	50
Streaming Engine	enabled
Alert metric	subscription oldest_unacked_message_age
Threshold	> 300 s for 5 min

Code.

# Bound the autoscaler and enable Streaming Engine for fast scaling
python pipeline.py --runner=DataflowRunner --job_name=orders-streaming \
  --region=us-central1 \
  --enable_streaming_engine \
  --autoscaling_algorithm=THROUGHPUT_BASED \
  --max_num_workers=50 \
  --number_of_worker_harness_threads=8

# Cloud Monitoring alert — backlog age too high => cap too low / job stuck
displayName: "Pub/Sub backlog too old"
conditions:
  - displayName: "oldest unacked > 300s"
    conditionThreshold:
      filter: >
        resource.type="pubsub_subscription"
        AND resource.label.subscription_id="orders-to-dataflow"
        AND metric.type="pubsub.googleapis.com/subscription/oldest_unacked_message_age"
      comparison: COMPARISON_GT
      thresholdValue: 300
      duration: 300s

Step-by-step explanation.

1. --autoscaling_algorithm=THROUGHPUT_BASED with --max_num_workers=50 lets Dataflow add workers up to 50 as backlog grows and remove them when it clears, bounding the maximum spend while still absorbing spikes.
2. --enable_streaming_engine keeps state off the workers, so scaling from 5 to 50 workers does not require shuffling large per-worker state — scale-up is fast enough to matter during a burst.
3. The alert watches oldest_unacked_message_age on the subscription: if the oldest unacked message is more than 5 minutes old for 5 minutes straight, the job is falling behind — usually because the worker cap is too low or a downstream sink is throttling.
4. Backlog age is a better signal than CPU for streaming, because a job can be CPU-idle yet backlogged when a sink (BigQuery quota) is the bottleneck; age captures end-to-end lateness.
5. The response runbook: raise max_num_workers if it is genuinely load, or investigate the sink/quota if workers are not CPU-bound — scaling workers against a sink bottleneck just wastes money.

Output.

Situation	Autoscaler	Alert
normal load	5–10 workers	quiet
10× spike	scales toward 50	quiet if it keeps up
sink throttled	pinned at 50, backlog grows	fires (age > 300 s)
spike over	scales back down	quiet

Rule of thumb. Always cap max_num_workers, enable Streaming Engine, and alert on backlog age (not CPU). Age is the metric that reveals whether you are actually keeping up end-to-end.

Data engineering interview question on exactly-once and autoscaling

A senior interviewer might ask: "You are streaming payment events from Pub/Sub through Dataflow into BigQuery for real-time revenue. The finance team needs exact numbers — no double-counting — and the pipeline must survive a 10× Black Friday spike and a bad upstream deploy without an outage. Walk me through exactly-once processing, the sink choice, autoscaling, and the failure-handling patterns."

Solution Using message-id dedup, an exactly-once BigQuery sink, bounded autoscaling, and a dead-letter branch

# revenue_pipeline.py — exactly-once processing + DLQ + bounded autoscaling
import apache_beam as beam
import json
from apache_beam.transforms.trigger import AfterWatermark, AccumulationMode

class Parse(beam.DoFn):
    def process(self, raw):
        try:
            yield json.loads(raw.decode("utf-8"))
        except Exception as e:
            yield beam.pvalue.TaggedOutput("dead", {"raw": str(raw), "error": str(e)})

with beam.Pipeline(options=opts) as p:               # opts: streaming, streaming_engine, max_num_workers=100
    parsed = (
        p | "Read" >> beam.io.ReadFromPubSub(
                subscription="projects/acme/subscriptions/payments-sub",
                timestamp_attribute="event_ts")       # Dataflow dedups by message_id
          | "Parse" >> beam.ParDo(Parse()).with_outputs("dead", main="good")
    )

    (parsed.good
        | "Win" >> beam.WindowInto(
              beam.window.FixedWindows(60),
              trigger=AfterWatermark(),
              accumulation_mode=AccumulationMode.ACCUMULATING)
        | "KV" >> beam.Map(lambda e: (e["region"], e["cents"]))
        | "Sum" >> beam.CombinePerKey(sum)
        | "Fmt" >> beam.Map(lambda kv: {"region": kv[0], "cents": kv[1]})
        | "BQ" >> beam.io.WriteToBigQuery(
              "acme:analytics.revenue_min",
              method=beam.io.WriteToBigQuery.Method.STORAGE_WRITE_API,  # exactly-once sink
              write_disposition=beam.io.BigQueryDisposition.WRITE_APPEND)
    )

    parsed.dead | "DLQ" >> beam.io.WriteToBigQuery("acme:analytics.payments_dlq")

# Launch with bounded autoscaling + Streaming Engine
python revenue_pipeline.py --runner=DataflowRunner --job_name=revenue \
  --region=us-central1 --streaming --enable_streaming_engine \
  --autoscaling_algorithm=THROUGHPUT_BASED --max_num_workers=100

Step-by-step trace.

Step	Mechanism	Guarantee
1	Pub/Sub source	at-least-once delivery
2	Dataflow message_id dedup	each message processed once internally
3	checkpointed state + windows	exactly-once processing on retry
4	Storage Write API sink	exactly-once output to BigQuery
5	dead-letter branch	bad records quarantined, job survives
6	bounded autoscaling	absorbs 10× spike up to 100 workers

After deployment, Pub/Sub delivers at-least-once, Dataflow de-duplicates on message_id and checkpoints state so a worker crash replays deterministically without double-applying to the per-region sums, and the BigQuery Storage Write API commits output exactly once by stream offset — so finance's revenue numbers are exact end to end. The dead-letter branch keeps a malformed upstream deploy from crashing the job, and throughput-based autoscaling scales to 100 workers for the Black Friday spike, then back down.

Output:

region	cents (window)	correctness
us-west	4,201,900	exactly-once, no double count
us-east	3,880,250	exactly-once, no double count
(bad record)	→ payments_dlq	quarantined, job alive

Why this works — concept by concept:

• Exactly-once processing — Dataflow's message_id dedup plus deterministic checkpointing means a retried or redelivered element affects the per-region sums exactly once, turning an at-least-once source into correct internal aggregates.
• Storage Write API sink — commits rows to BigQuery by stream offset so a pipeline retry cannot append the same window twice; without an exactly-once sink, output-side duplicates are still possible even with once-processing.
• Dead-letter branch — a tagged side output routes malformed records to a DLQ table, so a bad upstream deploy raises DLQ volume and an alert instead of crashing the streaming job.
• Bounded throughput autoscaling — scales workers to backlog up to a hard cap of 100, absorbing a 10× spike while bounding cost; Streaming Engine keeps state off workers so scale-up is fast.
• Cost — steady state runs a handful of workers; the cap bounds worst-case spend; exactly-once processing and the Storage Write API add modest overhead versus at-least-once. Net cost is O(events) with one shuffle per window — the price of correct, elastic real-time revenue that finance can trust.

Streaming
Topic — streaming
Streaming exactly-once and autoscaling problems

Practice →

Data eng
Topic — data-processing
Data-processing problems on fault-tolerant pipelines

Practice →

---

Cheat sheet — Pub/Sub and Dataflow streaming recipes

• Topic vs subscription. Publish to a topic (a fan-out point with no storage of its own); consume through a subscription (a durable per-consumer cursor with its own backlog and ack state). Two subscriptions on one topic each get every message. Always create the subscription before the first publish — a message published to a topic with no subscription is dropped.
• Ack lifecycle. A delivered message stays outstanding until ack() or the ack deadline (default 10 s, max 600 s) expires; expiry triggers redelivery (this is what makes delivery at-least-once). Extend the lease with modifyAckDeadline for slow handlers; nack() forces immediate redelivery; a dead-letter topic (--max-delivery-attempts 5–100) quarantines poison messages.
• Delivery guarantees. At-least-once by default (duplicates + no order). Exactly-once delivery is a per-subscription opt-in (--enable-exactly-once-delivery) that stops redelivery of acked messages but does not dedup publisher-retry duplicates. Always keep an idempotent consumer keyed on a business id, not message_id.
• Ordering. Enable enable_message_ordering on both publisher and subscription; messages sharing an ordering_key in one region arrive in publish order. Ordering caps per-key throughput, so never use one constant key for a whole stream. A failed ordered publish pauses the key until resume_publish.
• Beam vocabulary. PCollection (bounded/unbounded dataset), PTransform (operation), ParDo/DoFn (per-element), GroupByKey (shuffle), CoGroupByKey (join), Combine/CombinePerKey (associative aggregation), Flatten (union). Beam is the model; Dataflow is the managed runner; the same graph runs batch or streaming.
• Prefer Combine over GroupByKey. For associative aggregations use CombinePerKey so Dataflow lifts the combiner before the shuffle (combiner lifting), moving far less data. Reserve GroupByKey for when you genuinely need all values per key.
• Windowing. FixedWindows(n) (tumbling, non-overlapping) for per-interval metrics; SlidingWindows(size, period) (overlapping) for moving averages; Sessions(gap) (data-driven, auto-merging) for activity bursts; GlobalWindows for batch. Window on event time (set timestamp_attribute on ReadFromPubSub), never arrival time.
• Watermark + triggers. The watermark is Beam's estimate that event time ≤ W has all arrived; AfterWatermark() fires a window when it passes the window end. Add early=AfterProcessingTime(n) for speculative dashboards and late=... with allowed_lateness for corrections. Choose ACCUMULATING (each firing = full total, sink overwrites) vs DISCARDING (deltas) carefully — wrong choice double-counts.
• Exactly-once processing. Dataflow dedups Pub/Sub by message_id and checkpoints state deterministically, so aggregates are once even over at-least-once delivery. Pair with an exactly-once sink (BigQuery Storage Write API) or output duplicates are still possible. Non-idempotent DoFn side effects (external API calls) still need their own idempotency keys.
• Autoscaling. --autoscaling_algorithm=THROUGHPUT_BASED scales workers to Pub/Sub backlog; always set --max_num_workers to bound cost and --enable_streaming_engine for fast scaling with off-worker state. Alert on oldest_unacked_message_age (backlog age), not CPU — age reveals whether you keep up end-to-end.
• Streaming patterns. Dead-letter side output (TaggedOutput + with_outputs) so one bad record never crashes the job; side-input enrichment (AsDict) for small dimensions without a shuffle; drain (flush in-flight windows) or --update (carry state) for safe deploys — never cancel a healthy job.
• Replay. Snapshot a subscription before seek; seek --time rewinds unacked state within the retention window (default 7 days, max 31); replay only into idempotent sinks because seek redelivers, it does not dedup.

Frequently asked questions

What is the difference between Pub/Sub and Dataflow?

Pub/Sub and Dataflow are complementary, not competing: Pub/Sub is a durable, planet-scale messaging bus that decouples publishers from consumers via topics and subscriptions, guaranteeing at-least-once delivery of every message. Dataflow is a fully-managed runner for Apache Beam pipelines that consume those messages (usually from a pull subscription) and transform, window, aggregate, join, and load them — with autoscaling, exactly-once processing, and event-time windowing built in. A typical GCP streaming pipeline is publisher → Pub/Sub topic → subscription → Dataflow (Beam) → BigQuery. Pub/Sub moves the messages; Dataflow computes over them.

What is the difference between a topic and a subscription in Pub/Sub?

A topic is a named channel that publishers send messages to; it has no consumers of its own and stores nothing durably except to feed its subscriptions. A subscription is a durable named cursor attached to exactly one topic that accumulates its own backlog of undelivered messages and tracks its own acknowledgement state. The crucial consequences: two subscriptions on the same topic each receive every message independently (fan-out), and a subscription created after a message was published never sees that message — there is no backlog time-travel. Always create the subscription before the first publish, or those early messages are silently dropped.

Does Pub/Sub guarantee exactly-once delivery and ordering?

By default Pub/Sub is at-least-once and unordered — a message can be delivered more than once and in any order. You can enable exactly-once delivery per subscription, which stops redelivery of a successfully-acknowledged message, but it does not remove duplicates created by publisher retries (those are genuinely distinct messages), so idempotent consumers keyed on a business id are still recommended. You can separately enable message ordering: messages sharing an ordering_key, published to a single region, are delivered in publish order — at the cost of capping per-key throughput. Both features are explicit opt-ins, never assume them.

How does Dataflow achieve exactly-once when Pub/Sub is at-least-once?

Dataflow provides exactly-once processing (a different guarantee from Pub/Sub's exactly-once delivery) by de-duplicating incoming Pub/Sub messages on their message_id, checkpointing pipeline state, and replaying deterministically after a worker failure so retried work never double-applies to aggregates. To make the whole path exactly-once you also need an exactly-once sink — the BigQuery Storage Write API commits rows by stream offset so a retry cannot append a window twice. The one gap: non-idempotent external side effects inside a DoFn (calling a payment API, sending an email) are not covered by exactly-once processing and still need their own idempotency keys.

What is the difference between a watermark and a window in Apache Beam?

A window is a slice of the stream that groups elements for aggregation — fixed (tumbling), sliding (overlapping), session (gap-based), or global. A watermark is the runner's continuously-advancing estimate that all data with event time up to time W has arrived; it is what tells the runner a window is complete enough to emit its result. When the watermark passes a window's end, the default AfterWatermark trigger fires that window. Data arriving with an event time behind the watermark is late: it is dropped unless you set allowed_lateness, which keeps the window open to accept and re-fire on stragglers. Windows define what to group; watermarks and triggers define when to emit.

When should I use fixed, sliding, or session windows?

Use fixed (tumbling) windows for regular interval metrics where each event belongs to exactly one bucket — revenue per minute, requests per hour. Use sliding (hopping) windows for moving/rolling aggregations where windows overlap — a 5-minute average recomputed every minute — because each element contributes to several windows. Use session windows for activity defined by gaps of inactivity rather than fixed clock boundaries — user sessions, device bursts, support conversations — where a window grows as long as events keep arriving within the gap and closes after silence; Beam automatically merges session windows as new events fill gaps, which is why sessionization is impractical to hand-roll. Global windows are the batch default and need a custom trigger to be useful in streaming.

Practice on PipeCode

• Drill the streaming practice library → for the Pub/Sub delivery, ordering, exactly-once, and Dataflow autoscaling problems senior interviewers love.
• Sharpen event-time intuition on the real-time analytics practice library → for windowing, watermarks, late-data, and trigger-accumulation scenarios.
• Rehearse the transforms on the data-processing practice library → for Beam PTransforms, stateful ParDo, joins, and dead-letter fault-tolerance patterns.
• Stack the prerequisites against PipeCode's broader 450+ data-engineering catalogue to anchor the Pub/Sub-and-Dataflow decision model against real graded inputs.

Lock in Pub/Sub and Dataflow muscle memory

Docs explain the services. PipeCode drills explain the decisions — when at-least-once needs an idempotency ledger, when ordering keys cap throughput, when a window fires versus stays open for late data, when exactly-once processing still needs an exactly-once sink, and when to drain instead of cancel. 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 real-time analytics problems →