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 card: email, password, and this site's address. Your first sign-in asks for a new password; write it on the card
• Sign in using an InPrivate / Incognito window
• Nothing to install, nothing to download
• A Solutions folder in your workspace with a ready-made notebook for every notebook lab: fall behind, run the solution, keep going. Nobody gets left behind at a checkpoint

Ready? Start with Lab 0: Environment setup.

FAQ and concepts

Short, plain explanations of the ideas the labs are built on. Read it before the day starts or come back to it whenever a term goes by too fast.

The architecture

Getting data in

Making data trustworthy

Shaping for consumption

Running it

Fabric mechanics you will touch today

Lab 0 - Environment setup

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

In this lab you will: sign in, create the lakehouse schemas, set up your scratch notebook, and verify seed data and SQL connectivity

Step 1: Sign in

The card at your seat has three lines: Email, Password, and Lab url.

1. Open Microsoft Edge or Google Chrome in an InPrivate / Incognito window. This keeps your own corporate account out of the way.
2. Go to https://app.fabric.microsoft.com.
3. Sign in with the Email from your card (it looks like lake-user-047@fabconbar26.onmicrosoft.com) and the Password printed below it.
4. You will be asked to set a new password. Choose one and write it on your card. The printed password stops working the moment you change it, and nobody in the room can reset it for you.
5. There is no authenticator or MFA prompt. If prompted to stay signed in, click Yes.
6. The Lab url on your card is this site. Open it in a second tab.

Trouble?

Trouble? "You need an invitation", a Conditional Access error, or a password screen that loops: raise your hand. A helper will hand you a spare card. Do not try your own corporate or trial account.

Step 2: Find your workspace

1. Click Workspaces in the left sidebar.
2. Your workspace is named mosaic-att-NNN, where NNN is the number in your account name (lake-user-047 owns mosaic-att-047).
3. Click it to open it.

You should see:

• A lakehouse named Mosaic
• A notebook named nb_utils: shared utilities that every other notebook loads with one line
• A notebook named 00_setup_schemas: you run it once in the next step
• A notebook named 90_release_batch2: you will use it in Lab 3
• A folder named Solutions containing nine SOLUTION_ notebooks, one per notebook-based lab. If you fall behind at any point, open the solution for that lab, click Run all, and continue with the next lab

Step 3: Create the lakehouse schemas

1. Open 00_setup_schemas. The Mosaic lakehouse is already attached, as it is for every pre-provisioned notebook.
2. Click Run all in the toolbar. The first run takes a minute or two while a Spark session starts.
3. You should see: Done - bronze, silver, and gold schemas created
4. Go back to your workspace and open the Mosaic lakehouse. Under Schemas you now see bronze, silver, and gold alongside dbo.

Step 4: Verify the seed 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.

Step 5: Create your scratch notebook

You will use one notebook all day for checks and queries that do not belong in a pipeline. Every lab that says "in 00_scratch" means this one.

1. Go back to your workspace (click the workspace name in the breadcrumb at the top).
2. Click + New item > Notebook. Name it exactly 00_scratch.
3. Attach the Mosaic lakehouse: in the left panel, click Add data items > From OneLake catalog, find and select Mosaic, and confirm.
4. Before running anything, open the Connect dropdown in the notebook ribbon and choose New high concurrency session. Every notebook you create today attaches to this one session instead of starting its own: faster for you (attaching takes two seconds), and lighter on the capacity the whole room shares.
5. In the first cell, paste and run (click the play button on the left of the cell, or press Shift+Enter). New notebooks open with a first cell full of comment lines; delete them first, because the cell must contain only this one line.

%run nb_utils

6. In a new cell, test the connection to the shared Azure SQL database:

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}")

7. You should see: ticket_orders row count: 5151

Trouble?

Trouble? A connection error here is almost always a firewall or a typo made while editing the cell. The cell needs no edits. Raise your hand.

Step 6: Keep the scratch notebook

Do not delete 00_scratch. It is where every verification cell of the day goes. 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, nb_utils loads, 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

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.

How to work through this lab

Run each cell as soon as you have pasted it: click the play button on the left of the cell, or press Shift+Enter. Later cells depend on earlier ones having run. The cells are numbered below so later labs can refer to them.

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. Then open the Connect dropdown in the notebook ribbon and attach to your existing high concurrency session (it attaches in about two seconds; a new session would take longer and count against the shared capacity).

Step 2: Cell 1, shared utilities

New notebooks open with a first cell full of comment lines. Delete them: this cell must contain only the line below.

%run nb_utils

Step 3: Cell 2, parameters

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

This is an ordinary cell. Do not toggle it as a parameter cell: that setting only matters for notebooks a pipeline calls with parameters, and this notebook is always run by hand.

