In this post we are going to cover Pub/Sub to BigQuery and Dataflow. These are 7 real Pub/Sub interview questions for data engineers, asked the way they come up on a real project, and the last one is a scenario: clickstream events must land in BigQuery within a minute for dashboards, and raw copies must be kept for a year.
Most Pub/Sub pipelines end in a table or a data lake. Pub/Sub can write there by itself, with a BigQuery subscription or a Cloud Storage subscription, and Dataflow takes over when we need real processing. Interviewers ask about this because the choice decides how much code the team runs, and because duplicates and schema mismatches show up the moment data lands in a table.
Part 9 of the Real Time Pub/Sub Interview Questions and Answers series on TechBrothersIT.
- The short answer
- Let's set up the project
- Question 1: Your manager wants events in BigQuery with no code. How do BigQuery subscriptions work?
- Question 2: A teammate asks when to use a BigQuery subscription and when to use Dataflow. How do you choose?
- Question 3: Raw events must also be kept in a data lake. How do Cloud Storage subscriptions work?
- Question 4: Events need cleaning and windowed totals before loading. How does Dataflow read from Pub/Sub?
- Question 5: The BigQuery table schema and the messages don't match. What happens, and how do you handle it?
- Question 6: Someone asks how to avoid duplicates in BigQuery from Pub/Sub. What do you do?
- Scenario 7: Clickstream events must land in BigQuery within a minute for dashboards, and raw copies must be kept for a year. What is your design?
- The full script
The short answer
A BigQuery subscription writes messages straight into a table with no code; a Cloud Storage subscription writes batched text or Avro files to a bucket. Use Dataflow when you need cleaning, joins or event-time windows. Export subscriptions are at-least-once: write metadata, add a dead-letter topic, and remove duplicates by key.
Let's set up the project
We run Google's official Pub/Sub emulator locally, with small Python scripts that use the google-cloud-pubsub client, so every output is real. BigQuery, Cloud Storage and Dataflow don't run on it, so their setup is shown as code only, clearly marked. setup.py creates a clicks topic and two pull subscriptions, clicks-window and clicks-loader. publish.py publishes six click events, and table.json starts empty: it stands in for a table later.
from google.cloud import pubsub_v1
project = "tb-shop-dev"
publisher = pubsub_v1.PublisherClient()
subscriber = pubsub_v1.SubscriberClient()
topic = publisher.topic_path(project, "clicks")
publisher.create_topic(name=topic)
for name in ["clicks-window", "clicks-loader"]:
sub = subscriber.subscription_path(project, name)
subscriber.create_subscription(name=sub, topic=topic)
print("created", sub)import json
from google.cloud import pubsub_v1
clicks = [ # event_id, user, page, event time (UTC)
("e-1001", "u-7", "home", "2026-10-28T09:00:05Z"),
("e-1002", "u-7", "cart", "2026-10-28T09:00:40Z"),
("e-1003", "u-9", "home", "2026-10-28T09:01:10Z"),
("e-1004", "u-9", "checkout", "2026-10-28T09:01:30Z"),
("e-1005", "u-3", "home", "2026-10-28T09:01:45Z"),
("e-1006", "u-3", "cart", "2026-10-28T09:00:50Z"), # late
]
publisher = pubsub_v1.PublisherClient()
topic = publisher.topic_path("tb-shop-dev", "clicks")
for eid, user, page, ts in clicks:
data = json.dumps({"event_id": eid, "user": user,
"page": page, "ts": ts}).encode()
future = publisher.publish(topic, data, event_type="click",
event_id=eid, event_ts=ts)
print("published", eid, page, "-> message", future.result()){}Question 1: Your manager wants events in BigQuery with no code. How do BigQuery subscriptions work?
A BigQuery subscription is an export subscription. Pub/Sub itself writes every message into a BigQuery table, through the BigQuery Storage Write API. No subscriber code, no Dataflow job, nothing to run.
First, let's create the sample data with python setup.py and python publish.py.
$ python setup.py
created projects/tb-shop-dev/subscriptions/clicks-window
created projects/tb-shop-dev/subscriptions/clicks-loader
$ python publish.py
published e-1001 home -> message 1
published e-1002 cart -> message 2
published e-1003 home -> message 3
published e-1004 checkout -> message 4
published e-1005 home -> message 5
published e-1006 cart -> message 6We get the two subscriptions, and six clicks, e-1001 to e-1006, published as messages 1 to 6. Each carries attributes: event_type, event_id and event_ts, the event time.
The BigQuery subscription itself is one gcloud command, shown as code only: the emulator accepts this config, but never writes to BigQuery. We point it at an existing table, and turn on write metadata.
# code only: the emulator accepts this but writes nothing
gcloud pubsub subscriptions create clicks-to-bq \
--topic=clicks \
--bigquery-table=tb-shop-prod:analytics.clicks_raw \
--write-metadataWith no schema option, the whole payload lands in a data column. Write metadata adds the subscription name, message ID, publish time and attributes as columns. We want them for debugging and for removing duplicates.
| Table column | Type |
|---|---|
| subscription_name | STRING |
| message_id | STRING |
| publish_time | TIMESTAMP |
| data | BYTES, STRING or JSON |
| attributes | STRING or JSON |
The Pub/Sub service agent needs roles/bigquery.dataEditor on the table, and add a dead-letter topic, because failed writes are retried. Billing is per tebibyte, with no separate BigQuery ingestion charge, and the free tier does not apply.
Question 2: A teammate asks when to use a BigQuery subscription and when to use Dataflow. How do you choose?
If each message already looks like the row we want, use a BigQuery subscription. If we need to clean, join, enrich, or aggregate over time windows, use Dataflow.
| BigQuery subscription | Dataflow (Apache Beam) | |
|---|---|---|
| Code to run | none: it is a subscription | a pipeline you deploy |
| Transforms | per message: JavaScript SMT | anything: clean, join, enrich |
| Windows, aggregates | no | yes, by event time |
| Duplicates | at-least-once: dedupe in BigQuery | exactly-once mode: by message ID |
| You operate | nothing | workers: size, update, pay |
A BigQuery subscription has no code. Light per-message changes fit in a single message transform, a small JavaScript function. But there are no windows and no joins, and it is at-least-once, so we remove duplicates in BigQuery.
Dataflow runs an Apache Beam pipeline. It windows by event time, joins streams, and its exactly-once streaming mode removes duplicates by message ID. The price is workers that we size, update and pay for.
In between sits the Google-provided Pub/Sub to BigQuery template: a ready Dataflow job with a user-defined function for small fixes (JavaScript, or a Python UDF variant), and a dead-letter table for rows that fail.
A Dataflow job that only copies messages into a table. That's a BigQuery subscription with more to run.
Question 3: Raw events must also be kept in a data lake. How do Cloud Storage subscriptions work?
It's the other export type: Pub/Sub batches messages into files in our bucket, with no code. Code only again: the emulator stores the config, but writes no files.
# code only: the emulator stores this but writes no files
gcloud pubsub subscriptions create clicks-archive \
--topic=clicks \
--cloud-storage-bucket=tb-shop-clicks-raw \
--cloud-storage-file-prefix=clicks/ \
--cloud-storage-file-suffix=.avro \
--cloud-storage-output-format=avro \
--cloud-storage-write-metadataFirst, pick the format. Text writes just the payloads, one per line. Avro writes a record per message with the payload as bytes. Write metadata adds the subscription name, message ID, publish time and attributes, and use topic schema writes the payload's own fields. For a raw archive, I use Avro.
Then the batching. A new file starts when a limit is hit, and a file appears only when it is finalized.
| New file when | Limit |
|---|---|
| max duration | 1 to 10 min, default 5 |
| max bytes | 1 KB to 10 GiB |
| max messages | at least 1,000 |
File names are our prefix, the UTC date and time, a unique ID, and our suffix. The bucket must already exist without Requester Pays, and the service agent needs roles/storage.objectCreator and roles/storage.legacyBucketReader.
If analysts ask for Parquet, it's not an output format here. Use a BigQuery subscription instead, which can also write to BigQuery tables for Apache Iceberg.
Question 4: Events need cleaning and windowed totals before loading. How does Dataflow read from Pub/Sub?
Dataflow reads with its own Pub/Sub connector, not the client library: ReadFromPubSub in Python, here on a subscription made for this pipeline. Given a topic, it creates a temporary subscription per pipeline.
- Own connector - not the client library
- One subscription per pipeline - a topic gets a temporary one
id_label- publisher duplicates within 10 mintimestamp_attribute- event time for windows
id_label names an attribute for removing publisher duplicates sent within 10 minutes. timestamp_attribute gives the event time, so windows use when the click happened, not when it arrived. Then parse, filter, one-minute windows, a count per page, and BigQuery through the Storage Write API. This pipeline is code only.
# code only: Apache Beam, run with --runner=DataflowRunner
import json
import apache_beam as beam
from apache_beam.options.pipeline_options import PipelineOptions
from apache_beam.transforms import window
SUB = "projects/tb-shop-prod/subscriptions/clicks-dataflow"
with beam.Pipeline(options=PipelineOptions(streaming=True)) as p:
(p
| beam.io.ReadFromPubSub(subscription=SUB,
id_label="event_id",
timestamp_attribute="event_ts")
| beam.Map(json.loads)
| beam.Filter(lambda e: e.get("page")) # cleaning
| beam.Map(lambda e: (e["page"], 1))
| beam.WindowInto(window.FixedWindows(60)) # 1 minute
| beam.CombinePerKey(sum)
| beam.Map(lambda kv, w=beam.DoFn.WindowParam: {
"minute": w.start,
"page": kv[0], "clicks": kv[1]})
| beam.io.WriteToBigQuery(
"tb-shop-prod:analytics.page_clicks_1m",
schema="minute:TIMESTAMP,page:STRING,clicks:INTEGER",
method="STORAGE_WRITE_API"))What runs here is a plain Python stand-in, not Dataflow: it pulls the six clicks and counts them per minute of the event_ts attribute. We run python window_count.py.
# NOT Dataflow: a plain Python stand-in for 1-minute windows
import json
from collections import Counter
from google.cloud import pubsub_v1
subscriber = pubsub_v1.SubscriberClient()
sub = subscriber.subscription_path("tb-shop-dev", "clicks-window")
resp = subscriber.pull(subscription=sub, max_messages=50, timeout=10)
counts = Counter()
for m in resp.received_messages:
event_ts = m.message.attributes["event_ts"] # event time
window = event_ts[11:16] # "09:00" bucket
page = json.loads(m.message.data)["page"]
counts[(window, page)] += 1
ack_ids = [m.ack_id for m in resp.received_messages]
subscriber.acknowledge(subscription=sub, ack_ids=ack_ids)
for (window, page), n in sorted(counts.items()):
print("window", window, "page", page, "clicks", n)$ python window_count.py
window 09:00 page cart clicks 2
window 09:00 page home clicks 1
window 09:01 page checkout clicks 1
window 09:01 page home clicks 2Cart gets clicks 2 in the 09:00 window. The late cart click, e-1006, was published last, but its event time is 09:00:50. Real Dataflow adds watermarks, late data, state and autoscaling.
Never share one subscription between pipelines, and don't turn on exactly-once delivery or ordering for it. Dataflow removes duplicates itself, and acks once the first stage is persisted.
Question 5: The BigQuery table schema and the messages don't match. What happens, and how do you handle it?
With a schema option on, a message with a field the table doesn't have is not written. It stays in the subscription backlog, unless we set drop unknown fields, or a dead-letter topic.
- Extra field - not written, stays in the backlog
- Failed write - a nack, then retried
- Max attempts - dead-letter topic, error in an attribute
- DLQ set - no backoff for schema errors
The better fix is to stop bad data at the door. Let's give a new topic, clicks-v2, an Avro schema and publish two clicks. We run python schema_topic.py and python publish_v2.py.
{"type": "record", "name": "Click", "fields": [
{"name": "event_id", "type": "string"},
{"name": "user", "type": "string"},
{"name": "page", "type": "string"},
{"name": "ts", "type": "string"}]}from google.cloud import pubsub_v1
from google.pubsub_v1.types import Encoding, Schema
project = "tb-shop-dev"
schemas = pubsub_v1.SchemaServiceClient()
path = schemas.schema_path(project, "click")
schemas.create_schema(request={
"parent": f"projects/{project}", "schema_id": "click",
"schema": Schema(name=path, type_=Schema.Type.AVRO,
definition=open("click.avsc").read())})
publisher = pubsub_v1.PublisherClient()
topic = publisher.topic_path(project, "clicks-v2")
publisher.create_topic(request={"name": topic, "schema_settings": {
"schema": path, "encoding": Encoding.JSON}})
print("topic clicks-v2 -> schema", path)import json
from google.api_core.exceptions import InvalidArgument
from google.cloud import pubsub_v1
publisher = pubsub_v1.PublisherClient()
topic = publisher.topic_path("tb-shop-dev", "clicks-v2")
events = [{"event_id": "e-2001", "user": "u-4", "page": "home",
"ts": "2026-10-28T09:02:00Z"},
{"event_id": "e-2002", "user": "u-4", "page": 42, # number
"ts": "2026-10-28T09:02:05Z"}]
for e in events:
data = json.dumps(e).encode()
try:
mid = publisher.publish(topic, data).result()
print("published", e["event_id"], "-> message", mid)
except InvalidArgument as err:
print("rejected", e["event_id"], "->", type(err).__name__,
str(err).splitlines()[0])$ python schema_topic.py
topic clicks-v2 -> schema projects/tb-shop-dev/schemas/click
$ python publish_v2.py
published e-2001 -> message 7
rejected e-2002 -> InvalidArgument 400 Could not parse messageThe good click is published e-2001 -> message 7. The click with page as a number is rejected; on the emulator the error is InvalidArgument 400 Could not parse message.
Then use topic schema maps the Avro fields to the table columns, so publishers and the table share one contract. Here is that subscription, code only.
# code only: BigQuery subscription with the topic schema
gcloud pubsub subscriptions create clicks-to-bq \
--topic=clicks-v2 \
--bigquery-table=tb-shop-prod:analytics.clicks \
--use-topic-schema --write-metadata \
--drop-unknown-fields \
--dead-letter-topic=clicks-bq-dlq \
--max-delivery-attempts=5Add the column to the table first, then the field to the schema: without drop unknown fields, the table must hold every topic schema field. And allow time for table changes to be picked up.
Question 6: Someone asks how to avoid duplicates in BigQuery from Pub/Sub. What do you do?
Accept that they happen, and remove them by key. A BigQuery subscription is at-least-once, like a normal pull subscription.
loader.py skips message IDs it already loaded, then acks. table.json stands in for the table. Let's load, and crash before the ack, with python loader.py crash.
import json, sys
from google.cloud import pubsub_v1
subscriber = pubsub_v1.SubscriberClient()
sub = subscriber.subscription_path("tb-shop-dev", "clicks-loader")
table = json.load(open("table.json")) # stands in for BigQuery
resp = subscriber.pull(subscription=sub, max_messages=50, timeout=10)
msgs = resp.received_messages
new = 0
for m in msgs:
mid = m.message.message_id
if mid not in table: # dedupe on the message ID
table[mid] = json.loads(m.message.data)
new += 1
json.dump(table, open("table.json", "w"), indent=1)
ids = sorted(int(m.message.message_id) for m in msgs)
print("pulled IDs", ids, "inserted", new, "skipped", len(msgs) - new)
if sys.argv[1:] == ["crash"]:
sys.exit("crashed before the ack")
ack_ids = [m.ack_id for m in msgs]
subscriber.acknowledge(subscription=sub, ack_ids=ack_ids)
print("acked", len(msgs))$ python loader.py crash
pulled IDs [1, 2, 3, 4, 5, 6] inserted 6 skipped 0
crashed before the ackIt prints inserted 6 skipped 0, then crashed before the ack. Now we wait past the 10 second ack deadline and run python loader.py again.
$ sleep 13 && python loader.py
pulled IDs [1, 2, 3, 4, 5, 6] inserted 0 skipped 6
acked 6The same message IDs, [1, 2, 3, 4, 5, 6], come back. The loader prints inserted 0 skipped 6: no duplicate rows.
But message IDs don't catch everything. The publisher timed out and sent e-1003 again. We run python resend.py, python loader.py and python report.py.
import json
from google.cloud import pubsub_v1
# the publisher timed out on e-1003 and sent it again
publisher = pubsub_v1.PublisherClient()
topic = publisher.topic_path("tb-shop-dev", "clicks")
ts = "2026-10-28T09:01:10Z"
data = json.dumps({"event_id": "e-1003", "user": "u-9",
"page": "home", "ts": ts}).encode()
future = publisher.publish(topic, data, event_type="click",
event_id="e-1003", event_ts=ts)
print("re-published e-1003 -> message", future.result())import json
from collections import defaultdict
table = json.load(open("table.json"))
by_event = defaultdict(list)
for mid, row in table.items():
by_event[row["event_id"]].append(mid)
print("rows", len(table), "| unique message IDs", len(table),
"| unique event IDs", len(by_event))
for eid, mids in sorted(by_event.items()):
if len(mids) > 1:
print("duplicate", eid, "message IDs", sorted(mids, key=int))$ python resend.py
re-published e-1003 -> message 9
$ python loader.py
pulled IDs [9] inserted 1 skipped 0
acked 1
$ python report.py
rows 7 | unique message IDs 7 | unique event IDs 6
duplicate e-1003 message IDs ['3', '9']The resend got a new message ID, 9, so the loader inserts it. The report shows rows 7, but unique event IDs 6, and duplicate e-1003 message IDs ['3', '9']. On the emulator, IDs are a simple counter, so don't read anything into their format.
So in BigQuery, dedupe on the business key. With the topic schema and write metadata, a view keeps the first row per event ID, by publish time.
-- code only: BigQuery, keep the first row per event
CREATE OR REPLACE VIEW analytics.clicks_clean AS
SELECT * FROM analytics.clicks
WHERE TRUE
QUALIFY ROW_NUMBER() OVER (
PARTITION BY event_id ORDER BY publish_time) = 1;Dataflow's exactly-once mode removes redeliveries by message ID, and with an ID label, publisher retries too. For change data, a BigQuery subscription can upsert by primary key with _CHANGE_TYPE.
Scenario 7: Clickstream events must land in BigQuery within a minute for dashboards, and raw copies must be kept for a year. What is your design?
One clicks topic with an Avro schema, and two export subscriptions on it. Each subscription gets every click, and there is no subscriber code to run.
The BigQuery subscription writes through the Storage Write API with the topic schema and metadata, so dashboards query fresh rows. The Cloud Storage subscription writes Avro files with metadata. A file closes after 5 minutes by default, or sooner when it fills up.
Here it is as Terraform, code only.
# code only: Terraform, google provider
resource "google_pubsub_subscription" "clicks_bq" {
name = "clicks-to-bq"
topic = google_pubsub_topic.clicks.id
bigquery_config {
table = "tb-shop-prod.analytics.clicks"
use_topic_schema = true
write_metadata = true
}
dead_letter_policy {
dead_letter_topic = google_pubsub_topic.clicks_dlq.id
max_delivery_attempts = 5
}
}
resource "google_pubsub_subscription" "clicks_gcs" {
name = "clicks-archive"
topic = google_pubsub_topic.clicks.id
cloud_storage_config {
bucket = "tb-shop-clicks-raw"
filename_prefix = "clicks/"
filename_suffix = ".avro"
avro_config { write_metadata = true }
}
}The BigQuery subscription has a dead-letter policy. Give the archive one too, and grant the service agent its roles.
- Dead-letter policy - on both subscriptions
- Service agent roles - BigQuery and Cloud Storage
- 1 year in the bucket - Pub/Sub retention: max 31 days
- Alert -
oldest_unacked_message_age_by_region,dead_letter_message_count
The year lives in Cloud Storage, not in Pub/Sub. Subscription and topic retention top out at 31 days: that's a replay buffer, not an archive.
Alert on the oldest unacked message age by region for the BigQuery subscription, and on the dead-letter message count. If the age passes a minute, the dashboards are late.
Key points to remember
- BigQuery subscription: no code, Storage Write API, write metadata
- Dataflow for cleaning, joins and event-time windows
- Cloud Storage subscription: batched text or Avro files
- Schema on the topic, drop unknown fields or a dead-letter topic
- At-least-once: dedupe by message ID and by business key
The full script
This is a Python script that writes every file and runs every command above in order against the Pub/Sub emulator: start the emulator, run pip install google-cloud-pubsub first, and run it in an empty folder.
Show the full script that writes the files and runs every step
# ========================================================================
# Pub/Sub Interview Q&A #9 - Pub/Sub to BigQuery and Dataflow
# TechBrothersIT. Names and sample data are invented for teaching.
# Every step below was run on the official Google Cloud Pub/Sub emulator (gcloud component, version 0.8.37)
# with google-cloud-pubsub 2.43.0; the outputs in the video and the blog post are the real outputs.
# To run it yourself (Linux, macOS or WSL):
# 1. gcloud components install pubsub-emulator then gcloud beta emulators pubsub start --project=tb-shop-dev
# 2. In a second terminal: export PUBSUB_EMULATOR_HOST=localhost:8085 PUBSUB_PROJECT_ID=tb-shop-dev
# 3. pip install google-cloud-pubsub then python PubSub_Interview_QA_09_Code.py in an empty folder.
# Steps that restart the emulator say so in a comment: stop and start it by hand at that point.
# ========================================================================
import os, subprocess, time
def write(path, text):
if text is None:
if os.path.exists(path): os.remove(path)
return
os.makedirs(os.path.dirname(path) or ".", exist_ok=True)
with open(path, "w", encoding="utf-8") as f:
f.write(text.strip("\n") + "\n")
def run(cmd, expect_error=False, allow_error=False, env=None):
print("\n$ " + cmd, flush=True)
rc = subprocess.run(cmd, shell=True, env=dict(os.environ, **(env or {}))).returncode
if rc != 0 and not (expect_error or allow_error):
raise SystemExit("stopped: this command was expected to succeed")
# Step 0: the starting files
write('setup.py', """\
from google.cloud import pubsub_v1
project = "tb-shop-dev"
publisher = pubsub_v1.PublisherClient()
subscriber = pubsub_v1.SubscriberClient()
topic = publisher.topic_path(project, "clicks")
publisher.create_topic(name=topic)
for name in ["clicks-window", "clicks-loader"]:
sub = subscriber.subscription_path(project, name)
subscriber.create_subscription(name=sub, topic=topic)
print("created", sub)
""")
write('publish.py', """\
import json
from google.cloud import pubsub_v1
clicks = [ # event_id, user, page, event time (UTC)
("e-1001", "u-7", "home", "2026-10-28T09:00:05Z"),
("e-1002", "u-7", "cart", "2026-10-28T09:00:40Z"),
("e-1003", "u-9", "home", "2026-10-28T09:01:10Z"),
("e-1004", "u-9", "checkout", "2026-10-28T09:01:30Z"),
("e-1005", "u-3", "home", "2026-10-28T09:01:45Z"),
("e-1006", "u-3", "cart", "2026-10-28T09:00:50Z"), # late
]
publisher = pubsub_v1.PublisherClient()
topic = publisher.topic_path("tb-shop-dev", "clicks")
for eid, user, page, ts in clicks:
data = json.dumps({"event_id": eid, "user": user,
"page": page, "ts": ts}).encode()
future = publisher.publish(topic, data, event_type="click",
event_id=eid, event_ts=ts)
print("published", eid, page, "-> message", future.result())
""")
write('table.json', """\
{}
""")
# Step 1: A clicks topic, two pull subscriptions, six click events with attributes
run('python setup.py')
run('python publish.py')
# Step 2: NOT Dataflow: a plain Python stand-in that counts clicks per 1-minute event-time window
write('window_count.py', """\
# NOT Dataflow: a plain Python stand-in for 1-minute windows
import json
from collections import Counter
from google.cloud import pubsub_v1
subscriber = pubsub_v1.SubscriberClient()
sub = subscriber.subscription_path("tb-shop-dev", "clicks-window")
resp = subscriber.pull(subscription=sub, max_messages=50, timeout=10)
counts = Counter()
for m in resp.received_messages:
event_ts = m.message.attributes["event_ts"] # event time
window = event_ts[11:16] # "09:00" bucket
page = json.loads(m.message.data)["page"]
counts[(window, page)] += 1
ack_ids = [m.ack_id for m in resp.received_messages]
subscriber.acknowledge(subscription=sub, ack_ids=ack_ids)
for (window, page), n in sorted(counts.items()):
print("window", window, "page", page, "clicks", n)
""")
run('python window_count.py')
# Step 3: A topic with an Avro schema validates every publish
write('click.avsc', """\
{"type": "record", "name": "Click", "fields": [
{"name": "event_id", "type": "string"},
{"name": "user", "type": "string"},
{"name": "page", "type": "string"},
{"name": "ts", "type": "string"}]}
""")
write('schema_topic.py', """\
from google.cloud import pubsub_v1
from google.pubsub_v1.types import Encoding, Schema
project = "tb-shop-dev"
schemas = pubsub_v1.SchemaServiceClient()
path = schemas.schema_path(project, "click")
schemas.create_schema(request={
"parent": f"projects/{project}", "schema_id": "click",
"schema": Schema(name=path, type_=Schema.Type.AVRO,
definition=open("click.avsc").read())})
publisher = pubsub_v1.PublisherClient()
topic = publisher.topic_path(project, "clicks-v2")
publisher.create_topic(request={"name": topic, "schema_settings": {
"schema": path, "encoding": Encoding.JSON}})
print("topic clicks-v2 -> schema", path)
""")
write('publish_v2.py', """\
import json
from google.api_core.exceptions import InvalidArgument
from google.cloud import pubsub_v1
publisher = pubsub_v1.PublisherClient()
topic = publisher.topic_path("tb-shop-dev", "clicks-v2")
events = [{"event_id": "e-2001", "user": "u-4", "page": "home",
"ts": "2026-10-28T09:02:00Z"},
{"event_id": "e-2002", "user": "u-4", "page": 42, # number
"ts": "2026-10-28T09:02:05Z"}]
for e in events:
data = json.dumps(e).encode()
try:
mid = publisher.publish(topic, data).result()
print("published", e["event_id"], "-> message", mid)
except InvalidArgument as err:
print("rejected", e["event_id"], "->", type(err).__name__,
str(err).splitlines()[0])
""")
run('python schema_topic.py')
run('python publish_v2.py')
# Step 4: The loader inserts six rows, then crashes before the ack
write('loader.py', """\
import json, sys
from google.cloud import pubsub_v1
subscriber = pubsub_v1.SubscriberClient()
sub = subscriber.subscription_path("tb-shop-dev", "clicks-loader")
table = json.load(open("table.json")) # stands in for BigQuery
resp = subscriber.pull(subscription=sub, max_messages=50, timeout=10)
msgs = resp.received_messages
new = 0
for m in msgs:
mid = m.message.message_id
if mid not in table: # dedupe on the message ID
table[mid] = json.loads(m.message.data)
new += 1
json.dump(table, open("table.json", "w"), indent=1)
ids = sorted(int(m.message.message_id) for m in msgs)
print("pulled IDs", ids, "inserted", new, "skipped", len(msgs) - new)
if sys.argv[1:] == ["crash"]:
sys.exit("crashed before the ack")
ack_ids = [m.ack_id for m in msgs]
subscriber.acknowledge(subscription=sub, ack_ids=ack_ids)
print("acked", len(msgs))
""")
run('python loader.py crash', expect_error=True) # EXPECTED TO FAIL: the video shows this error
# Step 5: After the 10 s ack deadline the same message IDs come back: the loader skips them
run('sleep 13 && python loader.py')
# Step 6: A publisher retry sends e-1003 again with a new message ID
write('resend.py', """\
import json
from google.cloud import pubsub_v1
# the publisher timed out on e-1003 and sent it again
publisher = pubsub_v1.PublisherClient()
topic = publisher.topic_path("tb-shop-dev", "clicks")
ts = "2026-10-28T09:01:10Z"
data = json.dumps({"event_id": "e-1003", "user": "u-9",
"page": "home", "ts": ts}).encode()
future = publisher.publish(topic, data, event_type="click",
event_id="e-1003", event_ts=ts)
print("re-published e-1003 -> message", future.result())
""")
write('report.py', """\
import json
from collections import defaultdict
table = json.load(open("table.json"))
by_event = defaultdict(list)
for mid, row in table.items():
by_event[row["event_id"]].append(mid)
print("rows", len(table), "| unique message IDs", len(table),
"| unique event IDs", len(by_event))
for eid, mids in sorted(by_event.items()):
if len(mids) > 1:
print("duplicate", eid, "message IDs", sorted(mids, key=int))
""")
run('python resend.py')
run('python loader.py')
run('python report.py')
Part 10: Real-World Manager Scenarios
The sample data on this page is invented for teaching. Every terminal output is the real output of the command above it, run on the official Google Cloud Pub/Sub emulator with google-cloud-pubsub 2.43.0. Features the emulator does not support are shown as code. Part of the Real Time Pub/Sub Interview Questions and Answers series on TechBrothersIT. Features, limits and their status (GA or preview) are current as of October 2026; check the official Pub/Sub documentation (cloud.google.com/pubsub/docs) for anything you quote in an interview.
No comments:
Post a Comment
Note: Only a member of this blog may post a comment.