Building a Production-Ready Lakehouse in Microsoft Fabric

Full-day hands-on workshop - FabCon Europe 2026, Barcelona, Level 300
Speaker: Nikola Ilic (Data Mozart)

The story

At 9:00 AM, you inherit a problem. Mosaic Events Group runs conferences across Europe, and their prototype lakehouse reports EUR 2,847,310 in revenue. Finance doesn't trust that number, and they are right. Daily loads are unreliable, every source table has its own hardcoded pipeline, late-arriving updates corrupt the numbers without anyone noticing, and the Power BI model reads raw operational data.

By 5:00 PM, you will have rebuilt the platform: a metadata-driven ingestion framework where adding a source takes one insert into a control table, parent-child pipelines with incremental watermark loads, a trustworthy Silver layer, a star-schema Gold optimized for Direct Lake, and one orchestrated pipeline that logs its own history and makes failures impossible to miss. The trusted number is EUR 2,613,940, and you can prove where every euro of the difference went.

Along the way, three real production failures will find you: a SharePoint-renamed duplicate file, an API pagination bug, and a late-arriving refund hiding inside an incremental aggregate. You will get no warning. That is the point.

What you will build, lab by lab

The data

Agenda

What you need

• A laptop with Edge or Chrome - everything runs in the browser
• Your workshop account (on the card at your seat)
• Sign in using an InPrivate / Incognito window
• Nothing to install, nothing to download

Ready? Start with Lab 0: Environment setup.

Presenter blocks

Opening: the story and the architecture (09:00-09:20)

Instructor · Say this

Beats, in order: (1) Introduce Mosaic Events Group and the number on the screen, EUR 2,847,310, and the sentence "finance does not trust this number, and by tonight you will know they were right." (2) One architecture slide: three sources, medallion layers, Direct Lake at the end; every lab today is one hop on this diagram. (3) The rules of the day: everything is in the workshop site, checkpoints tell you that you are on track, helpers wear badges, spare accounts exist, nothing you can break matters. (4) Do NOT mention the scars. The word "scar" never gets said before one fires.

Instructor · Timing

20 minutes hard stop. The room is freshest now; resist the urge to lecture architecture. They will understand the diagram better at 17:00 than at 09:05.

The SharePoint war story (12:15-12:25)

Instructor · Say this

Structure that worked: (1) The setup, a manufacturing client, a daily SharePoint export, a sync job re-uploading a file. (2) The mechanism, same file, new name with a (1) suffix, every downstream aggregate inflated by a plausible amount for weeks. (3) The detection, finance could not close the month, three days of forensic archaeology to find one renamed file. (4) The connection, "you found yours in ninety seconds, because your Bronze had provenance columns and your Silver had a business-key contract; that is the entire argument for the discipline of this morning." (5) Land it just before lunch so the story marinates.

Instructor · Timing

10 minutes. Catch-up buffer follows, then lunch at 12:45.

The debrief (16:50-17:00)

Instructor · Say this

Run Lab 8 from the front as the script in that lab describes: eight smells, two sentences each, the bridge to the ghost number, the closing arc from 9:00 to 17:00, and the production checklist slide. End on the trusted number and the sentence "and you can prove it."

The three scars at a glance

Room logistics

• Helpers: badge-visible, sweep during every hands-on block, escalate anything over 3 minutes to a spare workspace or account swap
• Spare accounts and spare workspaces: att161-att170, provisioned identically, hand out on any sign-in or workspace failure
• The disaster workspace: shared read-only to the attendee security group, verified from an attendee account the day before
• Timing discipline: Lab 4 starts at 14:00 sharp; forensics is the shock absorber if anything runs long

Lab 0 - Environment setup

Time: pre-arrival or first 30 minutes of the day

In this lab you will: create the lakehouse schemas, verify seed data and SQL connectivity

Instructor · Timing

Doors open 8:30. Expect a third of the room to arrive after 8:50 and a handful during the opening slides. Helpers sweep rows checking for the two green outputs (schemas message, row count 5151). Anyone stuck at sign-in gets a spare account card immediately: never debug Conditional Access live.

Instructor · Concept

Why pre-provisioned workspaces instead of bring-your-own-trial: identical starting state for 160 people, blast-radius isolation (one broken workspace affects one person), and zero dependency on personal licensing. The trade-off is provisioning automation, which is why provision.py exists.

Instructor · Beyond the lab

In a real project this provisioning step is source-controlled infrastructure: fabric-cicd or Terraform creating dev/test/prod workspaces per domain, capacity assignment governed centrally, and secrets in Key Vault referenced by variable libraries instead of a sources.json in Files.

Step 1: Sign in

1. Open Microsoft Edge or Google Chrome in an InPrivate / Incognito window (this prevents your corporate account from interfering).
2. Go to https://app.fabric.microsoft.com.
3. Sign in with the workshop account printed on the card at your seat. The username and password are both on the card.
4. If prompted to stay signed in, click Yes.

Trouble?

Trouble? If you see "You need an invitation" or a Conditional Access error, raise your hand. A helper will assign you a spare account. Do not try to use your own corporate or trial account.

Step 2: Find your workspace

1. Click Workspaces in the left sidebar.
2. You should see a workspace named something like mosaic-att-NNN where NNN is your seat number.
3. Click it to open it.

You should see:

• A lakehouse named Mosaic
• A notebook named nb_utils (shared utilities loaded by every other notebook)
• A notebook named 00_setup_schemas
• A notebook named 90_release_batch2 (you will use this later in Lab 3)

Step 3: Create the lakehouse schemas

1. Open 00_setup_schemas. The Mosaic lakehouse is already attached.
2. Click Run all in the toolbar.
3. You should see: Done - bronze, silver, and gold schemas created

Instructor · Watch for

If this cell throws Py4JJavaError InvocationTargetException, the lakehouse was created without the schemas option. That means provisioning was wrong for this workspace: swap the attendee to a spare workspace rather than fixing live.

4. Go back to your workspace. Open the Mosaic lakehouse. Under Schemas you should now see bronze, silver, and gold alongside dbo.

Step 4: Open the lakehouse and verify the data

1. In the lakehouse explorer, expand Files.
2. Verify this structure exists:

Files/
  config/
    sources.json
  landing/
    batch1/         (6 CSV files)
  staging/
    batch2/         (3 CSV files, used later)

3. Click into landing/batch1/ and confirm you see files like events.csv, venues.csv, concessions_export_2026-08-31.csv, etc.

Step 5: Test the SQL connection

1. Go back to your workspace (click the workspace name in the breadcrumb at the top).
2. Click + New item > Notebook.
3. In the notebook, click Add data items in the left panel, select From OneLake catalog, find and select Mosaic, and confirm.
4. Paste the following into the first cell and run it (click the play button or press Shift+Enter):

import json

cfg = json.loads(
    notebookutils.fs.head("Files/config/sources.json", 1024 * 64)
)

url = (f"jdbc:sqlserver://{cfg['sql_server']}:1433;"
       f"database={cfg['sql_database']};encrypt=true;loginTimeout=30")
props = {"user": cfg["sql_user"], "password": cfg["sql_password"],
         "driver": "com.microsoft.sqlserver.jdbc.SQLServerDriver"}

df = spark.read.jdbc(url, "(SELECT COUNT(*) AS n FROM dbo.ticket_orders) q",
                     properties=props)
print(f"ticket_orders row count: {df.first().n}")

5. You should see: ticket_orders row count: 5151

Instructor · Watch for

Connection failures here are almost always the SQL firewall. Verify "Allow Azure services" is ON for the server. Second most common: attendee typo while editing the cell. The cell needs no edits at all.

Trouble?

Trouble? If this fails with a connection error, raise your hand. The most common cause is a firewall rule.

Step 6: Clean up

1. You can delete this test notebook. Right-click it in the workspace and choose Delete.
2. You are ready for Lab 1.

What you just confirmed: your workspace exists, the lakehouse has seed data in Files, the three schemas are ready, and the shared Azure SQL database is reachable from your Spark session.

Lab 1 - Bronze file ingestion

Time: 45 minutes

In this lab you will: land the six file-based entities in Bronze with provenance metadata, re-runnable without duplicating data

Instructor · Say this

Open with the tape recorder line: Bronze is a tape recorder, it records exactly what the sources said, including their lies. The value shows up weeks later when someone asks where a number came from and you can answer with a file name and a timestamp.

Instructor · Timing

45 minutes. The loader cell (Step 6) is the long pole: most of the room spends 15 minutes there. Fast finishers go to Step 8 re-runnability and the stretch thought.

The story

Mosaic Events Group runs conferences across Europe. The venue point-of-sale systems and the registration platform export CSV files daily. An integration job drops them into a landing/ folder inside the lakehouse. Your first job is to get that data into Bronze tables exactly as it arrived: no transformations, no cleanup, just faithful recording of what showed up and when.

What "Bronze" means

Bronze tables have three rules:

1. Append-only. Every file that arrives gets appended. You never update or delete a Bronze row.
2. Metadata stamped. Every row gets three extra columns: _load_timestamp (when you ingested it), _source_file (which file it came from), and _batch_id (which batch it belonged to).
3. Re-runnable. If you run the loader again, files already ingested are skipped. Running it twice must not double the data.

Instructor · Concept

The three metadata columns are the trust chain. _load_timestamp answers when, _source_file answers where from, _batch_id answers which delivery. Every debugging session in Silver and Gold ends up filtering on these. Teach them as non-negotiable, the same way a financial ledger records who posted an entry.

Instructor · Beyond the lab

Improvements worth mentioning out loud: (1) file identity by content hash instead of path, which would catch a renamed duplicate at the door; do not reveal why this matters yet. (2) An expected-schema contract per entity, rejecting files whose columns drift. (3) A quarantine folder for rejects instead of failing the whole run. (4) Event-driven triggers (storage events) instead of scheduled directory listing.

Step 1: Create your notebook

