Automating Fabric Lakehouse Metadata Refresh with Semantic Link: Syncing Delta Table Schema Changes to a Direct Lake Semantic Model Using Python and the Fabric REST API

Introduction

Here's a scenario that plays out quietly in data teams everywhere: a data engineer adds three new columns to a gold-layer Delta table — say, customer_lifetime_value, churn_risk_score, and last_interaction_channel — runs the Spark notebook, sees the table updated, and marks the ticket done. Meanwhile, the semantic model sitting on top of that table in Direct Lake mode has no idea any of this happened. The Power BI reports built on that model don't show the new columns. The business analyst asks why. The engineer says "the data is there." The analyst says "not in Power BI it isn't." And then someone manually clicks Refresh in the semantic model settings and everyone moves on — until it happens again next sprint.

This isn't a rare edge case. It's a structural gap between how Delta tables evolve and how Direct Lake semantic models track that evolution. Delta tables can grow, shed, and reshape their schemas dynamically. A Direct Lake semantic model, by contrast, builds its column metadata at a point in time and doesn't automatically re-survey the underlying Parquet files when the schema changes. The fix — triggering a metadata refresh — is simple. The problem is making it automatic, reliable, and woven into your existing data orchestration without adding manual steps.

By the end of this lesson, you'll know exactly how to close that gap. You'll build a Python-based automation that detects Delta table schema changes, calls the Fabric REST API to trigger a metadata-only refresh on a Direct Lake semantic model, and does it all from inside a Fabric notebook — making it a composable step you can drop into any pipeline. Here's what you'll walk away with:

What you'll learn:

• How Direct Lake framing works and why schema changes don't propagate automatically
• How to use the semantic-link Python library (sempy) to inspect and interact with Fabric semantic models programmatically
• How to call the Fabric REST API to trigger a metadata refresh (distinct from a full data refresh) on a Direct Lake model
• How to detect schema drift between a Delta table and a published semantic model using PySpark and sempy
• How to embed this automation into a Fabric data pipeline as a reliable post-transformation step

---

Prerequisites

Before diving in, you should be comfortable with:

• Working with Delta tables in a Fabric Lakehouse, including schema evolution patterns (if you need a refresher, see Handling Schema Evolution in Fabric Lakehouse Delta Tables)
• Running PySpark notebooks in Microsoft Fabric
• The concept of Direct Lake semantic models and how they read directly from OneLake Delta files
• Basic familiarity with the Fabric workspace structure and REST APIs
• Python comfort level: you should be able to read and adapt code involving classes, loops, and HTTP requests

You'll need a Fabric capacity (F2 or higher, or a trial), a Lakehouse with at least one Delta table, and a Direct Lake semantic model attached to that Lakehouse. If you're setting up the semantic model for the first time, the article on connecting a Power BI semantic model to a Fabric Lakehouse in Direct Lake mode covers the initial wiring.

---

Understanding Why Schema Changes Break the Sync

To automate the fix, you need to understand the mechanism that's failing. Let's be precise about what "Direct Lake" actually does under the hood.

When a Direct Lake semantic model is created or refreshed, the Analysis Services engine embedded in Fabric reads the Delta table's transaction log to establish the frame — the set of Parquet files and column definitions that constitute the current "snapshot" of the table. This frame is stored inside the semantic model as metadata: column names, data types, cardinalities, and pointers to specific file versions in OneLake.

When Power BI queries the model, it reads directly from those Parquet files using the frame it already knows about. There's no live schema discovery on each query — that would be prohibitively expensive. The framing was designed for data to stay fresh (because the files are read directly), not necessarily for schema to stay current.

Here's the subtle part: if you add a new column to a Delta table in Spark, the new column appears in subsequent Parquet files written by that operation. But the semantic model's existing frame was established before those files existed. The model doesn't know the column exists, so it doesn't expose it. The data is in OneLake; the model just can't see it.

A metadata refresh (sometimes called a schema sync or framing operation) tells the semantic model to re-read the Delta log and update its internal frame to reflect the current schema. This is different from a full import-style data refresh — no data moves into an in-memory cache. The operation just updates the structural metadata and re-establishes which files belong to the current snapshot.

