Big Data Processing Capstone

Data Engineering & Modern Data Platforms · Lesson 35 · Big Data Processing and Apache Spark

Lesson Overview

This is Capstone Project 3. The ingestion module left TechLearningPro with raw progress events, a pinned lesson manifest, and enrollments. The lessons in this module explained how a cluster shuffles, how a plan gets expensive, how a micro-batch tile closes a window, when Flink would take a different hop, and how a rerun doubles a day if the sink appends. The capstone is the job that has to do all of the batch half, on a date that is large enough to embarrass a laptop design, without inventing a platform that does not exist.

You will build a PySpark pipeline that publishes two curated tables for an explicit list of event_date values: fact_lesson_engagement at the grain (learner_key, lesson_id, event_date), and course_day at (course_id, event_date). Inputs are Parquet. The lesson manifest is broadcast. The enrollment join is a shuffle, with adaptive skew handling and a deterministic salt available for measured hot courses. Unknown lessons stay countable. A quality gate reads the staged bytes and is the only thing allowed to move a manifest pointer. A second job id for the same dates replaces the publish. A neighbor date stays. A late event, ingested after its event date, enters the restatement horizon and does not rewrite the live streaming tile.

The scale you design for, and label as a design target rather than a number you must generate on a laptop, is a peak day of about sixty million progress events, a lesson manifest of about forty thousand rows for one pin, and about twenty million enrollment rows at the grain (learner_key, course_id). One course may hold on the order of ten to twenty percent of events. Output files for a date land around 128–256 MB. The local fixture is a few thousand rows with the same schemas, the same hot-key shape, and the same tests. Passing the fixture without a written production config is an incomplete capstone. Passing a production config that the fixture tests do not exercise is also incomplete.

Spark 3.5 is the API surface: DataFrame.join, broadcast, explain(mode="formatted"), adaptive query execution, and partitionOverwriteMode. The publish pointer is an application convention from Distributed Processing Reliability, not a Spark built-in. No internal TechLearningPro SDK is required or allowed. If you run against object storage, use the cluster's documented filesystem scheme, such as s3a://, and still move a pointer rather than assuming a directory rename is atomic.

Learning Objectives

By the end of this capstone you will be able to:

1. State the business questions the two curated tables answer, and the questions they deliberately do not answer.
2. Fix the grain, the keys, and the columns that must not appear, including email and payment data.
3. Draw the physical plan: broadcast manifest, salted or adaptive enrollment join, aggregate, stage, gate, pointer.
4. Implement the pipeline in PySpark 3.5 so a local fixture runs and a production configuration is written beside it.
5. Keep unknown lesson_id values visible and fail the publish when their rate exceeds a cap you and the metric owner would sign.
6. Publish so two job ids produce one visible sum, and a crash before the pointer leaves the previous sum.
7. Restate a bounded list of event dates from event time, including an event whose ingest date is later.
8. Size files, enable adaptive execution, and attach a formatted explain that shows the broadcast.
9. Define monitors for the failure domains from lesson 34: shuffle loss, gate failure, doubled counts, late-data volume, and file-count drift.
10. Write the limits: the streaming tile, Flink session state, and Kafka retention are out of this job and pointed at by contract, not reimplemented.

Prerequisites

• Lessons 30 through 34 in this module. The capstone is their integration. Reread the join, the adaptive settings, the tile boundary, and the manifest publish before you write code.
• Data Quality Engineering: fail, quarantine, and the publish bit. The capstone uses a smaller gate with the same idea.
• Data Formats and Serialization: Parquet in, Parquet out, partition column event_date.
• PySpark 3.5.x and Python 3.11. A test runner you already know, such as pytest, is enough. No cloud account is required for the fixture.
• The streaming lessons are prerequisites for the boundary statement, not for a second implementation. Do not submit a Flink job as a substitute for the batch publish.

Detailed Core Concepts

The product

Content and learning-science staff ask, for a business date: how many events reached each lesson, how many of those events matched the pinned manifest, how many matched an enrollment, and which courses dominated the day. They ask it again the next morning, and Monday's answer may move until the restatement horizon closes. They do not ask this job for the five-minute tile. They do not ask it for per-learner session state. They do not ask it to charge a card.