Step 4: Cell 3, the file-tracking table

# 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: Cell 4, map file names to entity names

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)

This cell only defines a function, so it prints nothing. Prove it works with Cell 5:

print(entity_for("concessions_export_2026-08-31.csv"))   # concessions
print(entity_for("events.csv"))                          # events

Step 6: Cell 6, the loader

Exercise

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()}

loaded_files, loaded_rows = 0, 0
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)
            loaded_files += 1
            loaded_rows += n
            print(f"  {f.name:45s} -> {target:22s} {n:>7,} rows")
        except Exception as e:
            log(target, "failure", 0, f"{f.name}: {e}")
            raise

if loaded_files:
    print(f"Loaded {loaded_files} files, {loaded_rows:,} rows into Bronze")
else:
    print("Loaded 0 files (all already ingested)")

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.

An INFO:trident_token_library_wrapper line may appear above the output: that is Fabric fetching a token to read OneLake, and it is normal.

Step 7: Cell 7, checkpoint

expected = {"bronze.events": 12, "bronze.venues": 8, "bronze.sessions": 122,
            "bronze.products": 48, "bronze.concessions": 6200,
            "bronze.checkins": 2500}
for t, n in expected.items():
    got = spark.table(t).count()
    print(f"  {t:22s} {got:>7,}")
    assert got == n, f"{t}: expected {n:,}, got {got:,}"
print("✅  Checkpoint 1 green: six Bronze tables loaded, provenance stamped")

If the check mark line prints, your Bronze file loader works.

Step 8: Test re-runnability

Run Cell 6 again. It ends with Loaded 0 files (all already ingested). Then run Cell 7 again: identical counts. This is your file tracking at work.

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.

Stuck?

Stuck? Your workspace has a Solutions folder. Open SOLUTION_01_bronze_files, click Run all, and continue with the next lab. Every solution notebook is safe to run from any state.

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, and the instructor builds them on screen with you.

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). Then open the Connect dropdown in the notebook ribbon and attach to your existing high concurrency session (it attaches in about two seconds; a new session would take longer and count against the shared capacity).
2. Cell 1, shared utilities: (only this line in the cell; delete the default comment lines first)

%run nb_utils

3. Cell 2, 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, the parameter cell. Click the ... on the cell and select Toggle parameter cell. It must be marked as the parameter cell, or the pipeline cannot pass a value in:

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.

Morning break. Steps 3 to 8 continue after the break, with the instructor building the pipelines on screen.

Step 3: Build the child pipeline

1. + New item > Data pipeline, name it pl_ingest_child.
2. From the toolbar, add a Copy data activity (not "Copy job": they are different items). Name it Copy table. The Parameters tab only appears once the canvas has an activity, which is why the activity comes first.
3. Click an empty spot on the 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

4. Select Copy table again. Source tab:

• Connection: click Browse > Azure SQL Database. Fill in:
• Server: mosaic-sql-fabconeu.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

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.

Lookup the control table

2. Add a Lookup activity. Name it Read control table.
3. Click an empty spot on the canvas, open Parameters, and add one: ceiling, type String, default value 2026-09-01 00:00:00.
4. Select Read control table again. 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 and name it Invoke child pipeline.
8. Settings tab:

• Type: keep Fabric selected
• Connection: click Browse all > select Fabric Data Pipelines > Connect. On the connection screen, keep the name FabricDataPipelines and authentication Organizational account. If it shows a Sign in button instead of "You are currently signed in", sign in with your workshop account and the password you set this morning. This connection is created once and reused in Lab 7.
• 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, ceiling appears automatically: that is the payoff of toggling the parameter cell in Step 2. Set its value with Add dynamic content: @pipeline().parameters.ceiling.
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 Invoke child pipeline runs, each child runs a Copy, and then Advance watermarks runs once.
3. Wait for all activities to report Succeeded (a few minutes).

Step 6: The checkpoint

In 00_scratch, add a new cell. (If Cell 1 of 00_scratch was run more than an hour ago, run it again first: NameError: name 'F' is not defined means the Spark session expired.)

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()
    note = "   (source has 242: one refund is newer than the ceiling)" if t.endswith("refunds") else ""
    print(f"  {t:26s} {got:>7,}{note}")
    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.

Trouble?

Trouble? Counts doubled (bronze.attendees: expected 4,000, got 8,000)? The master pipeline ran twice before the watermarks advanced, almost always because the first run failed at Advance watermarks (parameter cell not toggled, notebook not saved, or the activity not wired). Bronze is append-only, so the second run added everything again. Fix the master first, then recover:

