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.

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

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

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

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.

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.

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.

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.

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.

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

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.

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

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.

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.

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

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, \
    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.

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

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

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.

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. 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

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.

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.

Step 5: Break it on purpose

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

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.

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.

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