Two tables serve that ask.

fact_lesson_engagement is the detailed fact. Grain: one row per (learner_key, lesson_id, event_date). Measures: event_count, first_event_time, last_event_time. Attributes from the manifest: course_id, manifest_id, lesson_title. manifest_id is null when the lesson was unknown. enrolled is true when the enrollment snapshot contained (learner_key, course_id).

course_day is the aggregate a dashboard can scan without reading learner rows. Grain: (course_id, event_date). Measures: event_count, matched_lesson_count, enrolled_event_count, unknown_lesson_count, learner_count. learner_count is a distinct count on this smaller aggregate, where the key space is courses times dates, not the streaming tile's hot state. Document that it is an exact distinct count in the batch job.

The streaming tile remains (course_id, window_start, window_end) with a ten-minute watermark, owned by lesson 32. Reconciliation against course_day is a diagnostic over a closed hour, not a publish dependency. Flink is not in this pipeline. The decision record from lesson 33 stays a record, not a second job, unless you are explicitly extending the capstone and you still deliver the Spark publish.

The numbers you design for

Write these in the README of the project so a reviewer sees which tests used the fixture and which settings would change at the design volume. spark.sql.shuffle.partitions of 8 is a fixture setting. It is not the production setting. Production starts at 200 with adaptive coalescing and parallelismFirst set to false, then changes only with a metric.

What the job refuses

• Email, name, payment token, and raw IP. learner_key is the pseudonymous id already on the raw event. If a column arrives that matches a prohibited list you configure (email, payment_token, card_number), the gate fails the publish.
• An inner join as the only handling of unknown lessons. The fact keeps the row with a null manifest_id. The gate fails if the unknown rate exceeds the cap. Default the cap to 2% of staged event rows and treat the number as a signed parameter, the way lesson 17 did.
• Append to the published prefix.
• F.uuid() or F.rand() anywhere in the transform. Salt uses pmod(hash(event_id), n).
• A Python UDF on the filter that selects dates. Partition pruning has to see event_date or the ingest range as a column predicate.
• Deleting raw inputs.
• Reading or writing Kafka. The contract for Module 4 is a paragraph in your design notes: topic learning.progress, JSON fields, retention longer than a checkpoint recovery, cursor lives in the streaming checkpoint. This job reads Parquet that ingestion already landed.

Ownership

You own the Spark job, the staging layout, the gate, and the manifest pointer. Ingestion owns raw files and the manifest pin. The metric owner owns the cap and the restatement horizon. The streaming owner owns the tile. A capstone that builds all of those systems has left the assignment. A capstone that names them and implements the batch publish has done it.

Architecture and Design Diagrams

mermaidCopy
flowchart TD
    subgraph inputs [Inputs kept immutable]
        raw[Progress Parquet partitioned by ingest_date]
        dim[Lesson manifest for one manifest_id]
        enr[Enrollment snapshot]
    end

    subgraph spark [PySpark job for an explicit date list]
        read[Filter ingest horizon and project columns]
        bcast[BroadcastHashJoin left on lesson_id]
        salt[Deterministic salt for hot courses]
        smj[SortMergeJoin left on learner course salt]
        fact[Aggregate fact_lesson_engagement]
        day[Aggregate course_day]
        stage[Staging prefix job_id]
    end

    subgraph publish [Publish]
        gate{Gate: null keys, prohibited columns, unknown-lesson cap, count reconciliation}
        pointer[Manifest pointer]
        prev[Previous pointer]
    end

    raw --> read
    dim --> bcast
    read --> bcast --> salt --> smj
    enr --> salt
    smj --> fact --> stage
    smj --> day --> stage
    stage --> gate
    gate -->|pass| pointer
    gate -->|fail| prev

Readers of both tables follow the pointer for each table, or one pointer that names both prefixes from the same job_id. One pointer is easier to keep consistent. Use one.

