R10 · Week 10 Recall & Drill

Weekly revision · Week 10 · Covers 5 sessions from Mon–Fri.

Sessions covered

• S046 — Batch vs Streaming — Mental Model & Use Cases
• S047 — Spark — RDD, DataFrame, Jobs/Stages/Shuffles
• S048 — Kafka — Topics, Partitions, Consumer Groups
• S049 — Stream Processing — Watermarks, Windows, Exactly-Once
• S050 — Orchestration — Airflow, DAGs, Retries, Backfills

90-min structure

Blank-page reconstruction · 30 min

S046 · Batch vs Streaming

1. Write the three questions you ask before recommending a streaming design.
2. Define bounded and unbounded data in one sentence each, and say what unboundedness breaks.
3. Name the four classic streaming pitfalls.

Gotcha you probably forgot: when someone asks for real-time, micro-batch is very often the correct answer. It gives most of the freshness benefit at a fraction of the operational complexity, and it keeps the completeness guarantee that makes reasoning about correctness tractable. The honest question is not "can we stream" but "what latency does the decision actually require".

S047 · Spark

1. Explain lazy evaluation and the difference between a transformation and an action.
2. State what breaks a job into stages, and what determines the number of tasks in a stage.
3. Explain why a broadcast join avoids the shuffle, and how it fails when misused.

Gotcha you probably forgot: collecting a distributed result back to the driver pulls the entire dataset into a single process's memory. It looks harmless in a notebook on sample data and kills the driver on production volume. Write results out to storage, or take a bounded sample deliberately — never materialise an unbounded result on one machine.

S048 · Kafka

1. Define topic, partition, offset, and consumer group in one sentence each.
2. State the exact condition under which ordering is guaranteed, and what breaks it.
3. Explain what consumer lag measures and how you would compute it.

Gotcha you probably forgot: partition count is close to a one-way door. You can add partitions but not remove them, and adding them changes which partition a given key lands on, which breaks the per-key ordering guarantee for keys that move. Choose the count with future throughput in mind, because the migration afterwards is genuinely painful.

S049 · Stream Processing

1. Define event time and processing time, and give one bug caused by confusing them.
2. Say what a watermark asserts, and what happens to events arriving after it.
3. Match a window type to each of: a per-minute count, a user activity session, and a rolling window refreshed frequently.

Gotcha you probably forgot: windowing on processing time is almost always wrong. It makes your results depend on when data happened to arrive rather than on when the events occurred, so a consumer restart or a network hiccup silently changes yesterday's numbers. Event time plus an explicit lateness policy is the only version that reprocesses to the same answer.

S050 · Orchestration

1. Define directed acyclic graph, task, and operator in one sentence each.
2. Explain why a task must be idempotent, and give a concrete idempotent write pattern for a daily aggregation.
3. Describe what a backfill re-runs and what it must not corrupt.

Gotcha you probably forgot: retries are dangerous on a non-idempotent task. A task that appends rather than replaces will duplicate its output on every retry, and because retries are automatic and often silent, the duplication is discovered days later in a total that no longer reconciles. Delete-then-insert for the partition, or write with a natural key and upsert.

Hands-on drill · 30 min

Task: simulate the four failure shapes of this week in plain Python, so the mechanisms are concrete rather than diagrams.

mkdir -p ~/projects/w10-drill && cd ~/projects/w10-drill

Step 1 — skew: why more workers does not help (8 min)

# skew.py
import collections
import random
 
rng = random.Random(0)
 
def make_keys(skewed: bool, n: int = 100_000) -> list[str]:
    if not skewed:
        return [f"user-{rng.randint(0, 999)}" for _ in range(n)]
    # One hot key takes most of the traffic. This is what real data looks like.
    return [
        "user-hot" if rng.random() < 0.8 else f"user-{rng.randint(0, 999)}"
        for _ in range(n)
    ]
 
def partition_load(keys: list[str], workers: int) -> list[int]:
    load = [0] * workers
    for k in keys:
        load[hash(k) % workers] += 1
    return sorted(load, reverse=True)
 
for label, skewed in (("even", False), ("skewed", True)):
    for workers in (4, 16, 64):
        load = partition_load(make_keys(skewed), workers)
        # Stage wall-clock is set by the slowest task, not the average one.
        print(f"{label:<7} workers={workers:<3} biggest={load[0]:<7} "
              f"smallest={load[-1]:<7} ratio={load[0]/max(load[-1],1):.1f}")

Expected outcome: in the even case the biggest partition shrinks roughly in proportion as workers increase. In the skewed case the biggest partition barely moves no matter how many workers you add, because every copy of the hot key hashes to the same partition. Since a stage finishes when its slowest task finishes, that is a precise demonstration of why scaling the cluster does not fix skew. Now add a salt — append a small random suffix to the hot key before partitioning, and re-run: the biggest partition drops sharply, which is exactly the salting fix, with the cost that you now need a second aggregation pass to recombine the salted groups.

Step 2 — a log is not a queue (7 min)

# log.py
from collections import defaultdict
 