# Drop the doubled Bronze tables and reset the watermarks, then run pl_ingest_master ONCE
for t in ["attendees", "ticket_orders", "ticket_order_lines", "payments", "refunds"]:
    spark.sql(f"DROP TABLE IF EXISTS bronze.{t}")
spark.sql("UPDATE meta_ingestion_control SET last_watermark = '1900-01-01 00:00:00'")
print("bronze SQL tables dropped, watermarks reset")

Step 7: Prove idempotency

Run pl_ingest_master again with the same ceiling. When it finishes, prove it copied nothing, using the provenance column your child pipeline stamps. In 00_scratch, add a new cell. (If Cell 1 of 00_scratch was run more than an hour ago, run it again first: NameError: name 'F' is not defined means the Spark session expired.)

# Rows per load window, per table. A second run of the same window adds nothing.
for t in ["attendees", "ticket_orders", "ticket_order_lines", "payments", "refunds"]:
    counts = (spark.table(f"bronze.{t}")
              .groupBy("_batch_id").count().orderBy("_batch_id").collect())
    print(f"  {t:22s} " + ", ".join(f"{r._batch_id}: {r['count']:,}" for r in counts))

Every table shows exactly one batch, sql<=2026-09-01 00:00:00, with the same count as before. Re-run the checkpoint from Step 6: identical counts. The window is empty because the watermark equals the ceiling. Your framework is re-runnable by design.

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.

Stuck?

Stuck? Your workspace has a Solutions folder. SOLUTION_02_setup_control builds the control table and SOLUTION_nb_update_watermarks is the watermark notebook; the pipelines are built along with the instructor. Every solution notebook is safe to run from any state.

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

Run each cell as soon as you have pasted it (play button or Shift+Enter).

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 (Add data items > From OneLake catalog > Mosaic). Then open the Connect dropdown in the notebook ribbon and attach to your existing high concurrency session (it attaches in about two seconds; a new session would take longer and count against the shared capacity).
3. Cell 1, shared utilities: (only this line in the cell; delete the default comment lines first)

%run nb_utils

Step 2: Page through the API

Cell 2, the API version:

API_VERSION = "v1"

Cell 3, 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}")
    print(f"  {entity}: {len(rows)} rows from {API_VERSION}")

Step 3: Cell 4, the assertion

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

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

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 Cell 2 to:

API_VERSION = "v2"

Re-run Cell 2, then Cell 3, then Cell 4. The checkpoint 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 90_release_batch2 from your workspace. The Mosaic lakehouse is already attached. Open the Connect dropdown, attach to your high concurrency session, then click Run all. 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: the 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. Some time has passed since you ran it, and the Spark session it used has expired, so everything it defined (entity_for, LANDING, even F) is gone. The notebook must run from the top.

1. In Cell 2 (parameters), change the batch list to:

BATCH_DIRS = ["batch1", "batch2"]

2. Click Run all.

Run all ends red. The loader (Cell 6) reports three new files and 14,300 new rows, and then the checkpoint (Cell 7) fails:

bronze.concessions: expected 6,200, got 18,200

An assertion you wrote this morning caught a problem this afternoon. Keep it in mind while you read on. (Checkins grew too, from 2,500 to 5,000: that is the legitimate day-2 file, and the checkpoint never reaches it because it stops at the first failure.)

Step 8: Look at the numbers

Day 1 had 6,200 concession sales, day 2 has 5,800. That is 12,000. Where did the extra 6,200 come from? In 00_scratch, add a new cell. (If Cell 1 of 00_scratch was run more than an hour ago, run it again first: NameError: name 'F' is not defined means the Spark session expired.)

(spark.table("bronze.concessions")
 .groupBy("_source_file").count().orderBy("_source_file")
 .show(truncate=False))

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. Leave the Lab 1 checkpoint red for now; Lab 4 makes the number right at the layer where it belongs.

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 to 3

Stuck?

Stuck? Your workspace has a Solutions folder. Open SOLUTION_03_bronze_api, click Run all, and continue with the next lab. Every solution notebook is safe to run from any state.

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

Run each cell as soon as you have pasted it (play button or Shift+Enter). If a cell that worked before lunch now throws NameError, your Spark session expired over the break: click Run all on that notebook and continue.

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. Then open the Connect dropdown in the notebook ribbon and attach to your existing high concurrency session (it attaches in about two seconds; a new session would take longer and count against the shared capacity).

Step 2: Cell 1, shared utilities

Only this line in the cell; delete the default comment lines first.

%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: Cell 3, the reusable patterns

Three patterns power all of Silver:

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: Cell 4, Silver concessions

This is where the SharePoint duplicate dies.

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())

Cell 5, verify the dedup:

n = spark.table("silver.concessions").count()
print(f"  silver.concessions {n:>7,}   (Bronze holds 18,200)")
assert n == 12_000, f"expected 12,000, got {n:,}"
print("✅  Concessions deduplicated: 18,200 in Bronze, 12,000 in Silver")

