R10 · Week 10 Recall & Drill

Assessment scope

Attempt the recall prompts unaided, then work through the implementations and answer keys. The local programs use explicit fixtures and independent checks; product behavior is tied to named primary sources, not inferred from the toy models. Service deployment and production readiness are separate from this conceptual review. Timing budgets are adjustable practice suggestions.

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

By the end of this revision you can

• Ask the three questions — how fresh, how bounded, how expensive to be wrong — before recommending any streaming design.
• Explain bounded versus unbounded data and why unboundedness changes the semantics rather than just the schedule.
• Trace a Spark job from action to stages to tasks, and name the shuffle as the thing that draws the stage boundary.
• Diagnose data skew and explain why adding executors alone may not fix an indivisible hot key.
• Describe Kafka as a replicated append-only log, and state the exact condition under which ordering is guaranteed.
• Define watermark and allowed lateness, and write an idempotent task so retries and backfills are safe.

90-min structure

The 90-minute budget is pacing guidance, not a claim that full execution and every extension fit every learner's session.

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.

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.

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.

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.

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.

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.

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

# skew.py
import collections
import random
import zlib
 
def stable_hash(key):
    return zlib.crc32(key.encode())
 
rng = random.Random(0)
 
def make_keys(skewed: bool, n: int = 5_000) -> list[str]:
    if not skewed:
        return [f"user-{rng.randint(0, 999)}" for _ in range(n)]
    # This synthetic fixture assigns roughly 80% of arrivals to one hot key.
    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[stable_hash(k) % workers] += 1
    return sorted(load, reverse=True)
 