This distinction matters enormously for automation design. You don't need to trigger a full refresh every time data changes — that would defeat the purpose of Direct Lake. You only need to trigger a metadata refresh when the shape of your Delta tables changes.

---

The Semantic Link Library: Your Programmatic Bridge

Microsoft ships a Python library called semantic-link (importable as sempy) specifically for interacting with Power BI and Fabric semantic models from within notebooks. It's pre-installed in Fabric Spark environments, so you can start using it immediately without any pip installs in most cases. For the REST API calls, you'll supplement it with the requests library and the notebookutils credential helpers.

Install or verify the semantic-link version in your notebook:

# Verify semantic-link is available and check version
import sempy
print(sempy.__version__)

# If you need a specific version or it's missing:
# %pip install semantic-link --quiet

The sempy library exposes two main namespaces relevant to our task:

• sempy.fabric — Functions for interacting with Fabric workspace metadata: listing semantic models, datasets, lakehouses, and triggering operations
• sempy.relationships — Tools for analyzing model relationships (not our focus here)

Let's start by listing the semantic models available in your current workspace:

import sempy.fabric as fabric

# List all semantic models in the current workspace
models = fabric.list_datasets()
print(models[['Dataset Name', 'Dataset Id', 'Configured By']])

This returns a pandas DataFrame with one row per semantic model. Note the Dataset Id for the model you want to automate — you'll use it repeatedly. In practice, you should look up the ID dynamically by name rather than hardcoding it, because IDs change when items are recreated.

# Robust pattern: look up the model ID by name
TARGET_MODEL_NAME = "Sales Analytics - Direct Lake"

models = fabric.list_datasets()
matched = models[models['Dataset Name'] == TARGET_MODEL_NAME]

if matched.empty:
    raise ValueError(f"No semantic model named '{TARGET_MODEL_NAME}' found in this workspace.")

dataset_id = matched.iloc[0]['Dataset Id']
workspace_id = fabric.get_workspace_id()

print(f"Dataset ID: {dataset_id}")
print(f"Workspace ID: {workspace_id}")

Now let's look at how to inspect the model's current schema using sempy:

# Get the columns currently registered in the semantic model
model_columns = fabric.list_columns(dataset=TARGET_MODEL_NAME)
print(model_columns[['Table Name', 'Column Name', 'Data Type', 'Column Type']].head(20))

fabric.list_columns() returns every column across every table in the model, including calculated columns and measures. You'll want to filter to just the physical (non-calculated) columns when comparing against Delta table schemas:

# Filter to physical columns only (exclude calculated columns and row-number columns)
physical_columns = model_columns[
    (model_columns['Column Type'] == 'Data') &
    (~model_columns['Column Name'].str.startswith('RowNumber'))
]

This gives you the ground truth of what the semantic model currently knows about. The next step is comparing that against what the Delta table actually contains.

---

Detecting Schema Drift Between Delta Tables and the Semantic Model

Schema drift detection is the intelligence layer of your automation. Rather than blindly triggering a metadata refresh on every pipeline run (which is wasteful and can cause unnecessary model interruptions), you want to refresh only when schema actually changed.

Here's the approach: read the current Delta table schema using PySpark's DeltaTable API, compare it against the semantic model's registered columns for that table, and flag any discrepancies. Discrepancies include added columns, removed columns, and data type changes.

from pyspark.sql import SparkSession
from delta.tables import DeltaTable
import pandas as pd

spark = SparkSession.builder.getOrCreate()

# Configuration
LAKEHOUSE_NAME = "sales_lakehouse"
TABLE_NAME = "gold_customer_metrics"
SEMANTIC_MODEL_TABLE_NAME = "gold_customer_metrics"  # as it appears in the semantic model

# Step 1: Get the current Delta table schema
delta_table_path = f"Tables/{TABLE_NAME}"
df = spark.read.format("delta").load(delta_table_path)
delta_schema = df.schema

# Build a normalized dictionary of {column_name_lower: spark_type_string}
delta_columns = {
    field.name.lower(): str(field.dataType)
    for field in delta_schema.fields
}

print(f"Delta table has {len(delta_columns)} columns:")
for col_name, col_type in delta_columns.items():
    print(f"  {col_name}: {col_type}")

Now retrieve what the semantic model currently knows about this same table:

# Step 2: Get the semantic model's view of the same table
model_columns = fabric.list_columns(dataset=TARGET_MODEL_NAME)

# Filter to the specific table and physical columns
table_columns = model_columns[
    (model_columns['Table Name'] == SEMANTIC_MODEL_TABLE_NAME) &
    (model_columns['Column Type'] == 'Data')
]

# Build a normalized dictionary of {column_name_lower: power_bi_type_string}
model_column_dict = {
    row['Column Name'].lower(): row['Data Type']
    for _, row in table_columns.iterrows()
}

print(f"Semantic model knows about {len(model_column_dict)} columns for this table:")
for col_name, col_type in model_column_dict.items():
    print(f"  {col_name}: {col_type}")

Now the comparison:

# Step 3: Detect drift
delta_col_names = set(delta_columns.keys())
model_col_names = set(model_column_dict.keys())

added_columns = delta_col_names - model_col_names
removed_columns = model_col_names - delta_col_names

print("\n=== Schema Drift Report ===")
if added_columns:
    print(f"Columns added to Delta table (not in model): {added_columns}")
if removed_columns:
    print(f"Columns removed from Delta table (still in model): {removed_columns}")
if not added_columns and not removed_columns:
    print("No schema drift detected. Schemas are in sync.")

schema_changed = bool(added_columns or removed_columns)

You now have a boolean schema_changed flag that can gate the rest of your automation. This pattern is clean and cheap to run — it does one Delta schema read (which hits the transaction log, not the data files) and one sempy API call.

---

Triggering a Metadata Refresh via the Fabric REST API

When schema_changed is True, you need to tell the semantic model to re-frame itself. The Fabric REST API exposes an endpoint for this, and the call is straightforward — but getting the authentication right inside a notebook requires a bit of care.

Authentication: Getting a Token Inside a Fabric Notebook