1. + New item > Notebook, name it 01_bronze_files.
2. Attach the Mosaic lakehouse: in the left panel, click Add data items > From OneLake catalog, find and select Mosaic, and confirm.

Step 2: Load shared utilities

In the first cell, load the shared utilities notebook. This gives you common imports (F, DeltaTable, uuid, etc.) and the log() function.

%run nb_utils

Step 3: Set up parameters

In a new cell:

# Parameters: which landing directories to scan
LANDING = "Files/landing"
BATCH_DIRS = ["batch1"]  # we will add batch2 later in Lab 3

Step 4: Build the file-tracking table

We need to track which files have already been ingested so running the notebook again skips them.

# File tracking table: prevents re-ingesting the same file on re-run
spark.sql("""CREATE TABLE IF NOT EXISTS _ingest_log (
  file_path STRING, entity STRING, row_count BIGINT,
  batch_id STRING, ingested_at TIMESTAMP) USING DELTA""")

Step 5: Map file names to entity names

Different files map to different Bronze tables. Concession exports and checkin exports have date suffixes that need to be stripped.

import re

def entity_for(file_name):
    """Map a CSV filename to its Bronze entity name."""
    n = file_name.lower()
    if n.startswith("checkins_export"):
        return "checkins"
    if n.startswith("concessions_export"):
        return "concessions"
    # For everything else, the entity name is the filename without .csv
    return re.sub(r"\.csv$", "", n)

Quick check

Quick check: entity_for("concessions_export_2026-08-31.csv") should return "concessions", and entity_for("events.csv") should return "events".

Step 6: Build the loader

This is the main ingestion logic.

Your task: the code below has three blanks marked ____. Replace each blank with the correct metadata column name from the Bronze rules at the top of this lab. Do not run the cell until all three are filled in.

# Get the set of files already ingested (by file path)
already = {r.file_path for r in spark.table("_ingest_log")
                                    .select("file_path").collect()}

for batch in BATCH_DIRS:
    folder = f"{LANDING}/{batch}"
    try:
        entries = notebookutils.fs.ls(folder)
    except Exception:
        continue  # batch directory does not exist yet
    for f in entries:
        if not f.name.lower().endswith(".csv") or f.path in already:
            continue
        entity = entity_for(f.name)
        target = f"bronze.{entity}"
        try:
            # Read the CSV, add three metadata columns, append to Bronze
            df = (spark.read.option("header", True).csv(f.path)
                  .withColumn("____", F.current_timestamp())
                  .withColumn("____", F.lit(f.name))
                  .withColumn("____", F.lit(batch)))
            n = df.count()
            df.write.mode("append").option("mergeSchema", "true").saveAsTable(target)
            # Record this file so we do not ingest it again
            (spark.createDataFrame(
                [(f.path, entity, n, batch)],
                "file_path string, entity string, row_count long, batch_id string")
             .withColumn("ingested_at", F.current_timestamp())
             .write.mode("append").saveAsTable("_ingest_log"))
            log(target, "success", n, f.name)
        except Exception as e:
            log(target, "failure", 0, f"{f.name}: {e}")
            raise

Hint

Hint: the three blanks are the metadata column names from the Bronze rules at the top of this lab: _load_timestamp, _source_file, _batch_id.

Instructor · Watch for

Frequent issues in this cell: blanks left unfilled (NameError or a literal ____ column), the mergeSchema option removed by an eager editor (needed later in Lab 3), and re-running before the parameter cell after edits. If someone double-loaded by deleting _ingest_log, the recovery is DROP TABLE on the affected bronze tables plus re-run.

Step 7: Run the checkpoint

expected = {"bronze.events": 12, "bronze.venues": 8, "bronze.sessions": 122,
            "bronze.products": 48, "bronze.checkins": 2500,
            "bronze.concessions": 6200}
for t, n in expected.items():
    got = spark.table(t).count()
    assert got == n, f"{t}: expected {n}, got {got}"
print("Checkpoint 1 green")

If it prints Checkpoint 1 green, your Bronze file loader works correctly.

Step 8: Test re-runnability

Run the loader cell (Step 6) again. Then run the checkpoint cell again. The counts should be exactly the same: no duplicates. This proves your file-tracking works.

Think about it

Think about it: your loader skips files it has already seen, by file path. What would happen if the same data arrived under a different file name? Hold that thought.

Instructor · Say this

Plant this verbally and move on. Do not answer it. Word it exactly once: our idempotency contract is file identity equals file name. Remember that contract. Then change the subject. The payoff lands in Lab 3.

Quick reference

Lab 2 - Metadata-driven framework

Time: 70 minutes, split across the morning break

In this lab you will: build a control table and a parent-child pipeline pair that extracts every SQL table incrementally. Adding a new source table takes one insert into the control table instead of a new pipeline.

Timing note: Steps 1 and 2 (the control table and the watermark notebook) fit before the break. The pipelines come after.

Instructor · Timing

This is the densest UI lab of the day. Strongly recommended: demo-along for Part 2. You drive the pipeline building on the projector at a deliberate pace, attendees mirror. Free-running 160 people through unfamiliar pipeline UI produces a support storm.

Instructor · Say this

The headline sentence, said at the start, the middle, and the end of this lab: adding a source table is one insert into a control table. If they remember one sentence from the morning, it is this one.

Instructor · Concept

Config-as-data is the deeper principle: the same idea returns in Lab 9 as quality rules. Pipelines are engines; tables are configuration. Engines are tested once; configuration scales.

The story

Mosaic's ticketing system lives in an Azure SQL database called mosaicops. It holds attendees, ticket orders, order lines, payments, and refunds. The prototype had one hardcoded pipeline per table: five pipelines, five places to fix every bug, and a sixth pipeline every time someone added a source.

You will build it the production way: one child pipeline that can load any table, driven by one parent pipeline that reads a control table. Five tables today, fifty tomorrow: same two pipelines.

The architecture you are building

meta_ingestion_control (Delta table)          one row per source table
        |
        v
pl_ingest_master (parent pipeline)
   Lookup: read control table
   ForEach row:
        |
        v
   pl_ingest_child (child pipeline, parameterized)
      Copy data activity:  SELECT * FROM dbo.<table>
                           WHERE <wm_col> > <last_watermark>
                             AND <wm_col> <= <ceiling>
                           --> bronze.<table> (append)
        |
   (after all copies)
        |
        v
   Notebook: advance all watermarks to the ceiling

Step 1: Create the control table

1. + New item > Notebook, name it 02_setup_control. Attach the Mosaic lakehouse (Add data items > From OneLake catalog > Mosaic).
2. In the first cell, load shared utilities:

%run nb_utils

3. In a new cell, create and populate the control table:

# One row per source table. This IS your ingestion configuration.
rows = [
    ("attendees",          "modified_ts"),
    ("ticket_orders",      "modified_ts"),
    ("ticket_order_lines", "modified_ts"),
    ("payments",           "modified_ts"),
    ("refunds",            "modified_ts"),
]

(spark.createDataFrame(rows, "source_table string, watermark_column string")
 .withColumn("last_watermark", F.lit("1900-01-01 00:00:00"))
 .withColumn("updated_at", F.current_timestamp())
 .write.mode("overwrite").saveAsTable("meta_ingestion_control"))

# Verify: five rows, all watermarks at 1900
spark.table("meta_ingestion_control").show(truncate=False)

Step 2: Create the watermark-update notebook

After all tables have been copied, one notebook advances every watermark to the ceiling in a single run. No concurrency, no conflicts.

1. + New item > Notebook, name it exactly nb_update_watermarks. Attach the Mosaic lakehouse (Add data items > From OneLake catalog > Mosaic).

2. Cell 1 - Parameter cell. Click the ... on the cell and select Toggle parameter cell (it must be marked as the parameter cell):

ceiling = ""

3. Cell 2 - Update all watermarks:

from pyspark.sql import functions as F

# Advance every watermark to the ceiling in one write.
# This notebook runs ONCE after all copies succeed,
# so there are no concurrent writers and no Delta conflicts.
(spark.table("meta_ingestion_control")
 .withColumn("last_watermark", F.lit(ceiling))
 .withColumn("updated_at", F.current_timestamp())
 .write.mode("overwrite").saveAsTable("meta_ingestion_control"))
print(f"All watermarks advanced to {ceiling}")
spark.table("meta_ingestion_control").show(truncate=False)

Save it. Do not run it: the master pipeline will call it once after all copies complete.

Instructor · Concept

Explain why the watermark update lives in the master, after the ForEach: five parallel children MERGEing one small Delta table produce ConcurrentAppendException. We hit this live in rehearsal. The design lesson generalizes: children do isolated work, the parent owns shared state. If someone asks about partial failure: if any copy fails, watermarks do not advance, and the next run retries the same window for every table. Re-copying an already-copied window appends duplicates in Bronze, and Silver dedup on business keys absorbs them. That is why Bronze dedup discipline matters.

Instructor · Beyond the lab

Upgrades to name: (1) ceiling from the source, SELECT MAX(modified_ts), instead of a manual parameter. (2) Change Tracking or CDC on the SQL side replacing timestamp scans. (3) Per-table watermark advancement recorded transactionally with per-table status. (4) A load audit table with row counts per window, which makes reconciliation trivial.

Morning break. Steps 3-8 continue after the break.

Step 3: Build the child pipeline

1. + New item > Data pipeline, name it pl_ingest_child.
2. Click anywhere on the empty canvas, then open the Parameters tab at the bottom. Add four parameters, all type String, no default values:

• source_table
• watermark_column
• last_watermark
• ceiling

The Copy data activity

3. From the toolbar, add a Copy data activity (not "Copy job": they are different items). Name it Copy table.
4. Source tab:

• Connection: click Browse > Azure SQL Database. Fill in:
• Server: mosaic-sql-fabcon.database.windows.net
• Database: mosaicops
• Authentication: Basic
• Username: mosaic_reader
• Password: F@bC0n!Barca2026
• Click Create.
• Use query: select Query, then click into the query box and select Add dynamic content. Paste:

SELECT * FROM dbo.@{pipeline().parameters.source_table}
WHERE @{pipeline().parameters.watermark_column} > '@{pipeline().parameters.last_watermark}'
  AND @{pipeline().parameters.watermark_column} <= '@{pipeline().parameters.ceiling}'

5. Still on the Source tab, expand Advanced and find Additional columns. Add three (use Add dynamic content for each value):

6. Destination tab:

• Connection: click Browse all > OneLake catalog > select Mosaic
• Root folder: Tables
• Tick the Enter manually checkbox. Two fields appear: schema name and table name.
• Schema name: type bronze
• Table name: click Add dynamic content (the small link below the field) and enter: @pipeline().parameters.source_table
• Table action: Append

Instructor · Watch for

The four traps in this activity, in order of frequency from rehearsal: picking Copy job instead of Copy data; missing the Enter manually checkbox and hunting for a dynamic table name in the dropdown; typos in the dynamic query (the expression builder does not validate SQL); forgetting Append and leaving Overwrite. A wrong connection password fails only at runtime, so a red child run with a login error means retyping the connection credential.

7. That is the entire child pipeline: just the Copy activity. Save it.

Production note

Production note: the child copies data, the master manages state. If five child pipelines all updated the same watermark table in parallel, they would get Delta concurrency conflicts. Keeping the watermark update out of the child avoids this entirely.

Step 4: Build the parent pipeline

1. + New item > Data pipeline, name it pl_ingest_master.
2. On the canvas, open Parameters and add one: ceiling, type String, default value 2026-09-01 00:00:00.

Lookup the control table

3. Add a Lookup activity. Name it Read control table.
4. Settings tab:

• Connection: click Browse all > OneLake catalog > select Mosaic
• Root folder: Tables
• Table: select dbo.meta_ingestion_control
• Untick "First row only": you want all rows.

ForEach table

5. Add a ForEach activity. Name it For each table. Connect On success from the Lookup to it.
6. Settings tab: Items > Add dynamic content: @activity('Read control table').output.value. Set Batch count to 4 and leave Sequential unticked: the tables load in parallel.
7. Click the pencil icon inside the ForEach to edit its inner activities. Add an Invoke pipeline activity.
8. Settings tab:

• Type: keep Fabric selected
• Connection: click Select... and create a new connection to your workspace if prompted (connection name can be anything, for example mosaic-conn; workspace is whichever workspace you are in)
• Pipeline: select pl_ingest_child
• Wait on completion: ON
• Parameters (use Add dynamic content for each value):
• source_table = @item().source_table
• watermark_column = @item().watermark_column
• last_watermark = @item().last_watermark
• ceiling = @pipeline().parameters.ceiling

Advance watermarks after all copies succeed

9. Navigate back to the main canvas of pl_ingest_master (click "Main canvas" in the breadcrumb above the ForEach editor).
10. Add a Notebook activity. Name it Advance watermarks. Connect the On success (green) output of For each table to it.
11. Settings tab: select notebook nb_update_watermarks. Under Base parameters, add one (String):

• ceiling = @pipeline().parameters.ceiling (use Add dynamic content)

12. Save.

Step 5: Run it

1. In pl_ingest_master, click Run. It will prompt for the ceiling parameter. Keep the default 2026-09-01 00:00:00 (this is load 1).
2. Watch the run in the Output pane: the Lookup returns 5 rows, the ForEach fans out into 5 child invocations, each child runs a Copy, and then the watermark notebook runs once.
3. Wait for all activities to report Succeeded (a few minutes).

Step 6: The checkpoint

Open your 02_setup_control notebook (or a new notebook). Run:

%run nb_utils

Then in a new cell:

expected = {"bronze.attendees": 4000, "bronze.ticket_orders": 5151,
            "bronze.ticket_order_lines": 8016, "bronze.payments": 5151,
            "bronze.refunds": 241}
for t, n in expected.items():
    got = spark.table(t).count()
    assert got == n, f"{t}: expected {n}, got {got}"

# Verify all watermarks advanced to the ceiling
wm = {r.source_table: r.last_watermark
      for r in spark.table("meta_ingestion_control").collect()}
for t in expected:
    name = t.replace("bronze.", "")
    assert wm[name] == "2026-09-01 00:00:00", f"watermark {name}: {wm[name]}"
print("Checkpoint 2 green: framework works, all watermarks advanced")

Production note

Notice the refund count: 241 of 242 total refunds. One refund in the source has a modified_ts after your ceiling. Your framework correctly did not load it. Remember this.

Step 7: Prove idempotency

Run pl_ingest_master again with the same ceiling. To verify it read zero rows:

1. After the run completes, click the Output tab at the bottom of the pipeline editor.
2. You will see a list of activity runs. Find the For each table row.
3. Click the small arrow or details icon on the right side of any Invoke child pipeline row to expand its output JSON.
4. Look for rowsRead and rowsCopied in the output. Both should be 0 for every table.
5. Re-run the checkpoint in your notebook: the counts should be identical to the first run.

The window is empty because the watermark equals the ceiling. Your framework is re-runnable by design.

Instructor · Watch for

If the checkpoint shows doubled counts (8000 attendees), a previous failed run copied data without advancing watermarks. Recovery: drop the five bronze tables, recreate the control table via 02_setup_control, re-run the master. If the ForEach shows six invocations instead of five, the control table has a duplicate row from running the setup cell twice with append semantics; the setup uses overwrite, so this only happens if someone modified it.

Instructor · Say this

After idempotency passes, connect it forward: this property is what makes tonight's 3am re-run safe. You will see the payoff in Lab 7.

Step 8: Prove the headline claim

You do not have a sixth table today, but walk through what adding one would take: a single INSERT into meta_ingestion_control, and the next master run picks it up. No pipeline edits, no new connections, no deployment. That is the entire point of the framework.

Stretch goal

Stretch goal if you are ahead: add an enabled boolean column to the control table and an If Condition inside the ForEach (@equals(item().enabled, true)) so tables can be switched off without deleting their row.

Quick reference

Lab 3 - API and the day-2 drop

Time: 30 minutes

In this lab you will: ingest the partner API into Bronze (and catch its pagination bug), then absorb the day-2 file drop - and discover what happens when a source lies about what's new

Instructor · Timing

15 minutes for the API, 15 for the drop. The war story slides follow immediately, so the reveal energy carries into the break story.

Instructor · Say this

Frame APIs before they start: a pagination contract is still a contract, and the only way to know a source honors its contract is to assert on it. Nobody reads pagination code reviews; assertions read them for you.

Part A: API ingestion (15 minutes)

The story

Exhibitor and sponsor data comes from the partner portal API. It returns paginated JSON - 25 records per page, with a has_more flag telling you whether to fetch the next page.

Step 1: Create your notebook

1. + New item > Notebook, name it 03_bronze_api.
2. Attach the Mosaic lakehouse.

Step 2: Page through the API

In the first cell, set the API version:

API_VERSION = "v1"

In a new cell, paste the helpers and the paging logic:

import requests

cfg = json.loads(
    notebookutils.fs.head("Files/config/sources.json", 1024 * 64)
)
base = cfg["api_base_url"].rstrip("/")

def fetch_all(entity):
    rows, page = [], 1
    while True:
        r = requests.get(f"{base}/{API_VERSION}/{entity}",
                         params={"page": page}, timeout=30)
        r.raise_for_status()
        payload = r.json()
        rows.extend(payload["data"])
        if not payload["has_more"]:
            return rows
        page += 1

for entity in ["exhibitors", "sponsors"]:
    rows = fetch_all(entity)
    df = (spark.createDataFrame(rows)
          .withColumn("_load_timestamp", F.current_timestamp())
          .withColumn("_source_file", F.lit(f"api:{API_VERSION}/{entity}"))
          .withColumn("_batch_id", F.lit(f"api-{API_VERSION}")))
    df.write.mode("append").saveAsTable(f"bronze.{entity}")
    log(f"bronze.{entity}", "success", len(rows), f"{API_VERSION}")

Step 3: Run the assertion

In a new cell:

latest = (spark.table("bronze.exhibitors")
          .filter(F.col("_batch_id") == f"api-{API_VERSION}"))
total = latest.count()
distinct = latest.select("exhibitor_id").distinct().count()
assert total == 80 and distinct == 80, (
    f"{API_VERSION}: {total} rows, {distinct} distinct exhibitors")
print("API checkpoint green ✓")

This will fail. You'll see 81 rows, 80 distinct. One exhibitor appears twice.

Instructor · Say this

Reveal script for the API bug: ask the room what they got. 81 and 80. Ask which exhibitor is duplicated, let someone find it with a groupBy count. Then the punchline: the API returned page 1 ending with EXH-025 and page 2 starting with EXH-025. No error, valid JSON, correct total pages. The only witness was your assertion. Then the fix is one character, v1 to v2, which is exactly how vendor fixes arrive in real life.

Instructor · Concept

Why Bronze keeps the buggy v1 load: append-only means evidence. When finance asks why last month's exhibitor count was wrong, the answer is in Bronze with a _batch_id. Deleting bad loads deletes your ability to explain history.

Step 4: Diagnose

The v1 API has a pagination bug: the last record of each page is repeated as the first record of the next page. No error, no missing data, one hidden duplicate per page boundary. Ask yourself: would your pipeline have noticed without this assertion?

Step 5: Fix by switching to v2

The vendor "fixed the API." Change the first cell to:

API_VERSION = "v2"

Re-run the fetch cell and the assertion. It goes green: 80 rows, 80 distinct.

Production note

Notice: Bronze now contains both the v1 (buggy) and v2 (clean) loads. You did not delete the bad data - Bronze is append-only. The v1 load is evidence, not embarrassment. Deduplication happens in Silver.