Step 5: Cell 6, Silver relational and reference tables

Now build Silver for the SQL-sourced entities and for the three reference entities from the files (events, venues, products), which the Gold dimensions read in Lab 5:

INT_COLS = {"order_total", "quantity", "unit_price", "line_amount", "amount", "capacity"}
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",
            # reference entities from the CSV files, used by the Gold dimensions
            "events": "event_id", "venues": "venue_id", "products": "product_id"}

# Track which order keys changed in this batch (you'll need this below)
changed_order_keys = None
# Silver state is advanced only after the aggregate succeeds (Cell 7)
pending_marks = {}

for entity, key in ENTITIES.items():
    changed = changed_bronze(entity)
    if changed.count() == 0:
        print(f"  {entity:22s} nothing new")
        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)
    pending_marks[entity] = max_ts          # marked processed at the end of Cell 7
    log(f"silver.{entity}", "success", src.count())
    print(f"  {entity:22s} {src.count():>7,} rows merged")

    # 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: Cell 7, the order financials aggregate

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

# First build (or rebuild after the table was dropped): compute every order
if not spark.catalog.tableExists("silver.order_financials"):
    changed_order_keys = spark.table("silver.ticket_orders").select("order_id").distinct()

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())
    print(f"  order_financials recomputed for {fin.count():,} orders")
else:
    print("  order_financials: no changed orders, nothing to recompute")

# Only now advance the Silver state for everything Cell 6 merged
for entity, ts in pending_marks.items():
    mark_processed(entity, ts)
pending_marks = {}

Trouble?

Trouble? Cell 7 printed "no changed orders" and silver.order_financials does not exist, typically after a session expired between Cell 6 and Cell 7 on an older version of this notebook: the Silver state for orders and lines was advanced, so the change set came back empty. Reset the two markers and rerun from Cell 6:

spark.sql("DELETE FROM _silver_state WHERE entity IN ('ticket_orders', 'ticket_order_lines')")
print("orders and lines will be reprocessed on the next run of Cell 6")

Step 7: Cell 8, the reconciliation checkpoint

TRUSTED = 2_614_339
revenue = (spark.table("silver.order_financials")
           .agg(F.sum("net_amount")).first()[0]
           + spark.table("silver.concessions")
           .agg(F.sum("sale_amount")).first()[0])
print(f"  reconciled revenue   EUR {revenue:,}")
assert revenue == TRUSTED, f"expected {TRUSTED:,}, got {revenue:,}"
print("✅  Checkpoint 3 green: Silver reconciles")

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 next day's data 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, see what the load brought in. In 00_scratch, add a new cell. (If Cell 1 of 00_scratch was run more than an hour ago, run it again first: NameError: name 'F' is not defined means the Spark session expired.)

for t in ["attendees", "ticket_orders", "ticket_order_lines", "payments", "refunds"]:
    n = (spark.table(f"bronze.{t}")
         .filter(F.col("_batch_id") == "sql<=2026-09-02 00:00:00").count())
    print(f"  {t:22s} {n} row(s) in load 2")

Four zeros, and refunds 1 row(s). 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 these three cells, in order:

• Cell 6, the relational entities loop (starts with INT_COLS = {...})
• Cell 7, the order financials aggregate (starts with if changed_order_keys is not None)
• Cell 8, the checkpoint (starts with TRUSTED = 2_614_339)

Cell 6 reports refunds 1 rows merged and nothing new for the other seven. Cell 7 prints that there were no changed orders. Cell 8 is green, at the same number as before.

Step 10: Assert the number that should be true

All 242 refunds are now in silver.refunds. REF-2001 cancelled ORD-1042, a EUR 399 festival pass, so the reconciled revenue should now be EUR 2,613,940. Update the constant in Cell 8 to the number that should be true, and run it:

TRUSTED = 2_613_940

It fails:

expected 2,613,940, got 2,614,339

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

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: Cell 9, 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 directly after Cell 7, the order financials aggregate (the one starting if changed_order_keys is not None). This becomes Cell 9, and the checkpoint moves down to Cell 10:

# Find orders where the aggregate is stale
# (refund total in silver.refunds does not match what is 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())

n_stale = stale_keys.count()
print(f"  stale orders found: {n_stale}")

if n_stale > 0:
    # Recompute financials for stale orders only
    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(),
        "late refund backfill: union-of-changed-keys rule")

Step 12: The trusted number

Run the checkpoint (now Cell 10) again:

  reconciled revenue   EUR 2,613,940
✅  Checkpoint 3 green: Silver reconciles

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

Step 13: Make the trusted number your standing assertion