Fabric notebooks run under the identity of the user who opened them (or the workspace's service principal if run via pipeline). The notebookutils module provides a clean way to get an OAuth token for the current identity:

# Get an access token for the Power BI/Fabric API
token = notebookutils.credentials.getToken("pbi")
headers = {
    "Authorization": f"Bearer {token}",
    "Content-Type": "application/json"
}

The "pbi" scope gives you a token valid for Power BI REST API calls, which is what the Fabric semantic model endpoints use.

The Refresh API Call

The Fabric REST API endpoint for triggering a dataset refresh is:

POST https://api.fabric.microsoft.com/v1/workspaces/{workspaceId}/datasets/{datasetId}/refreshes

For a metadata-only refresh (updating the schema frame without triggering a full data reload), you pass a specific request body:

import requests
import json

def trigger_metadata_refresh(workspace_id: str, dataset_id: str, headers: dict) -> dict:
    """
    Trigger a metadata refresh on a Direct Lake semantic model.
    Returns the API response as a dict.
    """
    url = f"https://api.fabric.microsoft.com/v1/workspaces/{workspace_id}/datasets/{dataset_id}/refreshes"
    
    # Refresh type: "automatic" lets the engine decide what needs refreshing.
    # For schema sync specifically, we want to force a full metadata re-frame.
    payload = {
        "type": "full",
        "commitMode": "transactional",
        "maxParallelism": 1,
        "retryCount": 0,
        "objects": []  # empty objects array = refresh all tables
    }
    
    response = requests.post(url, headers=headers, json=payload)
    
    if response.status_code == 202:
        print("Metadata refresh triggered successfully (202 Accepted).")
        # The 202 response includes a request ID in the headers
        request_id = response.headers.get("x-ms-request-id", "unknown")
        print(f"Refresh request ID: {request_id}")
        return {"status": "accepted", "request_id": request_id}
    else:
        print(f"Refresh failed. Status: {response.status_code}")
        print(f"Response: {response.text}")
        response.raise_for_status()

# Execute if schema has changed
if schema_changed:
    token = notebookutils.credentials.getToken("pbi")
    headers = {
        "Authorization": f"Bearer {token}",
        "Content-Type": "application/json"
    }
    result = trigger_metadata_refresh(workspace_id, dataset_id, headers)
else:
    print("Schema unchanged. Skipping metadata refresh.")
    result = {"status": "skipped"}

The API returns 202 Accepted immediately — the refresh runs asynchronously. You need to poll for completion if your pipeline has downstream steps that depend on the schema being updated.

Polling for Refresh Completion

import time

def wait_for_refresh_completion(
    workspace_id: str, 
    dataset_id: str, 
    headers: dict, 
    max_wait_seconds: int = 300,
    poll_interval_seconds: int = 10
) -> str:
    """
    Poll the refresh history endpoint until the latest refresh completes.
    Returns the final status string: 'Completed', 'Failed', or 'Cancelled'.
    """
    url = f"https://api.fabric.microsoft.com/v1/workspaces/{workspace_id}/datasets/{dataset_id}/refreshes"
    elapsed = 0
    
    while elapsed < max_wait_seconds:
        time.sleep(poll_interval_seconds)
        elapsed += poll_interval_seconds
        
        # Re-fetch token in case we're in a long poll loop
        fresh_token = notebookutils.credentials.getToken("pbi")
        poll_headers = {
            "Authorization": f"Bearer {fresh_token}",
            "Content-Type": "application/json"
        }
        
        response = requests.get(url, headers=poll_headers)
        response.raise_for_status()
        
        refresh_history = response.json().get("value", [])
        if not refresh_history:
            print(f"  [{elapsed}s] No refresh history yet, continuing to poll...")
            continue
        
        # The most recent refresh is first in the list
        latest = refresh_history[0]
        status = latest.get("status", "Unknown")
        
        print(f"  [{elapsed}s] Refresh status: {status}")
        
        if status in ("Completed", "Failed", "Cancelled", "Unknown"):
            if status == "Failed":
                error = latest.get("serviceExceptionJson", "No error details available")
                raise RuntimeError(f"Semantic model refresh failed: {error}")
            return status
    
    raise TimeoutError(f"Refresh did not complete within {max_wait_seconds} seconds.")

# Use polling after triggering the refresh
if schema_changed:
    final_status = wait_for_refresh_completion(workspace_id, dataset_id, headers)
    print(f"Refresh completed with status: {final_status}")

---

Using the sempy `refresh_dataset` Shortcut

For simpler cases where you don't need fine-grained control over the refresh payload, sempy.fabric offers refresh_dataset() as a higher-level wrapper:

# Simpler alternative using sempy directly
# This triggers a full refresh and blocks until completion
fabric.refresh_dataset(
    dataset=TARGET_MODEL_NAME,
    refresh_type="full"
)
print("Dataset refresh completed via sempy.")

The sempy wrapper handles token acquisition internally and polls for completion, so it's much less code. The trade-off is less control: you can't easily customize the refresh payload, set timeouts, or access the raw response headers. For production automation where you want detailed logging, retry logic, and error handling, the raw REST approach is worth the extra lines.

---

Building the Full Schema Sync Notebook

Now let's assemble the complete, production-quality notebook. This is the artifact you'll commit to your repository and trigger from a Fabric data pipeline. Good pipeline-ready notebooks should accept parameters, emit structured exit values, and handle failures gracefully. See the article on using notebook variables and parameters in Microsoft Fabric for the parameter-passing mechanics.

# ============================================================
# Cell 1: Parameters (toggle as Notebook Parameters cell in Fabric)
# ============================================================
LAKEHOUSE_NAME = "sales_lakehouse"
TARGET_MODEL_NAME = "Sales Analytics - Direct Lake"
TABLES_TO_CHECK = ["gold_customer_metrics", "gold_product_performance", "gold_order_summary"]
MAX_REFRESH_WAIT_SECONDS = 600
DRY_RUN = False  # Set True to detect drift without triggering refresh

# ============================================================
# Cell 2: Imports and Setup
# ============================================================
import sempy.fabric as fabric
import requests
import time
import json
from pyspark.sql import SparkSession
from typing import Optional

spark = SparkSession.builder.getOrCreate()

# Resolve workspace and dataset IDs
workspace_id = fabric.get_workspace_id()
models = fabric.list_datasets()
matched = models[models['Dataset Name'] == TARGET_MODEL_NAME]

if matched.empty:
    raise ValueError(f"Semantic model '{TARGET_MODEL_NAME}' not found in workspace.")

dataset_id = matched.iloc[0]['Dataset Id']
print(f"Targeting model: {TARGET_MODEL_NAME}")
print(f"  Workspace ID: {workspace_id}")
print(f"  Dataset ID: {dataset_id}")

# ============================================================
# Cell 3: Schema Drift Detection
# ============================================================

def get_delta_columns(table_name: str) -> dict:
    """Return {col_name_lower: spark_type_str} for a Delta table."""
    df = spark.read.format("delta").load(f"Tables/{table_name}")
    return {field.name.lower(): str(field.dataType) for field in df.schema.fields}

def get_model_columns(table_name: str) -> dict:
    """Return {col_name_lower: pbi_type_str} for a table in the semantic model."""
    all_cols = fabric.list_columns(dataset=TARGET_MODEL_NAME)
    table_cols = all_cols[
        (all_cols['Table Name'] == table_name) &
        (all_cols['Column Type'] == 'Data')
    ]
    return {row['Column Name'].lower(): row['Data Type'] for _, row in table_cols.iterrows()}

drift_report = {}
any_drift_detected = False

for table in TABLES_TO_CHECK:
    print(f"\nChecking: {table}")
    try:
        delta_cols = get_delta_columns(table)
        model_cols = get_model_columns(table)
        
        added = set(delta_cols.keys()) - set(model_cols.keys())
        removed = set(model_cols.keys()) - set(delta_cols.keys())
        
        drift_report[table] = {
            "added_columns": list(added),
            "removed_columns": list(removed),
            "drift_detected": bool(added or removed)
        }
        
        if added:
            print(f"  ✓ Columns added: {added}")
        if removed:
            print(f"  ✓ Columns removed: {removed}")
        if not added and not removed:
            print(f"  — No drift detected.")
        
        if added or removed:
            any_drift_detected = True
            
    except Exception as e:
        print(f"  ERROR checking {table}: {e}")
        drift_report[table] = {"error": str(e), "drift_detected": False}

print(f"\nDrift summary: {'DRIFT DETECTED' if any_drift_detected else 'All schemas in sync'}")
print(json.dumps(drift_report, indent=2))

# ============================================================
# Cell 4: Trigger Metadata Refresh (if drift detected)
# ============================================================

def get_auth_headers() -> dict:
    token = notebookutils.credentials.getToken("pbi")
    return {
        "Authorization": f"Bearer {token}",
        "Content-Type": "application/json"
    }

def trigger_refresh(workspace_id: str, dataset_id: str) -> Optional[str]:
    headers = get_auth_headers()
    url = f"https://api.fabric.microsoft.com/v1/workspaces/{workspace_id}/datasets/{dataset_id}/refreshes"
    payload = {"type": "full", "commitMode": "transactional", "maxParallelism": 1, "retryCount": 0}
    
    response = requests.post(url, headers=headers, json=payload)
    if response.status_code == 202:
        return response.headers.get("x-ms-request-id")
    else:
        raise RuntimeError(f"Refresh trigger failed [{response.status_code}]: {response.text}")

def wait_for_refresh(workspace_id: str, dataset_id: str, max_wait: int = 600) -> str:
    url = f"https://api.fabric.microsoft.com/v1/workspaces/{workspace_id}/datasets/{dataset_id}/refreshes"
    elapsed = 0
    interval = 15
    
    while elapsed < max_wait:
        time.sleep(interval)
        elapsed += interval
        
        headers = get_auth_headers()  # refresh token each poll
        response = requests.get(url, headers=headers)
        response.raise_for_status()
        
        history = response.json().get("value", [])
        if not history:
            continue
        
        status = history[0].get("status", "Unknown")
        print(f"  [{elapsed}s] Status: {status}")
        
        if status in ("Completed", "Failed", "Cancelled", "Unknown"):
            if status == "Failed":
                error_detail = history[0].get("serviceExceptionJson", "")
                raise RuntimeError(f"Refresh failed: {error_detail}")
            return status
    
    raise TimeoutError(f"Refresh did not complete in {max_wait}s")

# Main execution logic
if any_drift_detected:
    if DRY_RUN:
        print("\nDRY RUN: Drift detected but refresh suppressed. Exiting.")
    else:
        print("\nTriggering metadata refresh on semantic model...")
        request_id = trigger_refresh(workspace_id, dataset_id)
        print(f"Refresh accepted. Request ID: {request_id}")
        
        final_status = wait_for_refresh(workspace_id, dataset_id, MAX_REFRESH_WAIT_SECONDS)
        print(f"\nRefresh completed: {final_status}")
        
        # Output for pipeline consumption
        mssparkutils.notebook.exit(json.dumps({
            "drift_detected": True,
            "refresh_triggered": True,
            "refresh_status": final_status,
            "drift_details": drift_report
        }))
else:
    print("\nNo schema drift. Metadata refresh not required.")
    mssparkutils.notebook.exit(json.dumps({
        "drift_detected": False,
        "refresh_triggered": False,
        "drift_details": drift_report
    }))

The mssparkutils.notebook.exit() call at the end pushes a JSON string back to the calling pipeline, where you can use an If Condition activity to branch on drift_detected or alert on refresh_status != "Completed". See orchestrating multi-notebook workflows in Microsoft Fabric for how to wire up output variables in pipeline notebook activities.

---

Integrating the Refresh into a Data Pipeline

The notebook above is designed to be called as a step inside a Fabric data pipeline, positioned after your Spark transformation notebook that performs the schema-evolving writes. Here's the pipeline structure you should build:

1. Notebook Activity: Transform Data — Your gold-layer transformation that writes or evolves the Delta tables. This is where column additions happen, via writing data with PySpark to a Delta table using append, overwrite, and merge patterns.

2. Notebook Activity: Schema Sync — The notebook we just built. Set it to run on success of Activity 1. Pass parameters like TABLES_TO_CHECK dynamically using pipeline parameters so the same notebook works across multiple environments.

3. If Condition Activity — Branch on the drift_detected flag from the schema sync output.

4. Teams/Email Activity (optional) — In the True branch, send a notification that schema drift was detected and remediated.

To pass pipeline parameters into the notebook, use the Base Parameters configuration in the notebook activity settings. Parameters flow in as named string values and are received by the notebook's parameter cell:

{
  "LAKEHOUSE_NAME": "@pipeline().parameters.lakehouseName",
  "TARGET_MODEL_NAME": "@pipeline().parameters.semanticModelName",
  "TABLES_TO_CHECK": "@join(pipeline().parameters.tableList, ',')"
}

Note: TABLES_TO_CHECK arrives as a comma-delimited string, so add a parsing step in the notebook parameter cell:

# In the parameter cell, handle both list and string forms
if isinstance(TABLES_TO_CHECK, str):
    TABLES_TO_CHECK = [t.strip() for t in TABLES_TO_CHECK.split(",")]

---

Advanced Patterns and Edge Cases

Handling Multiple Semantic Models Against the Same Lakehouse

Large organizations often have multiple semantic models pointing at the same Lakehouse — a certified dataset for finance, a departmental model for operations, a live dashboard model. Your schema sync needs to refresh all of them, not just the first one found.

# Refresh multiple models that target the same lakehouse
TARGET_MODELS = [
    "Sales Analytics - Direct Lake",
    "Finance Dashboard - Direct Lake",
    "Operations Summary - Direct Lake"
]

refresh_results = {}

for model_name in TARGET_MODELS:
    matched = models[models['Dataset Name'] == model_name]
    if matched.empty:
        print(f"Model not found: {model_name}, skipping.")
        continue
    
    model_dataset_id = matched.iloc[0]['Dataset Id']
    
    try:
        req_id = trigger_refresh(workspace_id, model_dataset_id)
        status = wait_for_refresh(workspace_id, model_dataset_id, max_wait=300)
        refresh_results[model_name] = {"status": status, "request_id": req_id}
    except Exception as e:
        refresh_results[model_name] = {"status": "error", "error": str(e)}
        print(f"ERROR refreshing {model_name}: {e}")

print(json.dumps(refresh_results, indent=2))

Note that refreshes are sequential in this loop. For a large number of models, you might want parallel execution — but be aware that the Fabric capacity throttles concurrent refresh operations, and hammering the API from a single notebook can trigger rate limiting. Sequential with a small sleep between calls is safer.

Detecting Column Type Changes

If your team changes a column's data type (e.g., migrating from INT to BIGINT after a business growth event), the column name stays the same but the type changes. Your presence-based drift detection won't catch this. Here's an extension:

# Spark to Power BI type mapping (partial — extend as needed)
SPARK_TO_PBI_TYPE = {
    "StringType()": "Text",
    "LongType()": "Int64",
    "IntegerType()": "Int64",
    "DoubleType()": "Double",
    "FloatType()": "Double",
    "DecimalType(10,2)": "Decimal",
    "BooleanType()": "True/False",
    "DateType()": "DateTime",
    "TimestampType()": "DateTime",
}

type_changes = {}
for col_name in delta_cols.keys() & model_cols.keys():
    spark_type = delta_cols[col_name]
    expected_pbi_type = SPARK_TO_PBI_TYPE.get(spark_type, "Unknown")
    actual_pbi_type = model_cols[col_name]
    
    if expected_pbi_type != "Unknown" and expected_pbi_type != actual_pbi_type:
        type_changes[col_name] = {
            "spark_type": spark_type,
            "expected_pbi_type": expected_pbi_type,
            "actual_pbi_type": actual_pbi_type
        }

if type_changes:
    print(f"Type changes detected: {type_changes}")
    any_drift_detected = True

Respecting Rate Limits and Capacity Throttling

The Fabric REST API for dataset refreshes enforces rate limits that scale with your capacity SKU. On an F2, you can run roughly 8 on-demand refreshes per hour per dataset. On F64+, this ceiling is much higher. If your pipelines are frequent and you're on a smaller SKU, you should add exponential backoff:

import random

def trigger_refresh_with_backoff(workspace_id, dataset_id, max_retries=3):
    for attempt in range(max_retries):
        try:
            return trigger_refresh(workspace_id, dataset_id)
        except RuntimeError as e:
            if "429" in str(e) and attempt < max_retries - 1:
                wait = (2 ** attempt) + random.uniform(0, 1)
                print(f"Rate limited. Waiting {wait:.1f}s before retry {attempt + 2}/{max_retries}...")
                time.sleep(wait)
            else:
                raise

What Happens When You Delete a Column

Column removal is trickier than column addition. If you drop a column from a Delta table, the semantic model still has it registered. Measures, relationships, and report visuals that reference that column will break at query time, not at refresh time. The metadata refresh will update the frame and should remove the column from the schema — but if there are calculated columns or measures that reference the now-deleted column, the refresh itself will fail with a dependency error.

Best practice: before dropping a column from a Delta table, audit the semantic model for dependencies on that column:

# Check for measures that might reference a column we're about to remove
measures = fabric.list_measures(dataset=TARGET_MODEL_NAME)
target_col = "old_revenue_column"

# Simple text search in measure expressions
dependent_measures = measures[
    measures['Measure Expression'].str.contains(target_col, case=False, na=False)
]

if not dependent_measures.empty:
    print(f"WARNING: {len(dependent_measures)} measures reference '{target_col}':")
    print(dependent_measures[['Table Name', 'Measure Name', 'Measure Expression']])
    raise ValueError("Cannot remove column with active measure dependencies. Update measures first.")

---

Hands-On Exercise

Work through this end-to-end scenario in your own Fabric environment:

Scenario: You have a gold Delta table called gold_sales_summary with columns order_date, product_id, region, revenue, and units_sold. A Direct Lake semantic model called "Sales Summary Direct Lake" is built on this table.

Steps:

1. Open a Fabric notebook and use sempy.fabric.list_columns() to verify the current columns registered in your semantic model. Print them out.

2. In a second notebook cell, add a new column to your Delta table by running a Spark overwrite that adds a gross_margin column. Use spark.read.format("delta").load("Tables/gold_sales_summary") to verify the column exists in the Delta schema.

3. Re-run your list_columns() call. Confirm the new column is not visible in the semantic model yet. This is your proof of the schema drift problem.

4. Run the drift detection logic from this article against the table. Confirm schema_changed = True.

5. Trigger a metadata refresh using the REST API approach. Poll for completion and print the final status.

6. After the refresh completes, re-run list_columns() one more time. Confirm gross_margin now appears in the model.

7. Bonus: Navigate to Power BI and open a report built on this model. Add the gross_margin column to a visualization and confirm data is being read correctly through Direct Lake.

8. Challenge extension: Modify the notebook to accept TABLES_TO_CHECK as a pipeline parameter, add it to a data pipeline, and schedule it to run 5 minutes after your main transformation notebook completes.

---

Common Mistakes & Troubleshooting

Refresh API returns 401 Unauthorized

This almost always means the token was fetched before the user's session was fully authenticated, or the notebook is running under an identity that lacks semantic model write permissions. Verify that the notebook runner (user or service principal) has at least Contributor role on the workspace, or explicit Write permission on the semantic model item. The notebookutils.credentials.getToken("pbi") call will succeed even if the resulting token lacks the right scope — the 401 only surfaces when you actually make the API call.

fabric.list_columns() returns an empty DataFrame

The most common cause: the semantic model name doesn't match exactly (including case and spaces). Print fabric.list_datasets() and compare the Dataset Name column carefully against your TARGET_MODEL_NAME variable. Also check that the model hasn't been moved to a different workspace — sempy operates on the workspace the notebook is attached to.

Refresh status stays "Unknown" indefinitely

For Direct Lake models, a framing-only operation (no data movement) can complete so quickly that the refresh history entry is created and finalized in the same polling window you're missing. Add a short initial sleep (5-10 seconds) before your first poll and lower the polling interval. If "Unknown" persists, it may also indicate the model is in a state where framing is disabled — check whether the model has fallen back to DirectQuery mode.

Schema drift detected but new columns still don't appear after refresh

Check whether the column was actually committed to the Delta log. Use DESCRIBE HISTORY in a Spark SQL cell to verify the transaction that added the column actually completed:

# Verify the column addition is in the Delta log
spark.sql(f"DESCRIBE HISTORY delta.`Tables/gold_customer_metrics`").show(5, truncate=False)

If the most recent operation was a failed write, the column may exist in Spark's schema cache but not in the committed Delta log. The semantic model reads from the committed log, so it will correctly not show the column.

Rate limit errors (429) during pipeline runs

If multiple pipelines are triggering semantic model refreshes in parallel (common in environments with many tables and medallion layers), consolidate your refresh calls. One schema sync notebook at the end of your entire daily pipeline run, checking all tables at once, is far better than per-table refresh triggers scattered across individual activities.

mssparkutils.notebook.exit() output not visible in pipeline

The notebook exit value must be a string, not a dictionary. Always serialize with json.dumps() before passing to exit(). In the pipeline's notebook activity, access the output via @activity('YourNotebookActivityName').output.result.exitValue and parse it with the json() expression function.

---

Summary & Next Steps

You now have a complete, production-ready automation for keeping Direct Lake semantic models in sync with evolving Delta table schemas. The core insight driving all of this is that Direct Lake's performance advantage — reading directly from Delta Parquet files without data movement — comes with a trade-off: schema metadata is a point-in-time snapshot that must be explicitly refreshed when table structure changes. Manual refreshes are a liability; automated detection and remediation is the professional approach.

What you built today:

• A schema drift detector that compares Delta table schemas (via PySpark) against semantic model column registrations (via sempy)
• A Fabric REST API integration that triggers metadata refresh only when drift is detected
• Polling logic that waits for completion with token refresh on each poll iteration
• A pipeline-ready notebook with parameter support and structured exit values
• Advanced patterns for multi-model environments, type change detection, and rate limit handling

Where to go from here:

• Explore optimizing Delta table performance with V-Order, OPTIMIZE, VACUUM, and Z-Order — the framing operation reads from Parquet files, and well-optimized Delta tables produce faster frame establishment
• Review implementing incremental refresh for Direct Lake semantic models to understand how partitioning strategies interact with the framing operations you're now triggering automatically
• Consider Fabric Git integration and deployment pipelines as the right place to version-control your schema sync notebook — this automation is infrastructure code and deserves the same treatment as your transformation notebooks

The schema sync notebook you built today is the kind of automation that quietly saves hours of confused debugging every month. Put it in your pipeline, commit it to Git, and document it in your team's runbook. Your future self — and the business analyst who was waiting for those new columns — will thank you.