Part B: The day-2 file drop (15 minutes)

Step 6: Release batch 2

Open the notebook called 90_release_batch2 from your workspace. Attach the Mosaic lakehouse (Add data items > From OneLake catalog > Mosaic) and run it. It copies the staged batch 2 files into your landing folder.

Look at the output. You should see three files:

• concessions_export_2026-09-01.csv - legitimate day-2 POS export
• checkins_export_2026-09-01.csv - legitimate day-2 checkins
• concessions_export_2026-08-31 (1).csv - what is this?

Step 7: Re-run your file loader

Go back to your 01_bronze_files notebook.

1. Find the parameters cell (Step 3 of Lab 1, the cell with BATCH_DIRS = ["batch1"]). Change it to:

BATCH_DIRS = ["batch1", "batch2"]

2. Re-run the parameters cell first (so the variable updates in memory).
3. Then re-run the loader cell (Step 6 of Lab 1: the large cell with the for batch in BATCH_DIRS: loop). It will skip the batch 1 files (already tracked) and ingest the three new batch 2 files.

Step 8: Check the concessions count

In a new cell in your file loader notebook:

print("bronze_concessions:", spark.table("bronze.concessions").count())

You should see 18,200 rows.

Instructor · Say this

This is the first scar landing. Keep a flat tone while they count. Ask: day 1 had 6,200, day 2 had 5,800, so where did the extra 6,200 come from? Let the room find the file with a groupBy on _source_file. Then name it: SharePoint renamed a re-uploaded file and your loader honored the contract you gave it, file identity equals file name. The loader is innocent. The contract was wrong. Recall the seed from Lab 1.

Instructor · Watch for

Rehearsal-verified failure modes on the re-run: BATCH_DIRS typed as one string with a comma inside (the 404 error shows batch1%2C%20batch2 in the path); parameter cell edited but not re-run, so the loader still sees batch1 only; and the mergeSchema error if the option was removed. All three produce the same symptom, count stays 6,200, so check in that order.

Instructor · Beyond the lab

The definitive fix is content-hash identity: hash each file on arrival and skip known hashes. Mention manifest-driven ingestion (the source posts a manifest of what it sent) as the enterprise version. Neither belongs in Bronze cleanup: the duplicate rows are Silver's job, which is the bridge to Lab 4.

But wait: day 1 had 6,200 concession sales, day 2 had 5,800. That's 12,000. Where did the extra 6,200 come from?

The reveal

concessions_export_2026-08-31 (1).csv is byte-identical to the day-1 export. It is the same file, renamed. This is exactly what SharePoint does when someone re-uploads a file that already exists: it renames the new copy with a (1) suffix and keeps both. Nobody is notified.

Your file tracker did nothing wrong. It tracks by file name, and this is a new name. The idempotency you built in Lab 1 protects against re-running your own loader - it does not protect against the source lying about what is new. Sources lie.

Do not fix this in Bronze. Bronze's job is to record what arrived, exactly as it arrived. The duplicate file did arrive. Killing it on business key (sale_id) is Silver's job. You'll do that in Lab 4.

Production note

Real production story: this exact scenario happened at a manufacturing client. A SharePoint sync job duplicated a daily export, and every downstream aggregate was inflated for weeks. Nobody noticed because the numbers were wrong by a plausible amount. They were higher, never double. It surfaced when finance couldn't close the month.

Summary of what's in Bronze after Labs 1-3

Lab 4 - Silver and the late refund

Time: 75 minutes (the longest and most important lab of the day)

In this lab you will: make the data trustworthy - typed, deduplicated, incrementally maintained Silver tables - and survive the most dangerous kind of data bug: the one that's almost right

Instructor · Timing

The protected 75-minute block, straight after lunch. Do not let it start late. The scar fires around minute 45. Budget: 25 minutes to the first green checkpoint, 10 for load 2 and the failing assertion, 20 for diagnosis and fix, and the rest for step 13 hygiene.

Instructor · Say this

Set the stakes at the start: everything before lunch was about getting data in. This block is about making it worth trusting, and trust means surviving the bug that looks correct.

The story

Bronze is done. It faithfully recorded everything that arrived, including a duplicate concessions file and a buggy API load. Silver's job is to make that data trustworthy: correct types, deduplicated on business keys, incrementally maintained.

Step 1: Create your notebook

1. + New item > Notebook, name it 04_silver.
2. Attach the Mosaic lakehouse: in the left panel, click Add data items > From OneLake catalog, find and select Mosaic, and confirm.

Step 2: Add helpers and tracking

In the first cell:

%run nb_utils

Cell 2 - Silver state tracking

# Track how far each entity has been processed in Silver
spark.sql("""CREATE TABLE IF NOT EXISTS _silver_state (
  entity STRING, last_load_ts TIMESTAMP) USING DELTA""")

def last_processed(entity):
    rows = (spark.table("_silver_state")
            .filter(F.col("entity") == entity).collect())
    return rows[0].last_load_ts if rows else None

def mark_processed(entity, ts_value):
    src = spark.createDataFrame([(entity, ts_value)],
                                "entity string, last_load_ts timestamp")
    (DeltaTable.forName(spark, "_silver_state").alias("t")
     .merge(src.alias("s"), "t.entity = s.entity")
     .whenMatchedUpdateAll().whenNotMatchedInsertAll().execute())

Step 3: Build the reusable patterns

Two patterns power all of Silver. In a new cell, add these helper functions:

def changed_bronze(entity):
    """Get only the Bronze rows that arrived since last Silver run.

    The cast matters: rows loaded by your Lab 2 pipeline carry
    _load_timestamp as a STRING (Copy activity additional columns are
    always strings), while rows from your Lab 1 notebook carry a real
    timestamp. Cast once here and everything downstream compares cleanly
    - a classic mixed-ingestion gotcha."""
    df = (spark.table(f"bronze.{entity}")
          .withColumn("_load_timestamp",
                      F.col("_load_timestamp").cast("timestamp")))
    marker = last_processed(entity)
    if marker is not None:
        df = df.filter(F.col("_load_timestamp") > F.lit(marker))
    return df

def latest_per_key(df, key):
    """When multiple rows exist for the same business key, keep the latest."""
    w = Window.partitionBy(key).orderBy(F.col("_load_timestamp").desc())
    return (df.withColumn("_rn", F.row_number().over(w))
            .filter("_rn = 1").drop("_rn"))

def upsert(target, src, key):
    """MERGE into a Silver table: update existing rows, insert new ones."""
    if not spark.catalog.tableExists(target):
        src.write.saveAsTable(target)
        return
    (DeltaTable.forName(spark, target).alias("t")
     .merge(src.alias("s"), f"t.{key} = s.{key}")
     .whenMatchedUpdateAll().whenNotMatchedInsertAll().execute())

Step 4: Silver concessions - kill the duplicate

This is where the SharePoint duplicate dies. In a new cell:

changed = changed_bronze("concessions")
if changed.count() > 0:
    max_ts = changed.agg(F.max("_load_timestamp")).first()[0]
    src = (latest_per_key(changed, "sale_id")
           .select("sale_id", "event_id", "venue_id", "item_name",
                   F.col("quantity").cast("int").alias("quantity"),
                   F.col("unit_price").cast("int").alias("unit_price"),
                   F.col("sale_amount").cast("int").alias("sale_amount"),
                   F.to_timestamp("sold_at").alias("sold_at"))
           .withColumn("sale_date", F.to_date("sold_at")))
    upsert("silver.concessions", src, "sale_id")
    mark_processed("concessions", max_ts)
    log("silver.concessions", "success", src.count())

Verify the dedup worked:

assert spark.table("silver.concessions").count() == 12_000

Instructor · Say this
When the dedup lands, close the SharePoint story explicitly: 18,200 became 12,000, the phantom EUR 63,625 is gone, and it died in the right layer. Bronze recorded the lie; Silver refused to repeat it., \
    f"Expected 12000, got {spark.table('silver.concessions').count()}"
print("Concessions deduped: 18,200 Bronze → 12,000 Silver ✓")

Step 5: Silver relational tables - MERGE on business key

Now build Silver for the SQL-sourced entities. In a new cell:

INT_COLS = {"order_total", "quantity", "unit_price", "line_amount", "amount"}
TS_COLS = {"created_ts", "modified_ts"}

def conform(df):
    """Cast columns to proper types - Bronze stores everything as strings."""
    keep = [c for c in df.columns if not c.startswith("_")]
    cols = []
    for c in keep:
        if c in INT_COLS:
            cols.append(F.col(c).cast("int").alias(c))
        elif c in TS_COLS:
            cols.append(F.to_timestamp(c).alias(c))
        else:
            cols.append(F.col(c))
    return df.select(*cols, "_load_timestamp")

ENTITIES = {"attendees": "attendee_id", "ticket_orders": "order_id",
            "ticket_order_lines": "order_line_id", "payments": "payment_id",
            "refunds": "refund_id"}

# Track which order keys changed in this batch (you'll need this below)
changed_order_keys = None

for entity, key in ENTITIES.items():
    changed = changed_bronze(entity)
    if changed.count() == 0:
        continue
    max_ts = changed.agg(F.max("_load_timestamp")).first()[0]
    src = conform(latest_per_key(changed, key)).drop("_load_timestamp")
    upsert(f"silver.{entity}", src, key)
    mark_processed(entity, max_ts)
    log(f"silver.{entity}", "success", src.count())

    # Collect order keys from orders and lines for the aggregate below
    if entity in ("ticket_orders", "ticket_order_lines"):
        keys = src.select(F.col("order_id")).distinct()
        changed_order_keys = keys if changed_order_keys is None \
            else changed_order_keys.union(keys).distinct()

Step 6: Build the order financials aggregate

This is the key table for revenue reporting. It computes one row per order with gross, refunded, and net amounts. The spec requires it to recompute only the orders whose order or line rows changed this batch - this is the realistic incremental pattern.