for label, skewed in (("even", False), ("skewed", True)):
    keys = make_keys(skewed)  # one fixed dataset per comparison
    for workers in (4, 16, 64):
        load = partition_load(keys, 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 demonstrates a limit of this indivisible-key partition model, not a measured Spark cluster. 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):
        if partitions <= 0:
            raise ValueError("positive partition count required")
        self.parts = [[] for _ in range(partitions)]
 
    def produce(self, key: str | None, value: str) -> int:
        p = 0 if key is None else stable_hash(key) % len(self.parts)
        self.parts[p].append(value)
        return p
 
    def read(self, partition: int, offset: int) -> list[str]:
        if not 0 <= partition < len(self.parts) or not 0 <= offset <= len(self.parts[partition]):
            raise ValueError("invalid cursor")
        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(stable_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 this fixed-partition, single-threaded model’s append-order contract, not a complete Kafka producer/consumer guarantee. 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 — nonnegative integer event times, bounded teaching trace
 
def event_windows(events, window=10, delay=2):
    if type(window) is not int or window <= 0 or type(delay) is not int or delay < 0:
        raise ValueError("positive integer window and nonnegative integer delay required")
    max_seen = 0
    windows, emitted, dropped = {}, set(), []
    for t, name in events:
        if type(t) is not int or t < 0:
            raise ValueError("nonnegative integer event time required")
        max_seen = max(max_seen, t)
        watermark = max_seen - delay  # policy estimate, not an arrival guarantee
        w = (t // window) * window
        if w + window <= watermark:
            dropped.append(name)  # also rejects unseen windows already behind closure
            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)
    return windows, emitted, dropped
 
events = [(10,"a"),(11,"b"),(13,"c"),(12,"d"),(23,"e"),(14,"f")]
windows, emitted, dropped = event_windows(events)
assert emitted == {10} and dropped == ["f"]
assert windows[10] == ["a","b","c","d"]
assert event_windows(events,delay=10)[1:] == (set(), [])
assert event_windows([(23,"new"),(1,"unseen-old")])[2] == ["unseen-old"]
print("closed:", emitted, "windows:", windows, "dropped:", 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 this model's watermark delay postpones closure and can accept more out-of-order data; it is not a separate post-watermark allowed-lateness setting. With delay 10, timestamp 23 advances the watermark only to 13, so event 14 is retained and [10,20) remains open. With delay 2 it is dropped. A policy must trade finalization/state against late-data handling; corrections and side outputs are other choices.

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 (?,0)", (day,))
    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)
 
result = con.execute("SELECT day, total FROM daily_totals ORDER BY day").fetchall()
assert result == [("2026-01-01", 300), ("2026-01-02", 100)]
print(result)
con.close()

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 an important defence is idempotence at a defined output boundary; concurrency and changing input versions need additional rules.

What the models prove and what engines must supply

These four Python blocks are consecutive sections of one executable model. CRC32 is used only for stable toy partition assignment, not as Kafka's partitioner or a security hash. Hold a single dataset fixed across worker-count comparisons when measuring a controlled experiment. Row count is a proxy for work only if rows cost roughly the same; one expensive row can dominate a balanced count distribution. Salting can split a decomposable aggregation such as sum, then merge partials; it is not valid to change join keys or ordered per-key processing without repairing the semantics. Add workers only helps the portion of work that can actually be divided.

Spark transformations build a lazy dependency plan; actions request evaluation. Shuffle dependencies generally separate stages, with tasks processing partitions; adaptive execution can change the plan. A broadcast join replicates a sufficiently small side so matching need not repartition the large side for that join, but still pays broadcast/network/memory costs and does not erase other shuffles. A long task can also reflect I/O, spill, stragglers or expensive functions, so inspect task metrics before diagnosing skew. collect brings results to driver memory; use bounded samples or distributed writes when results cannot fit.

The log model retains everything forever. Kafka retains or compacts according to configuration, so reading does not remove records but replay is bounded by available history. A topic has ordered partitions; an offset identifies a position within one partition; groups maintain consumption positions and assignments. Per-key order requires consistent routing to one partition and correct producer/consumer processing configuration; retries and parallel processing need their own ordering analysis. Adding partitions can remap keys, so historical per-key streams can span partitions. Offset lag compares an appropriate end offset with consumed/committed offsets; it is not directly elapsed-time lag and transactional visibility changes which boundary matters.

In the repaired event trace, timestamps 10,11,13,12 enter [10,20). Seeing 23 advances this model's watermark to 21, emitting that window; event 14 then arrives for a closed window and is dropped. The original max timestamp 20 only advanced it to 18, so it never demonstrated the claimed closure/drop. WATERMARK_DELAY is an out-of-order estimate; engines may separately configure retention after a watermark (“allowed lateness”). This model accepts no events for a window after closure, but retains all accepted payloads and closed-window IDs for inspection. That is zero post-close allowed lateness, not zero retained state: memory grows with accepted input, and scanning retained windows can make total work quadratic in the number of events. It is a finite trace model, not a bounded-state streaming implementation. A real operator must separately evict finalized state under its correction/replay contract. It is not a guarantee no older event can arrive. Delaying closure can reduce late drops at greater latency/state cost; side outputs or corrections offer other policies. A far-future bad timestamp can advance this estimator disastrously, so production designs need timestamp validation and multi-partition watermark rules.

Bounded data has a finite defined extent; unbounded data continues arriving. A micro-batch engine can implement streaming semantics, and a bounded daily snapshot can still omit late business events. Processing-time windows are valid when arrival/processing time is the quantity of interest. Event-time windows with deterministic policies support meaningful replay, but changed data, watermark progression or nondeterministic code can still change results. Tumbling windows do not overlap, sliding windows do, and session windows depend on inactivity gaps.

An orchestration DAG is a directed graph with no directed cycle: an edge states a prerequisite, and acyclicity permits a topological execution order. It orders task dependencies; a task is a scheduled unit of work and an operator supplies execution behavior in frameworks such as Airflow. A backfill recomputes historical intervals using a declared data/code version. The replacement upsert is idempotent for the same day and total; it does not protect against an older run overwriting a newer result. Add a version/conflict policy and atomic commit. Delete-then-insert needs one transaction or an atomic publication protocol, not a crash-prone gap. Exactly-once effects require a boundary covering input offsets/state and sink commits or deduplication; a broker delivery label alone cannot certify external side effects.

Checks and transfer: after Step 3 assert emitted == {10}, dropped == ["f"], and windows[10] == ["a","b","c","d"]. Move f before timestamp 23 and it must be accepted. For the log, snapshot lengths before both consumer reads and confirm lengths do not change. For backfill, repeat the same replacement and require 100, then submit a different version and state whether replacement or rejection is correct.

Decision questions, Spark trace and controlled salting solution

Before choosing streaming ask how fresh, how bounded, and how expensive to be wrong. A daily finance close may prefer a bounded reconciled snapshot plus late corrections; fraud intervention may need low-latency events and explicit missed/duplicate-action policy. Four recurring hazards are out-of-order events, late arrivals, replay/duplicates and unbounded state (including session boundaries that move as data arrives). Micro-batching is an execution technique that can implement streaming semantics; a finite input is not proof all business events have arrived.

Spark trace exercise: suppose four input partitions feed filter → map → groupByKey → count using the RDD API, with map producing key-value pairs. Here the final count action counts grouped keys, not records per key; a per-key count would require a per-group size or reduceByKey sum. Filter/map are narrow transformations and can pipeline per partition; key grouping introduces shuffle dependencies; a terminal action requests a job. If the shuffle output has two partitions, expect a four-task map stage and a two-task reduce-side stage in this simplified nonadaptive plan. Real DataFrame planning can use partial aggregation and adaptive partition coalescing; inspect the executed plan. Broadcasting a small relation can avoid repartitioning the large join side but consumes broadcast memory/network and is not safe for an arbitrarily large “small” table. RDDs expose partitioned transformations; DataFrames add schema and optimizer-visible relational expressions. Excessive Python UDF work, spill or an I/O straggler can mimic skew, so compare input/task metrics first.

from collections import Counter
hot_keys = ['hot']*1600 + [f'other-{i}' for i in range(400)]
partials = Counter((key, i % 16 if key == 'hot' else 0) for i,key in enumerate(hot_keys))
merged = Counter()
for (key,salt),amount in partials.items():
    merged[key] += amount
assert merged == Counter(hot_keys)
unsalted_load = partition_load(hot_keys,16)
salted_keys = [f'{key}#{i%16}' if key == 'hot' else key for i,key in enumerate(hot_keys)]
salted_load = partition_load(salted_keys,16)
assert unsalted_load[0] >= 1600 and salted_load[0] < unsalted_load[0]
assert sum(salted_load) == sum(unsalted_load) == 2000
print('same input, hottest before/after salting',unsalted_load[0],salted_load[0])

The second pass recombines counts by the original key; hashing only salted keys without this semantic repair changes the answer. This is exact for decomposable counts/sums, not ordered state transitions or an arbitrary join. CRC32 and uniform per-row cost remain toy assumptions.

Log, watermark and orchestration exercise answers

A Kafka topic groups partitions; offsets are partition-local positions; consumer groups independently track progress while group assignment distributes partitions among consumers under the selected protocol. If a partition's next append offset is 120 and a group's committed next-read offset is 90, the simple offset distance is 30; it is not “30 seconds behind” and is not always 30 user records under compaction/transactional visibility. Partition expansion can remap keys, and stock Kafka does not support reducing an existing topic's partition count in place; migration is a separate operation. The model lacks replication, retention, rebalances and crash-safe offset commits.

Window answers: a per-minute nonoverlapping count uses tumbling windows; a rolling ten-minute count refreshed each minute uses sliding windows; an activity session uses an inactivity-gap rule. Event time is when the business event occurred; processing time is when the engine processes it. A delayed purchase can belong to yesterday's event-time revenue but today's arrival-count metric, both legitimate quantities when named. The model drops any event whose entire window is behind the closure boundary, including an unseen old window; checking only emitted would incorrectly reopen unseen expired windows. A production multi-input watermark often depends on multiple partitions and idle-input policy, not just one global maximum.

assert emitted == {10} and dropped == ['f']
assert windows[10] == ['a','b','c','d']
# Independent event-time grouping oracle for accepted events in this trace.
expected_closed = [name for t,name in events[:-1] if 10 <= t < 20]
assert windows[10] == expected_closed
lengths = [len(p) for p in log.parts]
for p in range(3):
    assert log.read(p,0) == log.parts[p]
assert lengths == [len(p) for p in log.parts]

For orchestration, define the DAG's data interval rather than using “today” inside every rerun. Task dependencies express order, not atomic multi-task effects. Replacing day D with a deterministic result avoids additive duplicates only for the same input/code version; concurrent old and new runs need versioned publication or serialization. Airflow retries do not make remote API side effects disappear. A pipeline claiming exactly-once effects must name the input-offset/state/sink commit boundary; independently committing a Kafka offset and an external charge leaves either a loss or duplicate window.

Recall quiz — five checks

1. Slow stage after adding executors? Compare task size/duration/spill/CPU/I/O. If one unsplittable key dominates, salt only with a valid merge or use supported skew-join handling; added executors alone cannot split it.
2. Log rather than destructive queue? Reads advance consumer position, not erase history; retention/compaction still limits replay.
3. Ordering? A partition supplies log order; same-key guarantees also require stable routing, producer retries/configuration and consumer processing that preserves required order.
4. Watermark? A policy/progress estimate, not proof no earlier event can arrive; explicitly choose drop, side output, update or retained-late policy.
5. Totals inflate on retry? Appending repeats effects. Use atomic keyed replacement/upsert over a defined interval/version, coordinate concurrent runs and check the external sink boundary.

Answer constraints and corrections

These are local models, not Spark/Kafka/stream-engine reproductions. A watermark is an estimate or policy boundary, not proof no earlier event can arrive. End-to-end exactly-once effects require a named boundary covering state/offsets and sink behavior. Processing time can be appropriate when arrival time is the desired quantity.

Gap analysis + next week preview · 10 min

• Did one fixed dataset keep the hot-key load concentrated as workers changed, and did recombined salted counts equal the original counts?
• Did independent groups retain the same log contents, and can you state routing/retention limits rather than confuse the log with destructive consumption?
• Can you explain why timestamp 23 closes [10,20), why 14 is late, and why an unseen expired window must also be dropped?

Mark each answer unaided, hinted or not yet; select a failed assumption and use a new counterexample for the retest rather than memorizing one fixture. Keep the input contract, boundary case and independent expected answer next to any execution result.

Next week: S051–S055 cover dbt, lakehouse table formats, data quality, governance/cost and HTTP. The interval/version contract and replay reasoning above become table, test and cache contracts.

Corrective source notes

• RDD Programming Guide - Spark 3.5.7 Documentation — Spark RDD transformations describe work; actions request evaluation. collect brings the entire RDD to one machine; bounded take avoids collecting an unbounded-size result. Retrieved 2026-10-07; exact passage reviewed 2026-10-07.
• Spark 3.5.7 tuning SQL — Spark broadcast hints prioritize a join strategy but are not guaranteed for every join type; planning depends on actual supported strategy and sizes. Retrieved 2026-10-07; exact passage reviewed 2026-10-07.
• Spark 3.5.7 Structured Streaming — Structured Streaming can implement streaming semantics through a micro-batch engine; execution interval is not the batch/stream semantic distinction. Spark watermark delay is not a guarantee that all older data is always dropped; the Python closure rule is explicitly a different model contract. Retrieved 2026-10-07; exact passage reviewed 2026-10-07.
• Apache Kafka 4.1 design — A Kafka consumer tracks a per-partition next offset and can replay retained history; consumption is not destructive dequeue. Exactly-once external effects require coordinating consumer position with output; a broker label alone does not establish that boundary. Retrieved 2026-10-07; exact passage reviewed 2026-10-07.
• Apache Kafka 4.1 basic operations — Adding partitions can remap hash-routed keys without redistributing old records; reducing a topic’s partition count in place is unsupported. Official documentation retrieved and passage reviewed 2026-10-09.
• Airflow 2.10.5 best practices — Airflow reruns should use explicit input/output intervals rather than the latest data or current wall clock. UPSERT is a recommended retry-safe write pattern, with transaction/version qualifications supplied by the lesson. Retrieved 2026-10-07; exact passage reviewed 2026-10-07.

These passages support the named propositions, not blanket product certification. The lesson-specific derivations and fixtures are separate evidence; runtime boundaries are stated beside each worked exercise.