Your checkpoint cell already asserts TRUSTED = 2_613_940. Keep it that way. 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. Click Run all once now to prove the notebook is green from the top.

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.

Stuck?

Stuck? Your workspace has a Solutions folder. Open SOLUTION_04_silver, click Run all, and continue with the next lab. Every solution notebook is safe to run from any state.

Quick reference

Lab 5 - Gold star schema

Time: 20 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

Run each cell as soon as you have pasted it (play button or Shift+Enter). The dimension exercises have collapsible solutions; using one is time management, not cheating.

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. Then open the Connect dropdown in the notebook ribbon and attach to your existing high concurrency session (it attaches in about two seconds; a new session would take longer and count against the shared capacity).

Step 2: Cell 1, shared utilities

Only this line in the cell; delete the default comment lines first.

%run nb_utils

Step 3: Dimensions

Cell 2, the event dimension, denormalized with venue attributes (no snowflaking):

events = spark.table("silver.events").select(
    "event_id", "event_name", "venue_id", "start_date", "end_date", "event_type")
venues = spark.table("silver.venues").select(
    "venue_id", "venue_name", "city", "country_code", "capacity")

(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")

Cell 3, 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")

Cell 4, your task: gold.dim_product from silver.products, columns product_id, event_id, product_name, unit_price

(spark.table("silver.products")
 .select("product_id", "event_id", "product_name", "unit_price")
 .write.mode("overwrite").saveAsTable("gold.dim_product"))
print(f"  gold.dim_product: {spark.table('gold.dim_product').count()} rows")

Cell 5, the date dimension. A conformed calendar, one row per day from 2024 through 2027, with the columns a semantic model expects:

# Conformed date dimension: one row per calendar day, 2024-01-01 to 2027-12-31.
# date_key is yyyyMMdd as an integer, matching the facts.
dim_date = (spark.sql("SELECT explode(sequence(to_date('2024-01-01'), to_date('2027-12-31'), interval 1 day)) AS date")
    .withColumn("date_key",        F.date_format("date", "yyyyMMdd").cast("int"))
    .withColumn("year",            F.year("date"))
    .withColumn("quarter",         F.quarter("date"))
    .withColumn("quarter_name",    F.concat(F.lit("Q"), F.quarter("date"), F.lit(" "), F.year("date")))
    .withColumn("month",           F.month("date"))
    .withColumn("month_name",      F.date_format("date", "MMMM"))
    .withColumn("month_short",     F.date_format("date", "MMM"))
    .withColumn("year_month",      F.date_format("date", "yyyy-MM"))
    .withColumn("year_month_key",  F.date_format("date", "yyyyMM").cast("int"))
    .withColumn("week_of_year",    F.weekofyear("date"))
    .withColumn("day",             F.dayofmonth("date"))
    .withColumn("day_of_year",     F.dayofyear("date"))
    .withColumn("day_of_week",     F.dayofweek("date"))            # 1 = Sunday ... 7 = Saturday
    .withColumn("day_name",        F.date_format("date", "EEEE"))
    .withColumn("day_short",       F.date_format("date", "EEE"))
    .withColumn("is_weekend",      F.dayofweek("date").isin(1, 7))
    .withColumn("is_month_start",  F.dayofmonth("date") == 1)
    .withColumn("is_month_end",    F.col("date") == F.last_day("date"))
    .withColumn("fiscal_year",     F.when(F.month("date") >= 10, F.year("date") + 1).otherwise(F.year("date")))   # FY starts 1 October
    .withColumn("fiscal_quarter",  F.floor(((F.month("date") + 2) % 12) / 3) + 1)
    .select("date_key", "date", "year", "quarter", "quarter_name", "month", "month_name", "month_short",
            "year_month", "year_month_key", "week_of_year", "day", "day_of_year", "day_of_week",
            "day_name", "day_short", "is_weekend", "is_month_start", "is_month_end",
            "fiscal_year", "fiscal_quarter"))

dim_date.write.mode("overwrite").saveAsTable("gold.dim_date")
n = spark.table("gold.dim_date").count()
assert n == 1461, f"expected 1,461 days (2024 to 2027 incl. leap day), got {n:,}"
print(f"  gold.dim_date: {n:,} rows")

Step 4: Facts

Production note

Why write the facts in ten loads? At workshop scale, each fact would land as one or two tidy files, and Lab 6 would have nothing to compact. In production, fact tables fragment because they are written in many small commits: daily loads, micro-batches, retries. The helper below simulates exactly that, ten separate appends per fact, so you can see OPTIMIZE do real work in Lab 6. You would never write a table this way on purpose in production.

Three facts, each at its natural grain. First, Cell 6, the small helper that writes a DataFrame as ten separate loads:

def write_in_loads(df, target, key, loads=10):
    """Workshop only: write df as `loads` separate commits to simulate daily appends."""
    for i in range(loads):
        part = df.filter(F.abs(F.hash(key)) % loads == i)
        part.write.mode("overwrite" if i == 0 else "append").saveAsTable(target)

Cell 7, 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"))

fact = (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"))
write_in_loads(fact, "gold.fact_ticket_sales", "order_line_id")
print(f"  gold.fact_ticket_sales: {spark.table('gold.fact_ticket_sales').count():,} rows")

Cell 8, gold.fact_concessions, grain: one row per POS sale

fact = (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"))
write_in_loads(fact, "gold.fact_concessions", "sale_id")
print(f"  gold.fact_concessions: {spark.table('gold.fact_concessions').count():,} rows")

Cell 9, your task: gold.fact_refunds, grain: one row per refund. 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, and write it with write_in_loads like the other facts (key: refund_id).

fact = (spark.table("silver.refunds").alias("r")
        .join(orders.alias("o"), "order_id")
        .select("refund_id", "order_id", "attendee_id", "event_id", "reason",
                F.date_format(F.to_date("r.created_ts"), "yyyyMMdd")
                 .cast("int").alias("date_key"),
                "amount"))
write_in_loads(fact, "gold.fact_refunds", "refund_id")
print(f"  gold.fact_refunds: {spark.table('gold.fact_refunds').count():,} rows")

Production note

Design note: net revenue is a measure definition (ticket sales + concessions - refunds), not a stored column. Each fact keeps its own clean grain, and the subtraction happens in the semantic model, where the definition is visible and versioned instead of forked across reports.

Step 5: Cell 10, final checkpoint

TRUSTED = 2_613_940
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])
print(f"  net revenue at Gold   EUR {net:,}")
assert net == TRUSTED, f"expected {TRUSTED:,}, got {net:,}"
print("✅  Checkpoint 4 green: the trusted number holds at the consumption layer")

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. 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 in the inherited build you will hear about at the end of the day.

Stuck?

Stuck? Your workspace has a Solutions folder. Open SOLUTION_05_gold, click Run all, and continue with the next lab. Every solution notebook is safe to run from any state.

Quick reference

Lab 6 - Optimize for Direct Lake

Time: 10 minutes, mostly runtime

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

Run each cell as soon as you have pasted it (play button or Shift+Enter).

The story

Your Gold tables are correct, but they were written as 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). Then open the Connect dropdown in the notebook ribbon and attach to your existing high concurrency session (it attaches in about two seconds; a new session would take longer and count against the shared capacity).
3. Cell 1, shared utilities: (only this line in the cell; delete the default comment lines first)