In a new cell:

if changed_order_keys is not None and changed_order_keys.count() > 0:
    gross = (spark.table("silver.ticket_order_lines")
             .join(changed_order_keys, "order_id")
             .groupBy("order_id")
             .agg(F.sum("line_amount").alias("gross_amount")))
    refunded = (spark.table("silver.refunds")
                .groupBy("order_id")
                .agg(F.sum("amount").alias("refunded_amount")))
    fin = (gross.join(refunded, "order_id", "left")
           .fillna(0, ["refunded_amount"])
           .withColumn("net_amount",
                       F.col("gross_amount") - F.col("refunded_amount")))
    upsert("silver.order_financials", fin, "order_id")
    log("silver.order_financials", "success", fin.count())

Step 7: The reconciliation checkpoint

In a new cell:

revenue = (spark.table("silver.order_financials")
           .agg(F.sum("net_amount")).first()[0]
           + spark.table("silver.concessions")
           .agg(F.sum("sale_amount")).first()[0])
assert revenue == 2_614_339, f"expected 2614339, got {revenue}"
print(f"Revenue: EUR {revenue:,} - duplicates dead, refunds applied ✓")

Checkpoint green. The number reconciles. The SharePoint duplicate is gone. Known refunds are subtracted. This looks done.

Part B: Load 2 and the scar (30 minutes)

Step 8: Trigger the second SQL extract

The framework you built in Lab 2 delivers the poison pill itself. Open pl_ingest_master and click Run, but this time set the ceiling parameter to:

2026-09-02 00:00:00

When it finishes, check the child runs in the Output pane: four tables read 0 rows, and refunds read exactly 1 row. That row is REF-2001, a EUR 399 full cancellation of order ORD-1042. Your framework did exactly what it was built to do - extracted the one thing that changed.

Step 9: Re-run Silver

Go back to your 04_silver notebook and re-run all cells from step 5 onward (the relational entities loop, the financials aggregate, and the checkpoint).

Step 10: Watch the checkpoint fail

The revenue assertion now says:

expected 2613940, got 2614339

Off by EUR 399. Exactly the amount of the refund you just loaded. Right after you fixed everything.

Instructor · Say this

The reveal is the centerpiece of the day. Choreography that worked in rehearsal: wait for the first hands, then freeze the room. Ask who is off by exactly 399. Every hand. Ask what in this dataset costs exactly 399. Someone finds REF-2001. Then walk the innocence chain on the projector: the extract was correct (one row, the right row), the merge was correct (the refund is in silver.refunds), the aggregate logic was correct (it recomputed every changed order). Every component passed review. The system still produced a wrong number, because ORD-1042 never changed, only its refund did. Name the pattern: an incremental aggregate must be driven by the union of changed keys across every contributing table, and the driving set here forgot refunds.

Instructor · Concept

Why this bug class is the most dangerous: it is invisible at component level and only appears at reconciliation level. The defense is not smarter code review; it is a standing reconciliation assertion that runs with every load. Almost-right numbers do more damage than obviously-wrong ones because nobody escalates them.

What happened

REF-2001 merged cleanly into silver.refunds. The refund row is there, correct. But look at the financials aggregate: it recomputes only orders whose order or line rows changed this batch. ORD-1042 was created in May and never touched again, so no order or line row arrived in load 2. The refund landed, but the aggregate never re-processed that order. The correction is sitting in Silver, correct and ignored.

Think about it

Ask yourself: would you have caught a EUR 399 error on EUR 2.6 million without the assertion? The answer in every room, every time, is silence. Almost-right numbers are more dangerous than obviously wrong ones.

Step 11: Fix the aggregate

The structural fix: drive the aggregate by the union of changed keys across all contributing tables (orders, lines, refunds, payments), not just orders and lines.

Create a new cell after the financials aggregate:

# Find orders where the aggregate is stale
# (refund total in `silver.refunds` doesn't match what's in the aggregate)
fin = spark.table("silver.order_financials")
refunds_agg = (spark.table("silver.refunds").groupBy("order_id")
               .agg(F.sum("amount").alias("refunded_amount")))
stale_keys = (fin.alias("f")
              .join(refunds_agg.alias("r"), "order_id")
              .filter("f.refunded_amount <> r.refunded_amount")
              .select("order_id").distinct())

print(f"Stale orders found: {stale_keys.count()}")

# Recompute financials for stale orders
gross = (spark.table("silver.ticket_order_lines")
         .join(stale_keys, "order_id").groupBy("order_id")
         .agg(F.sum("line_amount").alias("gross_amount")))
fixed = (gross.join(refunds_agg, "order_id", "left")
         .fillna(0, ["refunded_amount"])
         .withColumn("net_amount",
                     F.col("gross_amount") - F.col("refunded_amount")))

(DeltaTable.forName(spark, "silver.order_financials").alias("t")
 .merge(fixed.alias("s"), "t.order_id = s.order_id")
 .whenMatchedUpdateAll().execute())
log("silver.order_financials", "success", fixed.count(),

Instructor · Beyond the lab
After the fix lands, offer the upgrade ladder: (1) collect changed keys from all four contributing tables at merge time, the proactive version of today's reactive stale-scan; (2) event-driven recompute, where each merged entity emits keys to a recompute queue; (3) periodic full rebuild of the aggregate as a safety net, cheap at Silver scale; (4) the reconciliation assertion promoted into the pipeline as a gate, which Lab 9 turns into a framework.
    "late refund backfill: union-of-changed-keys rule")

Step 12: The trusted number

Run the standing assertion one more time:

revenue = (spark.table("silver.order_financials")
           .agg(F.sum("net_amount")).first()[0]
           + spark.table("silver.concessions")
           .agg(F.sum("sale_amount")).first()[0])
assert revenue == 2_613_940, f"expected 2613940, got {revenue}"
print(f"Trusted revenue: EUR {revenue:,} ✓")

EUR 2,613,940. That is the number the day has been building toward.

Step 13: Make the trusted number your standing assertion

Instructor · Watch for

Step 13 is the one attendees skip, and it detonates two hours later in Lab 7 when the orchestrator runs this notebook top to bottom against the interim number. Helpers should sweep during the break: open each struggling attendee's 04_silver, confirm the checkpoint cell asserts 2_613_940.

One housekeeping step that matters later: go back to the checkpoint cell from step 7 (the one asserting 2,614,339) and change it to assert 2_613_940. From now on, this notebook has exactly one revenue assertion - the trusted number - and it must pass every time the notebook runs top to bottom. In Lab 7 an orchestrator will run this notebook unattended, and an assertion pinned to the interim number would fail the whole platform.

The lesson

An incremental aggregate must be driven by the union of changed keys across every table that contributes to it - not just the driving table. The watermark discipline that made Bronze safe is exactly what hid the correction in Silver. The only defense is a reconciliation assertion checked on every run.

Quick reference

Lab 5 - Gold star schema

Time: 25 minutes

In this lab you will: model the Gold star schema - facts at their natural grain, denormalized dimensions - and prove the trusted number at the consumption layer

Instructor · Timing

25 minutes and intentionally lighter after the Lab 4 climax. Announce the collapsible solutions up front: exercises have a Show solution toggle, use it if stuck, no heroics. Dimensions are quick; the facts are the substance.

Instructor · Concept

The design conversation worth having out loud: why net revenue is a measure, never a stored column. Storing it couples the fact to a business definition that changes (does net include concessions? refunds of concessions?). Each fact keeps its own clean grain; the subtraction lives in the semantic model where definitions are versioned and visible.

The story

Silver is trusted - duplicates dead, refunds applied, types correct. But Silver tables mirror the operational schema. A semantic model needs a star schema: dimensions describe the "who, what, when, where" and facts record the measurable events at their natural grain.

Step 1: Create your notebook

1. + New item > Notebook, name it 05_gold.
2. Attach the Mosaic lakehouse: in the left panel, click Add data items > From OneLake catalog, find and select Mosaic, and confirm.

Step 2: Helpers

%run nb_utils

Step 3: Build dimensions

In a new cell, build the event dimension (denormalized with venue attributes - no snowflaking):

events = spark.table("bronze.events").select(
    "event_id", "event_name", "venue_id", "start_date", "end_date",
    "event_type").dropDuplicates(["event_id"])
venues = spark.table("bronze.venues").select(
    "venue_id", "venue_name", "city", "country_code",
    F.col("capacity").cast("int").alias("capacity")
).dropDuplicates(["venue_id"])

(events.join(venues, "venue_id", "left")
 .select("event_id", "event_name", "event_type",
         F.to_date("start_date").alias("start_date"),
         F.to_date("end_date").alias("end_date"),
         "venue_id", "venue_name", "city", "country_code", "capacity")
 .write.mode("overwrite").saveAsTable("gold.dim_event"))
print(f"gold_dim_event: {spark.table('gold.dim_event').count()} rows")

Now build the remaining dimensions. In new cells, write each one yourself using the pattern above. Here is what each should contain:

Your task: gold_dim_attendee - from silver.attendees, columns: attendee_id, first_name, last_name, company, country_code

(spark.table("silver.attendees")
 .select("attendee_id", "first_name", "last_name", "company", "country_code")
 .write.mode("overwrite").saveAsTable("gold.dim_attendee"))
print(f"gold.dim_attendee: {spark.table('gold.dim_attendee').count()} rows")

Your task: gold_dim_product - from bronze.products, columns: product_id, event_id, product_name, unit_price (cast to int), deduplicated on product_id

(spark.table("bronze.products")
 .select("product_id", "event_id", "product_name",
         F.col("unit_price").cast("int").alias("unit_price"))
 .dropDuplicates(["product_id"])
 .write.mode("overwrite").saveAsTable("gold.dim_product"))
print(f"gold.dim_product: {spark.table('gold.dim_product').count()} rows")

gold_dim_date - a generated calendar table for 2026:

(spark.sql("""SELECT explode(sequence(to_date('2026-01-01'),
                                      to_date('2026-12-31'))) AS date""")
 .select(F.date_format("date", "yyyyMMdd").cast("int").alias("date_key"),
         F.col("date"), F.year("date").alias("year"),
         F.month("date").alias("month"), F.quarter("date").alias("quarter"),
         F.date_format("date", "MMMM").alias("month_name"))
 .write.mode("overwrite").saveAsTable("gold.dim_date"))

Step 4: Build fact tables

Think about it

**Why .repartition(20) below?** At workshop scale, Spark writes each fact table as a single file. In production, daily appends and concurrent writers create fragmented file layouts naturally. The repartition here simulates that fragmentation so you can see OPTIMIZE do real work in Lab 6. You would never add .repartition(20) in a production pipeline.

Instructor · Say this

Read the repartition disclaimer to the room verbatim. In rehearsal this was the number-one candidate for cargo-culting into someone's real project. Say the sentence: this line exists so Lab 6 has something to compact, delete it from anything you build at work.

Instructor · Watch for

If someone's session restarted between cells, the orders DataFrame is gone and fact builds throw NameError: re-run from the fact_ticket_sales cell. Grain check for helpers: fact_ticket_sales 8,016 rows, fact_concessions 12,000, fact_refunds 242.

Three facts, each at its natural grain. In new cells:

gold_fact_ticket_sales - grain: one row per order line

orders = spark.table("silver.ticket_orders").select(
    "order_id", "attendee_id", "event_id",
    F.to_date("created_ts").alias("order_date"))

(spark.table("silver.ticket_order_lines")
 .join(orders, "order_id")
 .select("order_line_id", "order_id", "attendee_id", "event_id", "product_id",
         F.date_format("order_date", "yyyyMMdd").cast("int").alias("date_key"),
         "quantity", "unit_price", "line_amount")
 .repartition(20).write.mode("overwrite").saveAsTable("gold.fact_ticket_sales"))

gold_fact_concessions - grain: one row per POS sale

(spark.table("silver.concessions")
 .select("sale_id", "event_id", "venue_id", "item_name",
         F.date_format("sale_date", "yyyyMMdd").cast("int").alias("date_key"),
         "quantity", "unit_price", "sale_amount")
 .repartition(20).write.mode("overwrite").saveAsTable("gold.fact_concessions"))

gold_fact_refunds - grain: one row per refund. Build this one yourself:

• Join silver.refunds to the orders DataFrame on order_id to get attendee_id and event_id
• Derive date_key from the refund's created_ts
• Select: refund_id, order_id, attendee_id, event_id, reason, date_key, amount

Production note

Design note: net revenue is a measure definition (ticket sales + concessions - refunds), not a stored column. This is fork 3's contract with the semantic model - each fact keeps its own clean grain, and the subtraction happens in the measure.

Step 5: Final checkpoint

net = (spark.table("gold.fact_ticket_sales")
       .agg(F.sum("line_amount")).first()[0]
       + spark.table("gold.fact_concessions")
       .agg(F.sum("sale_amount")).first()[0]
       - spark.table("gold.fact_refunds")
       .agg(F.sum("amount")).first()[0])
assert net == 2_613_940, f"expected 2613940, got {net}"
print(f"Trusted revenue: EUR {net:,} - and you can prove it. ✓")

The same number, now in Gold, at the right grain, ready for a semantic model.

Grain trap to watch for

If you joined gold.fact_ticket_sales to payments at the order grain, a 3-line order would count its payment 3 times.

Instructor · Beyond the lab

Extensions for questions that will come: surrogate keys are unnecessary at this scale with stable business keys, but SCD2 dimensions would need them; incremental Gold (MERGE instead of overwrite) becomes worthwhile when rebuild time exceeds minutes; aggregation tables for dashboard-speed queries sit on top, and Direct Lake makes many of them unnecessary. Facts join to each other through dimensions, never directly - unless they share the same grain. This is the most common mistake in star schema design, and it is one of the smells planted in the disaster workspace you'll investigate at the end of the day.

Quick reference

Lab 6 - Optimize for Direct Lake

Time: 15 minutes

In this lab you will: compact and V-Order your Gold tables so a Direct Lake model frames fast, loads columns lean, and never surprises you

Instructor · Timing

15 minutes, mostly runtime. Narrate while OPTIMIZE runs.

Instructor · Concept

The mental model to give: Direct Lake reads Delta files straight from OneLake into memory on first query (framing). Every file is a separate read; every unsorted column costs memory. OPTIMIZE fixes file count, V-Order fixes column layout, VACUUM pays the storage bill. The trio is the last mile between correct data and fast reports.

Instructor · Say this

The maintenance-caboose principle: OPTIMIZE and VACUUM run at the end of every load, wired into the pipeline, never a quarterly heroic cleanup. Lab 7 wires exactly that.

Instructor · Beyond the lab

Worth naming: Fabric's automatic table maintenance features (optimized write, auto-compaction) reduce manual OPTIMIZE needs over time; deletion vectors change VACUUM behavior; monitoring file counts per table over weeks is a cheap early-warning metric. VACUUM retention below 168 hours deletes time travel, say it once, sternly.

The story

Your Gold tables are correct, but they were written by iterative notebook runs that produced many small files. A Direct Lake model reads Delta tables directly from OneLake - and every file means a separate read. File fragmentation is the most common cause of slow first-query latency and memory pressure in Direct Lake.

Step 1: Create your notebook

1. + New item > Notebook, name it 06_optimize.
2. Attach the Mosaic lakehouse (Add data items > From OneLake catalog > Mosaic).
3. In the first cell:

%run nb_utils

Step 2: Measure the current state

Note

About the file counts below: the fact tables were intentionally fragmented in Lab 5 (via .repartition(20)) to simulate what happens in production over weeks of daily appends. Dimension tables are still one file each, which is normal for small reference data.

Before optimizing, record how fragmented your tables are. In the first cell:

GOLD_TABLES = ["gold.fact_ticket_sales", "gold.fact_concessions",
               "gold.fact_refunds", "gold.dim_event", "gold.dim_attendee",
               "gold.dim_product", "gold.dim_date"]

before = {}
for t in GOLD_TABLES:
    d = spark.sql(f"DESCRIBE DETAIL {t}").first()
    before[t] = (d.numFiles, d.sizeInBytes)
    print(f"  {t}: {d.numFiles} files, {d.sizeInBytes:,} bytes")

Note

Note the file counts. Even at workshop scale, you'll likely see tables with more files than necessary. In production with daily appends, this gets dramatically worse.

Step 3: Run OPTIMIZE with V-Order

OPTIMIZE compacts small files into larger ones. V-Order is a Fabric-specific encoding that arranges data for faster column reads - exactly what Direct Lake needs.

In a new cell:

for t in GOLD_TABLES:
    spark.sql(f"OPTIMIZE {t} VORDER")
    print(f"  optimized {t}")

Step 4: Run VACUUM

OPTIMIZE leaves the old small files behind (for time-travel). VACUUM removes files older than the retention period. The default retention is 7 days (168 hours) - going lower requires a configuration override, and doing so in production deletes your time-travel safety net. Don't.

In a new cell:

for t in GOLD_TABLES:
    spark.sql(f"VACUUM {t} RETAIN 168 HOURS")
    print(f"  vacuumed {t}")

Step 5: Measure the improvement

In a new cell:

print("\nBefore → After:")
for t in GOLD_TABLES:
    d = spark.sql(f"DESCRIBE DETAIL {t}").first()
    b_files, b_bytes = before[t]
    print(f"  {t}: {b_files} → {d.numFiles} files, "
          f"{b_bytes:,} → {d.sizeInBytes:,} bytes")

assert all(
    spark.sql(f"DESCRIBE DETAIL {t}").first().numFiles <= before[t][0]
    for t in GOLD_TABLES)
print("\nGold is Direct Lake ready ✓")

What you just did, and why it matters for Direct Lake

OPTIMIZE merged small Delta files into larger, more efficient ones. Fewer files means fewer I/O operations when the Direct Lake engine frames a table into memory.

V-Order applied a special columnar encoding optimized for Fabric's Parquet reader. It reduces the memory footprint per column and speeds up the initial column load when a report first opens.

VACUUM reclaimed storage from the superseded small files. Without it, OneLake keeps growing with dead files that nothing reads but still cost storage.

In production, these three commands should run after every significant write to Gold - typically at the end of your nightly pipeline, not ad hoc.

Quick reference

Lab 7 - End-to-end orchestration

Time: 25 minutes guided (the failure exercise may run as a presenter demo if the day is tight; every step below works self-paced)

In this lab you will: wire ingestion, Silver, Gold, and maintenance into one master pipeline with failure logging - then run it end to end, break it on purpose, and watch the failure propagate through the Monitoring Hub.

Instructor · Timing

25 minutes guided. If the room is behind, build pl_mosaic_work and pl_run_all together, run the success path, and demo the break-it section from the front. The failure demo from the projector loses nothing.

Instructor · Concept

The architecture lesson is separation of work and observability, and it was earned live: our first design wired four On Failure edges into one logging activity, and none of them fired, because Fabric waits for every upstream to evaluate and skipped activities never do. One Invoke wrapper, one success edge, one failure edge: no ambiguity. Tell that story; design lessons from failures stick.

The story

Everything you built today runs by hand. Production doesn't work that way: one orchestrator runs the whole platform on a schedule, logs its own history, and makes failures impossible to miss. A pipeline that fails without telling anyone is worse than one that fails visibly, and both are worse than one that logs its own history.

By the end of this lab, "is the data fresh?" stops being a question someone asks in a chat and becomes a query.

Prerequisite check

Before starting, confirm your Lab 4 notebook's standing assertion is the trusted number (2,613,940). The interim checkpoint value must be gone from that notebook. If you haven't updated it yet, do it now - the orchestrator will run that notebook top to bottom, and an assertion pinned to the old number will fail the whole run.

