This exercise builds a complete Apache Beam pipeline that simulates the GCP Dataflow-to-BigQuery pattern for the IPL analytics platform. Using the Apache Beam Python SDK's DirectRunner, you will implement a pipeline that reads IPL delivery events from a simulated Pub/Sub message stream, applies three PTransforms (parse JSON, enrich with derived columns, validate), routes invalid records to a dead-letter path, and writes valid records to a simulated BigQuery table. The pipeline is then extended with a windowed aggregation that computes per-over run rates using Beam's fixed windowing model.
Because the Apache Beam DirectRunner executes the full pipeline logic in process without any GCP infrastructure, the exercise verifies all the conceptual patterns that would apply to a production Dataflow job: tagged outputs for dead-letter routing, PCollection transformations, windowed aggregations with timestamps, and BigQuery write patterns. A Pub/Sub idempotency test verifies that duplicate messages produce no duplicate rows when deduplication is implemented using the message ID as a uniqueness key.
Analogy🏏Cricket
🏏 Think of it like cricket: This exercise is the IPL official statistics team building their daily automated processing pipeline — the complete workflow that takes raw ball-by-ball records from every ground and produces the certified statistics published on the official website by midnight. Stage 1 is the data catalogue check: verify that the incoming scorecards match the expected format before any processing begins. Stage 2 is the statistics calculation: joins with match metadata, derivation of over-level stats, economy rate computation. Stage 3 is the official record update: load the new statistics into the production database using the certified upsert protocol — delete the old version of today's record and insert the freshly computed one — so no match ever has two records in the official database.
Step 1 — Beam Pipeline with Dead-Letter Routing
Implement the full Beam pipeline using pure Python to simulate DirectRunner execution: parse JSON delivery events, enrich with phase and boundary columns, validate runs in range [0,6], route invalid records to a dead-letter list, and write valid records to an in-memory BigQuery-style table. Inject 5 bad records (runs=9) into the stream and assert the dead-letter list captures exactly 5 records while the main output contains the remaining valid records.
Analogy🏏Cricket
🏏 Think of it like cricket: the Beam pipeline is the live scoring desk processing every ball as it is called from the middle. Just as the parse step is the scorer transcribing the umpire's spoken call into a written entry, `ParseDeliveryDoFn` turns each JSON message into a record. Just as the scorer annotates each ball with which phase of the innings it fell in — powerplay, middle or death — and flags whether it cleared the rope, `EnrichDoFn` adds the phase and is_boundary columns. Then, just as the third umpire rejects an impossible call — nobody scores nine runs off a single legal delivery — `ValidateDoFn` checks runs are in [0,6] and routes violations to a dead-letter queue with a reason attached, exactly as a disputed ball goes to the review log rather than onto the official scoreboard. Injecting five runs=9 records and asserting exactly five land in the DLQ while the rest reach the BigQuery table is the payoff: bad balls are quarantined for later inspection, and only clean, enriched deliveries ever update the live score.
python
# exercise_dataflow_bq.py — Step 1: Beam pipeline with dead-letter routing
import json
import numpy as np
from datetime import datetime, timezone, timedelta
from collections import defaultdict
from typing import Iterator, Tuple
np.random.seed(42)
# ── Simulate Apache Beam PTransform DoFn logic in pure Python ─────────────────
class ParseDeliveryDoFn:
def process(self, element: str) -> Tuple[list, list]:
valid, errors = [], []
try:
record = json.loads(element)
valid.append(record)
except json.JSONDecodeError as e:
errors.append({"raw": element, "error": str(e)})
return valid, errors
class EnrichDoFn:
def process(self, element: dict) -> dict:
over = element.get("over", 0)
runs = element.get("runs_scored", 0)
element["phase"] = (
"powerplay" if over <= 6 else
"middle" if over <= 15 else
"death"
)
element["is_boundary"] = runs >= 4
return element
class ValidateDoFn:
VALID_RUNS = set(range(7)) # 0..6
def process(self, element: dict) -> Tuple[list, list]:
valid, invalid = [], []
runs = element.get("runs_scored", -1)
if runs not in self.VALID_RUNS:
invalid.append({**element, "dlq_reason": f"invalid runs: {runs}"})
elif not element.get("bowler") or not element.get("batter"):
invalid.append({**element, "dlq_reason": "missing bowler or batter"})
else:
valid.append(element)
return valid, invalid
# ── Simulate DirectRunner pipeline execution ──────────────────────────────────
def run_pipeline(messages: list[str]) -> dict:
"""Execute pipeline: Parse → Enrich → Validate → BigQuery/DLQ."""
parser = ParseDeliveryDoFn()
enricher = EnrichDoFn()
validator = ValidateDoFn()
bq_table: list[dict] = []
dlq: list[dict] = []
parse_errors: list = []
for msg in messages:
valid, errors = parser.process(msg)
parse_errors.extend(errors)
for record in valid:
enriched = enricher.process(record)
v_valid, v_inv = validator.process(enriched)
bq_table.extend(v_valid)
dlq.extend(v_inv)
return {"bq_table": bq_table, "dlq": dlq, "parse_errors": parse_errors}
# ── Generate test stream with 5 bad records ────────────────────────────────────
NUM_GOOD = 120
NUM_BAD = 5
good_deliveries = [
{"delivery_id": i, "match_id": 10001, "over": (i//6)+1, "ball": (i%6)+1,
"runs_scored": int(np.random.choice([0,1,2,4,6])),
"bowler": np.random.choice(["Bumrah","Shami","Hardik"]),
"batter": np.random.choice(["Rohit","Kohli","Gill"]),
"event_ts": (datetime.now(timezone.utc) - timedelta(seconds=i)).isoformat(),
}
for i in range(NUM_GOOD)
]
bad_deliveries = [
{"delivery_id": 9900+i, "match_id": 10001, "over": 10, "ball": 1,
"runs_scored": 9, # invalid: > 6
"bowler": "Bumrah", "batter": "Rohit",
}
for i in range(NUM_BAD)
]
message_stream = [
json.dumps(d)
for d in good_deliveries + bad_deliveries
]
np.random.shuffle(message_stream) # simulate out-of-order messages
# Run the pipeline
result = run_pipeline(message_stream)
# ── Assertions ────────────────────────────────────────────────────────────────
assert len(result["bq_table"]) == NUM_GOOD, \
f"Expected {NUM_GOOD} valid rows, got {len(result['bq_table'])}"
assert len(result["dlq"]) == NUM_BAD, \
f"Expected {NUM_BAD} DLQ records, got {len(result['dlq'])}"
assert len(result["parse_errors"]) == 0
# Verify all DLQ records have dlq_reason
assert all("dlq_reason" in r for r in result["dlq"])
# Verify enrichment applied to all valid records
assert all("phase" in r and "is_boundary" in r for r in result["bq_table"])
assert all(r["phase"] in {"powerplay","middle","death"} for r in result["bq_table"])
print(f" Valid records → BigQuery: {len(result['bq_table'])} ✓")
print(f" Invalid records → DLQ: {len(result['dlq'])} (runs=9) ✓")
print(f" Enrichment verified: phase and is_boundary present on all ✓")
print("Step 1 ✓: Beam pipeline with dead-letter routing complete")Step 2 — Windowed Aggregation and Pub/Sub Deduplication
Implement a fixed-window aggregation that groups delivery events into 6-ball (one over) windows by over number, computing total runs and run rate per window. Then implement Pub/Sub deduplication: process the same 120-message stream twice (simulating at-least-once redelivery) and verify the deduplication logic — keyed on delivery_id — produces identical BigQuery row counts and aggregate values after both processing runs, proving the pipeline is idempotent against duplicate Pub/Sub messages.
Analogy🏏Cricket
🏏 Think of it like cricket: the fixed-window aggregation is the scoreboard's over-by-over summary — after every sixth legal ball a fresh line appears showing runs and run rate for that over. Just as the scorer closes off one over, tallies its runs and only then starts the next, `windowed_run_rate` groups deliveries into 6-ball windows by over number and computes total runs and run rate per window, with conservation checked so the windowed runs sum back to the raw total, exactly as the sum of every over must equal the innings score. The Pub/Sub deduplication is the scoring system's defence against a telegraph line that echoes: just as hearing 'Bumrah bowls a dot ball' twice must not knock a run off the board a second time, keying on delivery_id and processing the same 120-message stream twice yields an identical row count and identical aggregates. The payoff: even when messages are redelivered at-least-once, the published totals stay exactly right — the pipeline is idempotent against duplicates.
python
# exercise_dataflow_bq.py — Step 2: Windowed aggregation and deduplication
from collections import defaultdict
import pandas as pd
# ── Fixed window aggregation: 1 window = 1 over (6 balls) ────────────────────
def windowed_run_rate(
deliveries: list[dict],
window_size_balls: int = 6,
) -> list[dict]:
"""Simulate Beam fixed windowing: group by over, compute run rate."""
windows: dict[int, list[dict]] = defaultdict(list)
for d in deliveries:
window_key = d.get("over", 1) # 1 over = 1 window
windows[window_key].append(d)
results = []
for over, events in sorted(windows.items()):
total_runs = sum(e["runs_scored"] for e in events)
wickets = sum(1 for e in events if e.get("is_wicket", False))
run_rate = round(total_runs * (6 / len(events)), 2) # runs per over
results.append({
"over": over,
"balls": len(events),
"total_runs": total_runs,
"wickets": wickets,
"run_rate": run_rate,
})
return results
valid_records = result["bq_table"]
windows = windowed_run_rate(valid_records)
overs_completed = [w for w in windows if w["balls"] == 6]
assert len(overs_completed) == 20, f"Expected 20 complete overs, got {len(overs_completed)}"
assert all(w["run_rate"] >= 0 for w in windows)
total_windowed_runs = sum(w["total_runs"] for w in windows)
total_record_runs = sum(r["runs_scored"] for r in valid_records)
assert total_windowed_runs == total_record_runs, "Window aggregation run total mismatch"
print(f" Windows: {len(windows)} overs, {len(overs_completed)} complete (6 balls) ✓")
print(f" Run conservation: {total_windowed_runs} == {total_record_runs} ✓")
print(f" Sample: over 1 — {overs_completed[0]['total_runs']} runs, "
f"rate {overs_collected[0]['run_rate'] if (overs_collected:=overs_completed) else 0} ✓")
# ── Pub/Sub deduplication: at-least-once idempotency ─────────────────────────
bq_dedup: dict[int, dict] = {} # delivery_id → record (last write wins)
def load_with_dedup(records: list[dict], store: dict) -> int:
"""Simulate BigQuery MERGE with delivery_id as dedup key."""
for r in records:
store[r["delivery_id"]] = r # upsert on delivery_id
return len(store)
# First load: 120 unique records
count_after_load1 = load_with_dedup(valid_records, bq_dedup)
assert count_after_load1 == NUM_GOOD
# Second load: same 120 records (simulated Pub/Sub redelivery)
# Duplicates should not increase row count
count_after_load2 = load_with_dedup(valid_records, bq_dedup)
assert count_after_load2 == count_after_load1, \
f"Dedup failed: {count_after_load1} → {count_after_load2} rows"
print(f" Pub/Sub dedup: {count_after_load1} rows stable after re-delivery ✓")
# Verify aggregate values match between deduplicated store and original
df_bq = pd.DataFrame(list(bq_dedup.values()))
df_valid = pd.DataFrame(valid_records)
assert df_bq["runs_scored"].sum() == df_valid["runs_scored"].sum()
assert set(df_bq["delivery_id"]) == set(df_valid["delivery_id"])
print(f" Aggregate runs match after deduplication: {df_bq['runs_scored'].sum()} ✓")
print("Step 2 ✓: windowed aggregation and Pub/Sub deduplication complete")