mermaidCopy
sequenceDiagram
    participant O as Orchestrator
    participant J as Spark job
    participant S as Staging prefix
    participant G as Gate
    participant M as Manifest object
    O->>J: job_id and event_date list
    J->>S: Overwrite staging for this job_id only
    J->>G: Read staging back
    alt gate fails
        G-->>O: Non-zero, pointer unchanged
    else gate passes
        G->>M: Write pointer at job_id
        G-->>O: Success
    end

A second orchestrator attempt is a new job_id. It does not continue the first attempt's staging prefix. If the first attempt already moved the pointer, the second attempt replaces it with a recomputation that must match.

Technical Explanation

Schemas

Raw progress, Parquet, partitioned by ingest_date. The job does not require ingestion to partition by event_date, because event time can be late.

Drop any other column at project time. In particular, drop payload blobs before the shuffle.

Lesson manifest, filtered to one manifest_id before the broadcast:

Enrollments, snapshot id stored on the job arguments:

fact_lesson_engagement:

course_day:

unknown_lesson_count plus matched_lesson_count equals event_count for the date. That identity is a gate check. It is a stronger audit than either column alone.

Join and aggregation rules

Left-broadcast the manifest on lesson_id. If the event's course_id and the manifest's course_id both exist and differ, count the row as a conflict and fail the gate. Do not silently prefer one. Conflicts are a contract bug, not a skew bug.

Deduplicate enrollments on (learner_key, course_id) for the snapshot. A duplicate fails the gate if you would rather stop than drop, which is the stricter and better capstone choice: fail when count > countDistinct on that grain, and make the supplier fix the snapshot. The fixture includes one duplicate so this branch is tested.

Salt only courses in a HOT_COURSES argument. Default the fixture to one hot id. Join enrollments on (learner_key, course_id, _salt) and drop _salt before the aggregate. The fact grain does not contain salt. Adaptive skew join stays enabled in production config even with the salt, so an unlisted hot course is not invisible.

Aggregate to the fact grain with count, min(event_time), max(event_time). Then derive course_day from the fact, or from the pre-aggregate events if you need event-level enrolled counts. Be careful: learner_count is distinct learners with at least one event that day, which you can compute from the fact. enrolled_event_count is a count of events, not of learners. If you aggregate to the fact first, you no longer have event rows. Compute event-level measures before collapsing to (learner_key, lesson_id, event_date), or keep both aggregates in one pass over the joined events:

pythonCopy
events_joined.groupBy("course_id", "event_date").agg(
    F.count("*").alias("event_count"),
    F.count("manifest_id").alias("matched_lesson_count"),
    (F.count("*") - F.count("manifest_id")).alias("unknown_lesson_count"),
    F.count(F.when(F.col("enrolled"), True)).alias("enrolled_event_count"),
    F.countDistinct("learner_key").alias("learner_count"),
)

count("manifest_id") skips nulls. The subtraction is the unknown count. Assert the identity in the gate.

File layout and publish

Staging:

textCopy
{warehouse}/staging/job_id={job_id}/fact_lesson_engagement/event_date=...
{warehouse}/staging/job_id={job_id}/course_day/event_date=...

Choose the fact's file count from target_files(output_bytes) aiming at 128–256 MB. On the fixture, call the function with a fake byte count and assert the integer. On a real run, log output bytes and file count per date. repartition(n) before the write. Partition by event_date only.

The pointer is one JSON object written by the driver:

jsonCopy
{
  "job_id": "20260303T0600Z",
  "manifest_id": "dim-2026-03-03",
  "enrollment_snapshot_id": "enr-2026-03-03",
  "dates": ["2026-03-02", "2026-03-03"],
  "fact_path": ".../staging/job_id=20260303T0600Z/fact_lesson_engagement",
  "course_day_path": ".../staging/job_id=20260303T0600Z/course_day",
  "unknown_lesson_rate": 0.01
}