Step 1: Create the run-logging notebook

1. + New item > Notebook, name it exactly nb_log_run. Attach the Mosaic lakehouse (Add data items > From OneLake catalog > Mosaic).
2. In the first cell:

%run nb_utils

2. First cell - mark it as the parameter cell (cell ... menu > Toggle parameter cell):

run_id = ""
pipeline_name = ""
run_status = ""
message = ""

3. Second cell:

from pyspark.sql import functions as F

spark.sql("""CREATE TABLE IF NOT EXISTS _pipeline_log (
  run_id STRING, item STRING, status STRING, rows BIGINT,
  message STRING, logged_at TIMESTAMP) USING DELTA""")

(spark.createDataFrame(
    [(run_id, pipeline_name, run_status, 0, message)],
    "run_id string, item string, status string, rows long, message string")
 .withColumn("logged_at", F.current_timestamp())
 .write.mode("append").saveAsTable("_pipeline_log"))
print(f"logged: {pipeline_name} -> {run_status}")

Save it. The pipeline runs it, you don't.

Step 2: Build the work pipeline

The work pipeline chains all four processing steps. The orchestrator (next step) wraps it and handles logging.

1. + New item > Data pipeline, name it pl_mosaic_work.
2. Add a ceiling parameter (String, default 2026-09-02 00:00:00).
3. Add four activities chained On Success:

• Invoke pipeline Ingest: invokes pl_ingest_master, passes ceiling = @pipeline().parameters.ceiling, Wait on completion ON
• Notebook Silver: your 04_silver notebook
• Notebook Gold: your 05_gold notebook
• Notebook Maintenance: your 06_optimize notebook

4. Save.

Step 3: Build the orchestrator

The orchestrator's only job is to invoke the work pipeline and log the outcome. One success edge, one failure edge, no ambiguity.

1. + New item > Data pipeline, name it pl_run_all.
2. Add a ceiling parameter (String, default 2026-09-02 00:00:00).
3. Add one Invoke pipeline activity named Invoke mosaic work, pointing to pl_mosaic_work, passing ceiling = @pipeline().parameters.ceiling, Wait on completion ON.

1. + New item > Data pipeline, name it pl_run_all.
2. Add the activities below, chaining each with the On success (green) connector from the previous one:

Activity 1 - Invoke pipeline: name Ingest, invoked pipeline pl_ingest_master, Wait on completion ON. Parameters: ceiling = 2026-09-02 00:00:00 (the current high-water mark - everything through load 2).

Activity 2 - Notebook: name Silver, notebook = your 04_silver notebook.

Activity 3 - Notebook: name Gold, notebook = your 05_gold notebook.

Activity 4 - Notebook: name Maintenance, notebook = your 06_optimize notebook. Maintenance is the caboose of every load - part of the pipeline, not a heroic quarterly cleanup.

Step 3: Wire the failure and success logging

5. Add a Notebook activity named Log success. Connect On success from Maintenance to it. Settings: notebook nb_log_run, base parameters (dynamic content):

• run_id = @pipeline().RunId
• pipeline_name = @pipeline().Pipeline
• run_status = Succeeded
• message = ''

6. Add a Notebook activity named Log failure. Connect the On failure (red) output of all four main activities (Ingest, Silver, Gold, Maintenance) into this one activity - multiple failure edges into one activity act as OR. Same notebook, same parameters, except:

• run_status = Failed
• message = @activity('Invoke mosaic work').error.message

Production note

Production note: in a real deployment you'd add an alert after Log failure - an Activator rule on the failure event, or a Teams or email activity. The workshop tenant has no M365 services, so today the log row plus the Monitoring Hub is your alarm. The shape is identical.

7. Save.

Step 4: The end-to-end run

1. Click Run.
2. While it runs, open the Monitoring Hub (left navigation > Monitor). Filter to your workspace. Watch pl_run_all fan out: the Ingest activity spawns pl_ingest_master, which fans out into five pl_ingest_child runs, then Silver, Gold, and Maintenance execute in sequence.
3. Drill into one child run and check rows read: it should be 0 - the ceiling hasn't moved, so the window is empty. This morning the same copy read thousands of rows.

When it finishes, verify in a notebook:

from pyspark.sql import functions as F

# The trusted number survived a full end-to-end re-run
net = (spark.table("gold.fact_ticket_sales").agg(F.sum("line_amount")).first()[0]
       + spark.table("gold.fact_concessions").agg(F.sum("sale_amount")).first()[0]
       - spark.table("gold.fact_refunds").agg(F.sum("amount")).first()[0])
assert net == 2_613_940, f"expected 2613940, got {net}"

# And the run logged its own history
spark.table("_pipeline_log").orderBy(F.col("logged_at").desc()).show(5, truncate=False)

Zero new rows ingested, same trusted number, a Succeeded row in the log. Incremental + idempotent = re-runnable at 3 a.m. without fear. That sentence is the payoff of every MERGE and watermark you wrote today.

Instructor · Say this

Land the 3am sentence slowly and connect it backward: file tracking from Lab 1, watermarks from Lab 2, business-key MERGE from Lab 4, overwrite Gold from Lab 5. Every piece existed so this one run could be boring. Boring is the achievement.

Step 5: Break it on purpose

Trust an error path you've tested; distrust one you haven't.

Instructor · Say this

The break-it philosophy line that worked in rehearsal: an untested error path is a decorative error path. You are not testing whether it fails, you are testing whether you find out.

Instructor · Watch for

Three wiring mistakes dominate: the error-message dynamic content references an activity name that does not match (must be the exact Invoke activity name); the ceiling parameter added to only one of the two pipelines; and 04_silver still asserting the interim number, which fails the whole chain at the Silver step, a useful teachable failure if it happens publicly.

Instructor · Beyond the lab

The alerting story for the debrief: on a full tenant this failure edge feeds an Activator rule or a Teams webhook; the run log becomes the freshness SLA source (last Succeeded timestamp per pipeline); and deployment pipelines or fabric-cicd promote these pipelines dev to test to prod instead of hand-building per workspace.

1. In a notebook, corrupt one control-table row:

spark.sql("""UPDATE meta_ingestion_control
             SET watermark_column = 'NoSuchColumn'
             WHERE source_table = 'payments'""")

2. Run pl_run_all again.
3. Watch the failure propagate: the payments child fails on an invalid column name → the ForEach reports the failure → pl_ingest_master fails → the Ingest activity in pl_run_all fails → Log failure fires.
4. Open the Monitoring Hub and find the actual SQL error: drill into pl_run_all > Ingest > pl_ingest_master > the failed child > the Copy activity output. The real error ("Invalid column name 'NoSuchColumn'") is three levels deep - which is exactly why you test children standalone before wiring them into a parent.
5. Verify the failure was logged:

spark.table("_pipeline_log").filter("status = 'Failed'").show(truncate=False)

Step 6: Fix it and prove recovery

1. Repair the control table:

spark.sql("""UPDATE meta_ingestion_control
             SET watermark_column = 'modified_ts'
             WHERE source_table = 'payments'""")

2. Run pl_run_all once more. It succeeds.
3. Look at the log: a Failed row followed by a Succeeded row. That pair of rows is your incident report - when it broke, when it recovered, no chat archaeology required.

Step 7: Schedule it (2 minutes)

Open pl_run_all > Schedule (top toolbar) > daily at 06:00. You won't keep the schedule today, but look at what production turns into: this orchestrator, plus a schedule, plus the run log is the operations story.

Quick reference

Lab 8 - Disaster forensics

Time: presented as the closing debrief; the read-only workspace stays open, so you can run this checklist yourself during the catch-up slot or after the workshop

In this lab you will: diagnose the lakehouse Mosaic inherited - the one that started this whole story - and connect every smell back to something you built correctly today

The story

Before you arrived, another team built the Mosaic platform. Finance stopped trusting its numbers months ago. Nobody can explain why. The build is in a shared read-only workspace - you can read everything but change nothing.

Your job is to diagnose, not fix. Each item you find is a production anti-pattern from the morning's labs. The person (or pair) who finds the most smells wins bragging rights.

Instructor · Timing

Presented as the closing debrief (10 minutes from the front). Attendees can run the checklist themselves during catch-up or after; the workspace stays shared read-only.

Instructor · Say this

Debrief script skeleton: walk smells 1 through 8 fast, two sentences each, always naming the lab that fixed it. Then the bridge: mentally repair the disaster one smell at a time and you land near EUR 2,847,310, the ghost from 9:00 this morning. Same disease, earlier stage. Close the arc: at 9:00 you inherited a number nobody trusted; at 17:00 you can prove EUR 2,613,940 and show where every euro of the difference went, a renamed file, a buggy API, a late refund. Then the production checklist slide: provenance columns, business-key dedup, metadata-driven ingestion, union-of-changed-keys aggregates, reconciliation assertions, maintenance in the pipeline, one orchestrator that logs and fails visibly.

Instructor · Watch for

Presenter prep the day before: run the disaster build in the shared workspace, import the broken mosaic_revenue_report, share the workspace read-only to the attendee security group, and open it once from an attendee account to verify permissions.

Setup

1. Your presenter will give you the name of the disaster workspace. Open it from the Workspaces menu.
2. Everything in this workspace is read-only. You can open notebooks and run read-only queries (SELECT, DESCRIBE DETAIL, COUNT) but you cannot modify tables or code.

The investigation

Work through these questions. For each one, write down the evidence (a query result, a line of code, a count) and which production principle it violates.

Question 1: Provenance

Pick any Bronze table. Can you tell which file or extract any given row came from, and when it arrived?

Note

Hint: look at the columns. Compare to what your Bronze tables have.

Question 2: Duplicate loads

Count the rows in bronze.ticket_order_lines. The source database has 8,016 rows. What do you see? What does that tell you about how the loader was operated?

spark.table("bronze.ticket_order_lines").count()