%run nb_utils

Step 2: Cell 2, measure the current state

Note

About the file counts below: the fact tables were intentionally written in ten separate loads in Lab 5 to simulate what happens in production over days of appends. Dimension tables are one file each, which is normal for small reference data.

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:26s} {d.numFiles:>3} files  {d.sizeInBytes:>10,} bytes")

Step 3: Cell 3, 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.

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

Step 4: Cell 4, 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.

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

Step 5: Cell 5, measure the improvement

print("  before -> after")
for t in GOLD_TABLES:
    d = spark.sql(f"DESCRIBE DETAIL {t}").first()
    b_files, b_bytes = before[t]
    print(f"  {t:26s} {b_files:>3} -> {d.numFiles:>3} files  "
          f"{b_bytes:>10,} -> {d.sizeInBytes:>10,} bytes")

assert all(
    spark.sql(f"DESCRIBE DETAIL {t}").first().numFiles <= before[t][0]
    for t in GOLD_TABLES)
print("✅  Checkpoint 5 green: Gold compacted and V-Ordered, 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 run after every significant write to Gold, typically at the end of the nightly pipeline. Lab 7 wires exactly that.

Stuck?

Stuck? Your workspace has a Solutions folder. Open SOLUTION_06_optimize, click Run all, and continue with the next lab. Every solution notebook is safe to run from any state.

Quick reference

Lab 7 - End-to-end orchestration

Time: 35 minutes. The two full runs take about ten minutes each, so the lab is built so that you never wait for one: while a run executes, the instructor walks the Monitoring Hub and the failure path on screen. The break-it exercise is an instructor demo; your workspace stays yours after the workshop, so you can run it yourself later.

In this lab you will: wire ingestion, Silver, Gold, and maintenance into one work pipeline, wrap it in an orchestrator that logs success and failure, run it end to end, and prove it is safe to re-run.

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

Confirm your 04_silver checkpoint asserts TRUSTED = 2_613_940 and the notebook is green top to bottom (Lab 4, Step 13). The orchestrator runs that notebook unattended; an assertion pinned to the interim number fails the whole run at the Silver step.

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. Cell 1, shared utilities: (only this line in the cell; delete the default comment lines first)

%run nb_utils

3. Cell 2, the parameter cell. Click the cell's ... menu and select Toggle parameter cell:

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

4. Cell 3, write the log row:

(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 the four processing steps and knows nothing about logging.

1. + New item > Data pipeline, name it pl_mosaic_work.
2. Add the four activities below first, each connected with the On success (green) connector from the previous one. Leave the ceiling parameter of the Ingest activity for a moment.
3. Click an empty spot on the canvas, open Parameters, add ceiling (String, default 2026-09-02 00:00:00), then go back to Ingest and set its ceiling parameter to @pipeline().parameters.ceiling. (The Parameters tab only appears once the canvas has an activity.)

The four activities:

• Invoke pipeline, name it Ingest. Connection: FabricDataPipelines (the one you created in Lab 2). Pipeline: pl_ingest_master. Wait on completion: ON. Parameter ceiling = @pipeline().parameters.ceiling (Add dynamic content).
• Notebook, name it Silver. Notebook: 04_silver.
• Notebook, name it Gold. Notebook: 05_gold.
• Notebook, name it Maintenance. Notebook: 06_optimize. Maintenance is the last step of every load: part of the pipeline, not a quarterly cleanup job.

4. On each of the three Notebook activities, open Settings > Advanced and set Session tag to mosaic (the same value on all three). Activities with the same tag share one Spark session instead of each starting their own, which takes a few minutes off every run.
5. Save.

Step 3: Build the orchestrator

The orchestrator's only job is to invoke the work pipeline and log the outcome. One activity that can succeed or fail, so exactly two edges to wire.

1. + New item > Data pipeline, name it pl_run_all.
2. Add an Invoke pipeline activity, name it Invoke mosaic work. Connection: FabricDataPipelines. Pipeline: pl_mosaic_work. Wait on completion: ON.
3. Click an empty spot on the canvas, open Parameters, add ceiling (String, default 2026-09-02 00:00:00). Then select Invoke mosaic work again and set its ceiling parameter to @pipeline().parameters.ceiling.
4. Add a Notebook activity, name it Log success. Drag the green (On success) connector from Invoke mosaic work to it. Notebook: nb_log_run. The four base parameters appear automatically (that is the parameter cell from Step 1 at work). Set them with Add dynamic content:

• run_id = @pipeline().RunId
• pipeline_name = @pipeline().PipelineName
• run_status = success
• message = leave empty

5. Add a second Notebook activity, name it Log failure. Drag the red (On failure) connector from Invoke mosaic work to it. Same notebook, same parameters, except:

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

6. Save.

Production note

Production note: in a real deployment an alert follows 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.

Step 4: The end-to-end run

1. In pl_run_all, click Run. Keep the ceiling at 2026-09-02 00:00:00.
2. The run takes around ten minutes. While it runs, follow the instructor through the Monitoring Hub (left navigation > Monitor): filter to your workspace and watch pl_run_all invoke pl_mosaic_work, which invokes pl_ingest_master, which fans out into five pl_ingest_child runs, then Silver, Gold, and Maintenance execute in sequence. Three levels deep, one click each.
3. When it finishes, verify in 00_scratch, in a new cell. (If Cell 1 of 00_scratch was run more than an hour ago, run it again first: NameError: name 'F' is not defined means the Spark session expired.)

# The trusted number survived a full end-to-end re-run
TRUSTED = 2_613_940
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 == TRUSTED, f"expected {TRUSTED:,}, got {net:,}"
print(f"✅  net revenue after the orchestrated run: EUR {net:,}")

# And the run logged its own history
(spark.table("_pipeline_log")
 .select("logged_at", "run_id", "item", "status", "rows", "message")
 .orderBy(F.col("logged_at").desc())
 .show(8, truncate=False))

You should see the check mark line, then a log like this (your IDs and times differ):

logged_at            run_id        item                      status   rows  message
2026-09-14 07:31:00  863548e4-...  pl_run_all                success  0
2026-09-14 07:30:12  ae0b6ff2-...  silver.order_financials   success  0     late refund backfill
2026-09-14 07:29:40  ae0b6ff2-...  silver.refunds            success  0
...

How to read it: the top row is the orchestrator logging itself, with the pipeline's run ID. Below it are the notebooks the work pipeline ran, each carrying its own run ID, all within the same minute. Rows with 0 are incremental steps that found nothing new: the watermark equals the ceiling, so the ingest copied nothing and Silver merged nothing.

Step 5: Prove it is safe to re-run

Run pl_run_all again with the same ceiling, then move on: this run finishes during the debrief. The last thing you do today is come back to 00_scratch, re-run the cell above, and see a second success row for pl_run_all, zero rows everywhere, the same number.

Incremental + idempotent = re-runnable at 3 a.m. without fear. That sentence is the payoff of every MERGE and watermark you wrote today.

Step 6: Break it on purpose (instructor demo)

Trust an error path you've tested; distrust one you haven't. The instructor runs this on screen; the steps are here so you can do it yourself after the workshop.

1. In 00_scratch, corrupt one control-table row (run Cell 1 again first if the session expired):

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

2. Run pl_run_all. The payments child fails on an invalid column name, the ForEach fails, pl_ingest_master fails, the Ingest activity fails, pl_mosaic_work fails, Invoke mosaic work fails, and Log failure fires.
3. Read the failure from the log, root cause first. In 00_scratch:

# Failed runs, compact: the root cause is the innermost "Message=" in the wrapped error
failed = (spark.table("_pipeline_log").filter("status = 'failure'")
          .withColumn("root_cause", F.regexp_extract("message", r"Message=([^,]+)", 1))
          .select("logged_at", "item", "root_cause")
          .orderBy(F.col("logged_at").desc()))
failed.show(truncate=False)

row = failed.first()
print(row.root_cause if row else "no failures logged")

You should see one row for pl_run_all with root cause Invalid column name 'NoSuchColumn'. The full message in the log is the entire chain, every pipeline level wrapping the one below it; the root cause is what an alert should carry.

4. Repair and prove recovery:

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

Run pl_run_all once more. It succeeds, and the log now holds a failure row followed by a success row: your incident report, with timestamps, no chat archaeology required.

Step 7: Schedule it

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.

Stuck?

Stuck? Your workspace has a Solutions folder. SOLUTION_nb_log_run is the logging notebook; the pipelines are built along with the instructor. Every solution notebook is safe to run from any state.

Quick reference

Lab 8 - Disaster forensics

Time: the closing debrief, presented from the front

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

This is a read-and-discuss lab. There is nothing to run: the instructor walks the inherited build on screen, and your job is to name what is wrong before it is named for you.

The story

Before you arrived, another team built the Mosaic platform. Finance stopped trusting its numbers months ago. Nobody can explain why. What follows is what that build looks like from the inside.

Your job is to diagnose, not fix. Each item is a production anti-pattern that one of today's labs addressed. Keep score of how many you can name from the evidence alone.

The investigation

For each question, the instructor shows the evidence. Decide which production principle it violates and which lab fixed it.

Question 1: Provenance

Any Bronze table in the inherited build. Can you tell which file or extract a 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

The inherited bronze.ticket_order_lines has twice the row count of the source database (16,032 against 8,016). What does that tell you about how the loader was operated?

Question 3: The file that shouldn't be there

The landing folder listing, shown on screen. Look at the file names carefully. Does anything look familiar from this morning?

Question 4: The revenue report's grain bug

The inherited revenue report notebook, first code cell, shown on screen.

• At what grain does it join payments to order lines?
• 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 in the revenue report: 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 inherited build for the word "refund". What do you find?

Note

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

Question 7: The TODO comment

The revenue report carries a TODO: "numbers look a bit high since March?" Given everything you've found, is "a bit high" the right diagnosis?

Scoring

And the eighth smell you will hear about without a query: tables written as thousands of tiny files, never compacted, which is what Lab 6 prevents.

The debrief

Here is the punchline: mentally fix the inherited build 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. A build with correct-looking mechanics that still double-counts one renamed file and ignores refunds.

The inherited build 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. Run each cell as soon as you have pasted it.

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: The rules table

1. + New item > Notebook, name it 09_data_quality. Attach the Mosaic lakehouse. Then open the Connect dropdown in the notebook ribbon and attach to your existing high concurrency session (it attaches in about two seconds; a new session would take longer and count against the shared capacity).
2. Cell 1, shared utilities: (only this line in the cell; delete the default comment lines first)

%run nb_utils

3. Cell 2, 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: Cell 3, the quality engine

The engine reads every rule, evaluates it against the target table, and returns pass or 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: Cell 4, 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")

Production note

Production note: every result is persisted in _quality_log with its run ID, rule, target, severity, and timestamp, which makes the log a fact table in its own right. Point a Direct Lake semantic model at it and you have a data quality dashboard for free: pass/fail trend per rule over time, failures by severity per table, and the last green run per entity as a freshness signal. It is modelled exactly like the facts you built in Lab 5.

Step 4: Cell 5, 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: Cell 6, 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.

Stuck?

Stuck? Your workspace has a Solutions folder. Open SOLUTION_09_data_quality, click Run all, and continue. Every solution notebook is safe to run from any state.

Quick reference