Readers use fact_path and course_day_path. They do not glob staging/*. A failed gate writes nothing to the pointer path. It may write a rejection record beside it, under rejections/job_id=..., so the orchestrator has a body to show.

Dynamic partition overwrite is not required if readers use the pointer. You may still set spark.sql.sources.partitionOverwriteMode to dynamic for any direct table write you add. If you do, the neighbor-date test stays in the suite. The pointer test is mandatory either way.

Production configuration

Ship a config file, not only builder.config calls buried in main.

Executor memory is a comment in the config: set it after a fixture is impossible to teach you, and after a production-shaped explain shows spill on every task rather than one hot task. The capstone submission includes the sentence you would use to ask for memory, or the sentence that says you would not.

Gate

Fail the publish when any of these are true:

1. A prohibited column name is present on the event schema you actually read.
2. lesson_id is duplicated inside the manifest pin.
3. Enrollment grain is duplicated inside the snapshot.
4. Any staged fact row has a null learner_key, lesson_id, course_id, or event_date.
5. unknown_lesson_count / event_count for a date exceeds the cap. Print both the rate and the counts.
6. For each date, matched_lesson_count + unknown_lesson_count differs from event_count.
7. The sum of fact event_count differs from the sum of course-day event_count for that date.
8. The requested date list is empty, or staging has no partition for a requested date that the input showed had events.

Quarantine is optional in this capstone. If you split unknown lessons into a side dataset, still count them in the rate, and still do not drop them from raw. Failing the whole publish over the cap is enough.

Monitoring

Log and, where you have a metrics sink already, emit:

The Spark UI remains the place you confirm shuffle read and the broadcast. The log line is what the orchestrator can alert on tomorrow. A capstone that says "we will watch the Spark UI" and prints nothing has not closed the loop.

Practical Implementation or Design Walkthrough

Repository layout

textCopy
capstone_engagement/
  README.md                 # volumes, horizon, cap, out of scope
  config/
    fixture.conf
    production.conf
  src/
    job.py                  # build_spark, transform, stage, gate, pointer
    transforms.py
    publish.py
  tests/
    test_gate.py
    test_publish.py
    test_late_event.py
    test_plan.py
  fixtures/
    events/
    lessons/
    enrollments/

Split modules however you like. The tests are not optional.

Steps

1. Parse arguments: job_id, manifest_id, enrollment_snapshot_id, dates, warehouse, hot_courses, unknown_cap. Refuse to run if job_id is empty.
2. Read raw Parquet with a column list. Filter ingest_date to a horizon that covers the dates plus lateness (the fixture can use the dates themselves plus one later ingest date). Derive event_date from event_time. Keep rows whose event_date is in dates.
3. Fail fast if prohibited fields are in the schema.
4. Read the manifest for manifest_id. Check uniqueness of lesson_id.
5. Broadcast-join left on lesson_id. Flag course-id conflicts.
6. Read the enrollment snapshot. Check uniqueness. Salt and left-join.
7. Drop _salt. Aggregate fact and course_day.
8. explain(mode="formatted") to the log on the fixture flag, and always in a test that captures the string and asserts BroadcastHashJoin.
9. Write staging under job_id. Repartition the fact using the file-count function.
10. Read staging. Run the gate. On failure, exit non-zero and leave the pointer.
11. On success, write the pointer from the driver with a single replace of the JSON object.
12. Print the monitor lines: counts, rate, file count, job_id.

Reference transform

The structure below is the contract. Constants that the tests must control are arguments, not globals buried in a notebook.

pythonCopy
"""Capstone transform. PySpark 3.5. Fixture and production share this function."""

from pyspark.sql import DataFrame
from pyspark.sql import functions as F

PROHIBITED = ("email", "payment_token", "card_number", "raw_ip")


def assert_no_prohibited(df: DataFrame) -> None:
    found = [c for c in df.columns if c.lower() in PROHIBITED]
    if found:
        raise ValueError(f"prohibited_columns={found}")


def with_event_date(events: DataFrame, dates: tuple) -> DataFrame:
    assert_no_prohibited(events)
    return (
        events.select(
            "event_id", "learner_key", "lesson_id", "course_id", "event_time", "ingest_date"
        )
        .withColumn("event_date", F.to_date("event_time"))
        .where(F.col("event_date").isin(*dates))
    )


def salt_for(hot_courses: tuple, buckets: int):
    hot = F.col("course_id").isin(*hot_courses) if hot_courses else F.lit(False)
    return F.when(hot, F.pmod(F.hash("event_id"), F.lit(buckets))).otherwise(F.lit(0))


def target_files(num_bytes: int, target_bytes: int = 192 * 1024 * 1024) -> int:
    # 192 MB sits inside the 128–256 MB band. At least one file.
    if num_bytes <= 0:
        return 1
    return max(1, (num_bytes + target_bytes - 1) // target_bytes)

Wire salt_for into the enrollment join the way lesson 30 did, and call broadcast on the manifest. target_files(20 * 1024**3) is 107 for a 20 GB day at 192 MB. Assert that in a unit test that does not start Spark. Assert target_files(0) == 1.

Tests

Run the Spark tests in local[*]. They should finish in seconds on the fixture. A test that needs sixty million rows is not a unit test. Mention in the README how you would sample a production day to compare task-duration ratio, without pretending the fixture measured shuffle bytes.

Expected fixture output

For a hand-built day with 100 events, 4 unknown, and 90 of the known events enrolled, the course-day row for that course shows event_count=100, unknown_lesson_count=4, matched_lesson_count=96, enrolled_event_count=90. If your enrollments do not cover some known lessons, enrolled_event_count moves and the test should use the number the fixture implies, not this example. Write the expected numbers next to the fixture so the test is a comparison, not a printout.

Operational notes

Schedule the job after raw for the newest ingest date is closed and the manifest pin for the run exists. Pass that pin as an argument. Do not look up "latest" inside a task. The orchestrator retries the process. The retry is safe because of the pointer. A retry does not delete the previous pointer first. Freshness: if the newest ingest_date is older than the bound you were given, fail. An empty date with an empty input is a valid publish of zeros only when the source window itself was empty and the gate's check 8 says so. An empty output with a non-empty input is a failure.

Cost: the dominant knobs are shuffle bytes of the enrollment join, the distinct count on course_day, and file-listing pain if you regress the layout. Broadcast of forty thousand lessons is noise next to sixty million events. Do not cache the raw events. Persist the joined projection only if two actions would otherwise recompute it; reading staging back is a separate read of a much smaller aggregate, so you usually do not need a cache of the events.

Enterprise Real-World Scenarios

Scenario 1: Launch week and a hot course

A cohort launch puts roughly one fifth of events on course-k8s. The fixture already contains that shape at small volume. At design volume the enrollment stage's slowest task stretches if the salt list is empty and the skewed partition exceeds both adaptive thresholds. Your production note says HOT_COURSES is loaded from the previous run's top course shares, and AQE skew join remains on. The file-count function still runs after the aggregate, so the hot course does not become thousands of files. The gate's identity check does not change on a hot day. What changes is elapsed time, and the monitor is the max-to-median task ratio. You attach one formatted explain from a launch-sized sample or, if you cannot run it, a written prediction of which operator is the shuffle and which is the broadcast. A reviewer can reject a prediction that broadcasts enrollments.

Scenario 2: A catalog pin from yesterday

The manifest argument points at yesterday's pin while today's lessons are already in the events. Unknown-lesson rate crosses 2%. The gate refuses to move the pointer. Yesterday's curated dates stay visible. Raw is untouched. The orchestrator retries after the pin argument is corrected, with a new job_id, and the rate falls. Nobody edits Parquet in the published prefix by hand. This is the lesson 17 story expressed as a Spark publish. Your test test_cap_blocks_pointer is the regression that keeps the inner join from coming back.

Scenario 3: What you will not build

Product asks the capstone to "also do the live tile in Flink, exactly-once, into Kafka." That is a different hop. The acceptable response inside this project is a short contract: the tile's grain and watermark stay lesson 32's, a Flink replacement must match them before cutover, and this job's course_day is the system of record for the date. You may include that page in the README. You may not drop the replay test to spend the week on a second cluster. Scope is a reliability feature.

Best Practices

• One transform function for fixture and production. Configuration selects volumes and partition counts.
• Arguments for every pin: manifest_id, enrollment_snapshot_id, job_id, dates, cap, hot courses.
• Project the raw schema before any join. Prohibited names fail before a shuffle.
• Broadcast the manifest. Deduplicate or reject a bad enrollment snapshot before replicating salts.
• Drop _salt before the aggregate. Prove it with a schema assertion.
• Stage by job_id. Move one pointer from the driver after a read-back. Never append the published data.
• Test two publishes, a blocked cap, a neighbor date, and a late event. Those four are the capstone.
• Log the monitor lines on every run, including fixture runs, so the format is real.
• Keep the streaming tile and Flink out of the critical path. Name them as boundaries.
• Write the production shuffle and AQE settings in production.conf even though CI uses fixture.conf.

Common Mistakes and Anti-Patterns

A notebook that cannot be run twice. The second run appends or overwrites the only copy. The tests exist to make that impossible to miss.

Inner join to the manifest to "clean" the fact. The unknown rate becomes zero and the gate looks healthy. The missing events are gone. The test with lesson-missing is mandatory because of this.

Salting in the output. Distinct counts and file partitions split by salt. The schema test catches it if the column is still named _salt, so do not rename it to bucket and keep it.

latest manifest looked up from inside the executors. Two tasks can observe two pins. Pass one manifest_id in.

Cache of raw events plus a wide select. Memory pressure, no second action that needs it. Cache nothing until a measured recompute says otherwise.

File count left at the mercy of partitionBy("course_id"). The design volume will create a small-file incident. Partition by event_date and use target_files.

Pointer written before the gate, or written by each task. A failed gate that already moved the pointer is a bad publish. A pointer written by tasks races. One driver write, after the gate.

Fixture-only configuration submitted as production. local[*] and eight partitions do not describe sixty million rows. The second config file is part of the grade.

Reimplementing Kafka or Flink inside the capstone. You will ship neither well. Document the boundary.

Dividing a doubled count instead of republishing from raw. The source of truth is the raw files and a new job_id, not arithmetic on a bad publish.

A distinct count added to the streaming tile "while you are here." That state budget belongs to lesson 32's design review, not to a quiet extra aggregate.

Troubleshooting and Design Review

Symptom: test_two_job_ids_one_sum sees twice the events.

Likely cause: the test reads every staging prefix, or the job appends into a shared path, or the pointer is not what the assertion reads.

Investigation: print the pointer path and the sum you computed. List directories under staging.

Corrective action: read only fact_path from the pointer. Overwrite staging per job_id. Do not union historical job ids.

Symptom: The explain plan shows SortMergeJoin on lesson_id.

Likely cause: broadcast was not applied, the join is not a left join with the manifest on the right, or the test looked at the enrollment join and thought it was the manifest join.

Investigation: two joins exist. Mark them in the plan output.

Corrective action: F.broadcast on the manifest only. Assert on that node. The enrollment join should be a sort-merge or shuffled hash. Seeing a broadcast of enrollments is a defect at design volume.

Symptom: The cap test passes on the fixture and would pass in production for the wrong reason.

Likely cause: the fixture's unknown rate is 50% and the cap is 100%, so nothing can fail. Or the cap is zero and every realistic day fails.

Investigation: the README's signed cap and the fixture's actual rate.

Corrective action: one fixture under the cap that publishes, one over the cap that does not. The cap in the test should be the parameter, set tighter than the bad fixture and looser than the good one.

Symptom: Late-event test is green but Monday's partition is empty.

Likely cause: the filter on ingest_date excluded the late file, or event_date was taken from ingest_date.

Investigation: show event_time and the derived date in the test row.

Corrective action: derive event_date from event_time. Widen the ingest filter to the horizon. Publish the business date.

Symptom: File-count function returns 200 for every input because it returns spark.sql.shuffle.partitions.

Likely cause: the shuffle count was reused as the file count.

Investigation: target_files unit test with a byte value. It should not need a Spark session.

Corrective action: base the count on output bytes. Log both numbers so a future change that couples them is obvious.

A design review for the capstone asks the lesson 30 questions about the broadcast and the salt, the lesson 31 questions about AQE and files, and the lesson 34 questions about the pointer and the crash window. It also asks whether the unknown-lesson test would still fail the publish if someone changed the join to inner. If the answer is no, the gate is measuring the output of a join that already dropped the evidence.

Interview Questions and Answers

Q1. What is the grain of each output, and what question is each grain for?

fact_lesson_engagement is one row per learner, lesson, and event date, with counts and the first and last event time. It answers which lessons a learner touched and whether that lesson was on the pinned manifest and whether the learner was on the enrollment snapshot. course_day is one row per course and date, with event counts, unknown lessons, enrolled events, and distinct learners. It answers the dashboard without a learner-level scan. The five-minute tile is a third grain and is not produced here. Mixing them is how a live window gets compared to a restated day and paged as an incident.

Q2. Why is the manifest a broadcast and the enrollment side a shuffle?

The manifest is one pin of about forty thousand rows, small enough to broadcast, and the join is a left join so the right side is the legal broadcast side. Enrollments at tens of millions of keys are not a broadcast; they would be collected on the driver. They shuffle to meet events on (learner_key, course_id), with salt on measured hot courses. I want BroadcastHashJoin in the explain for the manifest and a sort-merge, not a broadcast, for enrollments. Adaptive execution may still change a genuinely small side. It is not allowed to be the reason I broadcast the fact.

Q3. How can this job be retried without doubling Tuesday?

Each attempt writes a staging prefix that contains only its job_id. The gate reads that prefix. The driver updates one pointer object after success. Readers resolve the pointer and read that prefix only. A second attempt writes a different prefix and then moves the pointer. A crash before the pointer write leaves readers on the previous good prefix. Append at a shared path is forbidden. Raw is still there if a lost shuffle forces a recompute.

Q4. What does the gate have to prove that an explain plan cannot?

The plan proves the broadcast and the absence of a Cartesian product. The gate proves the bytes: no prohibited columns, unique dimension keys, unknown-lesson rate under the cap, the identity matched + unknown = event_count, and fact totals equal to course-day totals. A green stage with an inner join can still drop unknowns. The gate plus a test that the unknown row exists is what catches that. The plan is necessary and not sufficient.

Q5. Where does a late event go?

Onto the event_date taken from event_time, when that date is inside the restatement list the orchestrator passed. The read filter on ingest_date has to be wide enough to see the late file. The streaming tile may already have closed the five-minute window; this job does not reopen it. The pointer moves for that business date when the gate passes. A reader of the pointer sees one Monday, restated, not Monday plus a correction row hiding in Wednesday.

Q6. What do you monitor on the morning the job is late?

I separate domains. If the enrollment stage's max task time is many times the median, I look at hot course_id and the salt list before I add executors. If spill is on every task and rows are narrow, I talk about memory. If the gate rejected the rate, I look at the manifest pin, not at Spark memory. If the pointer did not move and the gate passed, the driver failed between the gate and the write, and readers are still correct on the old pointer. If file counts jumped, the write grew a new partition column. Lag of the streaming tile is a different owner's page. I do not start with a cluster resize.

Q7. What is intentionally absent from the curated schema?

Email, payment tokens, card numbers, raw IP, the salt column, and any payload blob that rode in on the raw file. learner_key stays. job_id, manifest_id, and enrollment_snapshot_id stay so a row can be traced to the pins that built it. I fail the read if a prohibited name is present rather than dropping it quietly, because a quiet drop hides a producer change. The next review should see the failure.

Q8. How do you know the fixture did not train you to broadcast the wrong side?

The plan assertion names BroadcastHashJoin on the manifest key, and a second assertion or a written production note says enrollments are not broadcast. The hot-course fixture is large in proportion and small in bytes, so a broadcast of events would also "work" locally. That is why production.conf and the design volumes are part of the submission. A reviewer who only runs the unit tests should still be able to read which side is large. I would reject my own capstone if the README's volume table and the explain assertion could not be lined up.

Q9. What would you hand Module 4, and what are you glad this job does not do?

I hand them the topic name, the JSON fields, UTC event_time, and the need for retention longer than a streaming checkpoint recovery. I am glad this job does not commit Kafka offsets, choose a transactional prefix, or run a second engine. The system of record for the date is the pointer this job writes. The tile is allowed to disagree by the watermark. Writing both into one process would couple a micro-batch checkpoint to a batch overwrite and give the on-call one knotted failure domain.

Q10. Which test, if it were deleted, would let the doubled Tuesday come back?

test_two_job_ids_one_sum, if the assertion reads the pointer and not the union of all staging paths. The crash-before-pointer test keeps the other half of the window honest. The neighbor-date test keeps a publish of Monday from replacing Wednesday. I would rather lose a style nit in review than lose those three. The broadcast assertion can fail while the publish is still idempotent; the idempotent tests can pass while the join is the wrong shape. The suite needs both families.

Q11. A stakeholder asks for learner-level streaming updates from this job. What do you answer?

I answer with the boundary. This publish is a restated daily grain, idempotent per date, hours behind the event by design. Per-learner session state at a couple of seconds is the Flink-shaped requirement from lesson 33, with its own checkpoint and sink guarantee. I can put a contract paragraph in the README. I will not add foreachBatch to the batch job and call it streaming. The failure modes are different, and the capstone's pointer does not make a micro-batch exactly-once.

Hands-On Exercises

Capstone Project 3: Curated engagement on Spark

Problem statement. Implement the pipeline in this lesson so a reviewer can run the fixture tests and read the production configuration. You are graded on the publish behavior and the plan, not on standing up a cluster.

Tasks.

1. Write the README with grains, design volumes, the cap, the restatement horizon, prohibited columns, and the streaming boundary.
2. Implement the transform, the staging write, the gate, and the driver pointer.
3. Include fixture.conf and production.conf with the settings in the table above.
4. Implement the tests listed in the walkthrough. The four publish tests and the plan assertion are mandatory.
5. Capture a formatted explain from the fixture and mark the broadcast and the enrollment join.
6. Log the monitor lines. Include one sentence on the failure domain each line is for.
7. Document the crash window of the pointer in a paragraph a new on-call engineer can use.
8. State the first lever you would pull if the enrollment stage were skewed, and the first lever if every task spilled. They must be different levers.

Expected deliverables. The repository, the test output, the explain plan, and the README.

Evaluation rubric.

You pass when a reviewer can run the tests, point at the broadcast in the plan, and explain what a reader sees if the job dies after staging and before the pointer write. You do not pass on a successful count() alone.

Lesson Summary

Capstone Project 3 publishes fact_lesson_engagement and course_day from raw progress events, one lesson manifest pin, and one enrollment snapshot. The manifest is a broadcast left join so unknown lessons remain visible to a rate cap. Enrollments shuffle, with a deterministic salt for hot courses and adaptive skew join left on. Aggregates share an identity the gate checks. Staging is per job_id. One pointer, written by the driver after the gate, is what readers follow. Two attempts produce one visible sum. A late event restates its event_time date inside a short horizon and does not reopen the streaming tile.

The design volume is a peak day of tens of millions of events. The fixture is small and the same shape. Production settings live in their own file: adaptive execution, a real initial shuffle count, speculation off, files near 128–256 MB. Kafka, Flink, and the five-minute tile stay neighboring systems with a written contract. The next module, starting with Streaming Data Architecture, builds the log those neighbors read. This job keeps the historical day honest when that log, the cluster, or the scheduler tries twice.

Further Reading

• Lessons 30–34 in this module — the join, the plan, the tile boundary, Flink's separate hop, and the pointer. The capstone is those lessons applied once.
• Spark 3.5 SQL performance tuning — broadcast threshold, adaptive coalescing, and skew join, which production.conf must not contradict.
• Structured Streaming Programming Guide — the tile you are not reimplementing, and the file-sink guarantee you can cite in the boundary statement.
• Data Quality Engineering — severity and the publish bit. The capstone gate is a narrower instance.
• Streaming Data Architecture — Module 4's start, where learning.progress becomes a system instead of a paragraph.
• Hadoop S3A committers — background if you later replace the pointer with a filesystem committer. The capstone does not require that replacement.