Question 3: The file that shouldn't be there

List what's in the landing folder. Look at the file names carefully.

for f in notebookutils.fs.ls("Files/landing/batch1"):
    print(f.name)
for f in notebookutils.fs.ls("Files/landing/batch2"):
    print(f.name)

Does anything look familiar from this morning?

Question 4: The revenue report's grain bug

Open the notebook called mosaic_revenue_report (pre-loaded in this disaster workspace by the presenter). Read the code in the first code cell (don't run it - just read).

• At what grain does it join payments to order lines?
• Write down what happens to a 3-line order's payment amount in this join.
• What direction does this push the revenue number?

Question 5: What layer does the report read from?

Still looking at the revenue report notebook: which tables does it query? Bronze, Silver, or Gold? What does that mean for every other consumer of this data?

Question 6: The missing entity

Search the entire workspace for the word "refund." What do you find?

Note

Hint: look at the SQL extract logic, the Bronze tables, the report. Is there a refunds table anywhere?

Question 7: File fragmentation

Run this on the gold.sales_snapshot table:

d = spark.sql("DESCRIBE DETAIL gold_sales_snapshot").first()
print(f"Files: {d.numFiles}, Size: {d.sizeInBytes:,} bytes")
print(f"Avg file size: {d.sizeInBytes // max(d.numFiles, 1):,} bytes")

How many files? What's the average file size? What does a Direct Lake query over this table have to do?

Question 8: The TODO comment

Find the TODO comment in the revenue report. It says "numbers look a bit high since March?" Given everything you've found, is "a bit high" the right diagnosis?

Scoring

The debrief (presenter-led)

Here is the punchline: mentally fix the snapshot one smell at a time. Correct the join grain, deduplicate the double loads, remove the renamed file, and you arrive at a number around EUR 2,847,310. That is the ghost number - the prototype's answer from this morning's checkpoint. A build with correct-looking mechanics that still double-counts one renamed file and ignores refunds.

The disaster workspace and the prototype are the same disease at different stages. The trusted EUR 2,613,940 exists only because every layer earned it - and you can prove exactly how.

Lab 9 - Data quality framework

In this lab you will: build a rules table that defines quality checks per entity, an engine that evaluates them, and a log that records every result. Adding a quality rule becomes a row, not code.

Note

This is a bonus lab for attendees who finish ahead of the group. It follows the same metadata-driven philosophy as Lab 2's ingestion framework.

Instructor · Concept

For anyone who reaches this lab, the meta-lesson is that config-as-data generalizes: ingestion control in Lab 2, quality rules here, and (mention it) the same shape works for masking rules, retention policies, and export contracts. One pattern, many frameworks.

Instructor · Beyond the lab

The wiring that turns this into production: a quality gate activity between Silver and Gold in pl_mosaic_work, where error severity blocks the Gold write and warning severity logs and continues. Compare honestly with the ecosystem: this is a teaching-scale version of what dbt tests or Great Expectations provide, and the build-vs-adopt decision belongs to team standards, not preference.

The concept

Just like the ingestion control table tells the pipeline what to load, a quality rules table tells a quality engine what to check. Each rule is a row: which table, which column, what kind of check, what threshold. The engine reads the rules, runs each one, and logs pass or fail. No hardcoded assertions scattered across notebooks.

Step 1: Create the rules table

1. + New item > Notebook, name it 09_data_quality. Attach the Mosaic lakehouse.
2. Load shared utilities:

%run nb_utils

3. Create the rules table:

# Each row is one quality rule. The engine evaluates them all.
rules = [
    # rule_id, schema, table, rule_type, column, parameters, severity
    ("DQ-001", "silver", "concessions",         "not_null",   "sale_id",    "{}", "error"),
    ("DQ-002", "silver", "concessions",         "not_null",   "sale_amount", "{}", "error"),
    ("DQ-003", "silver", "concessions",         "positive",   "sale_amount", "{}", "error"),
    ("DQ-004", "silver", "concessions",         "row_count",  None,          '{"min": 10000}', "error"),
    ("DQ-005", "silver", "ticket_orders",       "not_null",   "order_id",    "{}", "error"),
    ("DQ-006", "silver", "ticket_orders",       "not_null",   "order_total", "{}", "error"),
    ("DQ-007", "silver", "ticket_orders",       "positive",   "order_total", "{}", "warning"),
    ("DQ-008", "silver", "ticket_order_lines",  "not_null",   "order_id",    "{}", "error"),
    ("DQ-009", "silver", "ticket_order_lines",  "referential","order_id",    '{"parent": "silver.ticket_orders", "parent_key": "order_id"}', "error"),
    ("DQ-010", "silver", "refunds",             "not_null",   "refund_id",   "{}", "error"),
    ("DQ-011", "silver", "refunds",             "referential","order_id",    '{"parent": "silver.ticket_orders", "parent_key": "order_id"}', "error"),
    ("DQ-012", "gold",   "fact_ticket_sales",   "row_count",  None,          '{"min": 5000}', "error"),
    ("DQ-013", "gold",   "fact_ticket_sales",   "unique",     "order_line_id", "{}", "error"),
    ("DQ-014", "gold",   "fact_refunds",        "row_count",  None,          '{"min": 200}', "error"),
]

schema = "rule_id string, schema string, table string, rule_type string, column string, parameters string, severity string"
(spark.createDataFrame(rules, schema)
 .write.mode("overwrite").saveAsTable("meta_quality_rules"))
spark.table("meta_quality_rules").show(truncate=False)

Step 2: Build the quality engine

The engine reads every rule, evaluates it against the target table, and returns pass/fail with a message.

import json as json_lib

def evaluate_rule(rule):
    """Evaluate one quality rule. Returns (pass/fail, message)."""
    full_table = f"{rule.schema}.{rule.table}"
    params = json_lib.loads(rule.parameters) if rule.parameters else {}

    try:
        df = spark.table(full_table)
    except Exception as e:
        return ("fail", f"Table {full_table} not found: {e}")

    if rule.rule_type == "not_null":
        nulls = df.filter(F.col(rule.column).isNull()).count()
        if nulls > 0:
            return ("fail", f"{nulls} null values in {rule.column}")
        return ("pass", f"No nulls in {rule.column}")

    elif rule.rule_type == "positive":
        negatives = df.filter(F.col(rule.column) <= 0).count()
        if negatives > 0:
            return ("fail", f"{negatives} non-positive values in {rule.column}")
        return ("pass", f"All values positive in {rule.column}")

    elif rule.rule_type == "unique":
        total = df.count()
        distinct = df.select(rule.column).distinct().count()
        if total != distinct:
            return ("fail", f"{total - distinct} duplicates in {rule.column}")
        return ("pass", f"All values unique in {rule.column}")

    elif rule.rule_type == "row_count":
        count = df.count()
        min_count = params.get("min", 0)
        if count < min_count:
            return ("fail", f"Row count {count} below minimum {min_count}")
        return ("pass", f"Row count {count} meets minimum {min_count}")

    elif rule.rule_type == "referential":
        parent_table = params["parent"]
        parent_key = params["parent_key"]
        parent_df = spark.table(parent_table)
        orphans = (df.join(parent_df, df[rule.column] == parent_df[parent_key], "left_anti")
                   .count())
        if orphans > 0:
            return ("fail", f"{orphans} orphan records: {rule.column} not in {parent_table}.{parent_key}")
        return ("pass", f"All {rule.column} values found in {parent_table}")

    return ("fail", f"Unknown rule type: {rule.rule_type}")

Step 3: Run all rules and log results

# Create the quality log table
spark.sql("""CREATE TABLE IF NOT EXISTS _quality_log (
  run_id STRING, rule_id STRING, target STRING, rule_type STRING,
  severity STRING, result STRING, message STRING,
  checked_at TIMESTAMP) USING DELTA""")

# Evaluate every rule and log the result
rules_df = spark.table("meta_quality_rules").collect()
results = []

for rule in rules_df:
    result, message = evaluate_rule(rule)
    full_table = f"{rule.schema}.{rule.table}"
    results.append((RUN_ID, rule.rule_id, full_table, rule.rule_type,
                    rule.severity, result, message))
    icon = "PASS" if result == "pass" else "FAIL"
    sev = f"[{rule.severity.upper()}]" if result == "fail" else ""
    print(f"  {icon} {sev} {rule.rule_id}: {full_table}.{rule.column or '*'} ({rule.rule_type}) - {message}")

# Write all results in one batch
(spark.createDataFrame(results,
    "run_id string, rule_id string, target string, rule_type string, severity string, result string, message string")
 .withColumn("checked_at", F.current_timestamp())
 .write.mode("append").saveAsTable("_quality_log"))

# Summary
passed = sum(1 for r in results if r[5] == "pass")
failed = sum(1 for r in results if r[5] == "fail")
print(f"\n{passed} passed, {failed} failed out of {len(results)} rules")

Step 4: Query the quality history

# Most recent results per rule
(spark.table("_quality_log")
 .filter(F.col("run_id") == RUN_ID)
 .orderBy("rule_id")
 .select("rule_id", "target", "rule_type", "severity", "result", "message")
 .show(truncate=False))

Step 5: Add your own rule

Your task: add a rule that checks whether every gold.fact_ticket_sales row has a valid date_key that exists in gold.dim_date. This is a referential integrity check.

spark.sql("""INSERT INTO meta_quality_rules VALUES
  ('DQ-015', 'gold', 'fact_ticket_sales', 'referential', 'date_key',
   '{"parent": "gold.dim_date", "parent_key": "date_key"}', 'error')""")

The production principle

This framework follows the same pattern as the ingestion control table from Lab 2: configuration as data, not code. Adding a quality check is one INSERT. The engine is generic. The log is queryable. In production, you would wire this into the orchestrator pipeline (Lab 7) as a step between Silver and Gold, with error-severity failures blocking the Gold write and warning-severity failures logging but continuing.

Quick reference