class PartitionedLog:
    """Append-only, retained regardless of who has read it."""
    def __init__(self, partitions: int):
        self.parts = [[] for _ in range(partitions)]
 
    def produce(self, key: str | None, value: str) -> int:
        p = 0 if key is None else hash(key) % len(self.parts)
        self.parts[p].append(value)
        return p
 
    def read(self, partition: int, offset: int) -> list[str]:
        return self.parts[partition][offset:]
 
log = PartitionedLog(3)
for i in range(6):
    log.produce("order-1", f"order-1 event {i}")   # same key -> same partition
log.produce("order-2", "order-2 event 0")
 
# Two independent consumer groups, each with its own cursor.
offsets = defaultdict(int)
for group in ("billing", "analytics"):
    for p in range(3):
        msgs = log.read(p, offsets[(group, p)])
        offsets[(group, p)] = len(log.parts[p])
        if msgs:
            print(f"{group:<10} partition={p} got {len(msgs)} msgs; first={msgs[0]!r}")
 
# Reading changed nothing. Replay proves it.
print("replay billing from 0:", len(log.read(hash('order-1') % 3, 0)), "messages still there")

Expected outcome: every event for the same key lands in one partition and is read in append order, while the second key may land elsewhere with no ordering relationship between them — that is the whole ordering guarantee, stated precisely. Both consumer groups see the full stream independently, and replaying from offset zero still returns every message, because reading never removed anything. If you expected the second group to find an empty log, you were still thinking in queue semantics.

Step 3 — watermark and late data (8 min)

# watermark.py
events = [                        # (event_time, processing_order implied by list order)
    (10, "a"), (11, "b"), (13, "c"), (12, "d"),   # d is out of order
    (20, "e"), (14, "f"),                          # f is late
]
 
WINDOW = 10
ALLOWED_LATENESS = 2
max_seen = 0
windows: dict[int, list[str]] = {}
emitted: set[int] = set()
dropped: list[str] = []
 
for t, name in events:
    max_seen = max(max_seen, t)
    watermark = max_seen - ALLOWED_LATENESS      # "no event older than this will arrive"
    w = (t // WINDOW) * WINDOW
    if w in emitted and t < watermark:
        dropped.append(name)                     # too late; window already closed
        continue
    windows.setdefault(w, []).append(name)
    for closed in [k for k in windows if k + WINDOW <= watermark and k not in emitted]:
        emitted.add(closed)
        print(f"emit window [{closed},{closed+WINDOW}) = {sorted(windows[closed])} "
              f"at watermark={watermark}")
 
print("still open:", {k: sorted(v) for k, v in windows.items() if k not in emitted})
print("dropped as too late:", dropped)

Expected outcome: the out-of-order event still lands in its correct window, because event-time assignment does not care about arrival order — only the watermark decides when the window closes. Raising the allowed lateness delays emission but rescues more late events; lowering it emits sooner and drops more. Run it with both settings and confirm the trade moves in the direction you predicted. There is no setting that gives both, and that is the entire design tension.

Step 4 — idempotent backfill (7 min)

# backfill.py
import sqlite3
 
con = sqlite3.connect(":memory:")
con.executescript("""
CREATE TABLE daily_totals (day TEXT PRIMARY KEY, total INTEGER NOT NULL);
""")
 
def task_append(day, total):
    """Non-idempotent: retries duplicate."""
    con.execute("INSERT OR IGNORE INTO daily_totals VALUES (?,?)", (day, total))
    con.execute("UPDATE daily_totals SET total = total + ? WHERE day = ?", (total, day))
 
def task_replace(day, total):
    """Idempotent: rerunning any number of times converges to the same state."""
    con.execute(
        "INSERT INTO daily_totals(day,total) VALUES (?,?) "
        "ON CONFLICT(day) DO UPDATE SET total = excluded.total",
        (day, total),
    )
 
for _ in range(3):                        # simulate two retries
    task_append("2026-01-01", 100)
for _ in range(3):
    task_replace("2026-01-02", 100)
 
print(con.execute("SELECT day, total FROM daily_totals ORDER BY day").fetchall())

Expected outcome: the appending task's row is inflated by the retries while the replacing task's row is exactly the intended value however many times it runs. Write the number down — that inflation is what a silent automatic retry does to a production metric, and the only durable defence is that the task's effect depends solely on its inputs, not on how many times it ran.

Gap analysis + next week preview · 10 min

• In Step 1, did the skewed case behave as you predicted before running it? If you expected more workers to help, the "stage ends with its slowest task" model has not landed yet.
• Did you expect the second consumer group in Step 2 to find an empty log? That instinct is the queue model still in place.
• Could you state the ordering guarantee precisely — including the partition-count caveat — without notes?

Next week (S051–S055) turns to data quality and platform work: testing and validating data with contracts and expectations; the modern data stack and warehouse-versus-lakehouse choices; the analytics-engineering workflow; metrics layers and semantic models; and cost and performance management for warehouses. The idempotence and grain ideas from the last two weeks are the foundation everything there assumes.

Part of the 6-month evergreen learning plan.