Implementing a Fabric Notebook-Based Data Quality Framework: Validating Row Counts, Null Thresholds, and Referential Integrity Across Medallion Layers Before Pipeline Promotion

Introduction

Picture this: your overnight pipeline finishes at 3 AM, your Power BI reports refresh on schedule, and your stakeholders open their dashboards at 9 AM to discover that the sales figures for the entire Northeast region are missing — because a source system schema change silently dropped a join key, and your pipeline happily promoted that broken data from bronze all the way to gold without complaint. Nobody gets alerted. The data just quietly disappears.

This is the failure mode that a data quality framework is designed to prevent. Not just "did the pipeline run?" but "is the data that ran through the pipeline actually correct enough to trust?" The distinction matters enormously in production, and it's the gap between a data platform that engineering thinks is working and one that stakeholders actually believe in.

In this lesson, you'll build a reusable, notebook-based data quality framework in Microsoft Fabric that runs validation checks — row count comparisons, null threshold enforcement, and referential integrity verification — at each layer of the medallion architecture before data gets promoted to the next layer. The framework will produce structured pass/fail results, log them to a dedicated Delta table, and integrate with Fabric data pipelines so that a failed check stops the pipeline rather than silently allowing bad data downstream.

What you'll learn:

• How to design a structured quality check result schema and log validation outcomes to a Delta table in your lakehouse
• How to implement row count variance checks that catch both data loss and unexpected data explosions
• How to enforce configurable null thresholds on critical columns across bronze, silver, and gold tables
• How to validate referential integrity between related Delta tables — for example, ensuring every fact row has a matching dimension key
• How to surface check results as a pipeline exit condition so that a failed quality gate actually stops promotion

---

Prerequisites

You should be comfortable with:

• PySpark and Delta Lake operations in Fabric notebooks — if you need a refresher, start with Transforming Data with Spark Notebooks in Microsoft Fabric: PySpark for Lakehouse Tables
• The medallion architecture pattern — we'll be using bronze/silver/gold layers throughout. See Implementing the Medallion Architecture in Microsoft Fabric: Bronze, Silver, and Gold Layers if you need grounding
• Basic Fabric pipeline construction, including notebook activities and activity dependencies — covered in Orchestrating Multi-Notebook Workflows in Microsoft Fabric: Using Pipeline Notebook Activities, Activity Dependencies, and Output Variables to Chain PySpark Transformations Across Medallion Layers
• Parameterizing notebooks from pipelines — see Using Notebook Variables and Parameters in Microsoft Fabric: Passing Pipeline Values to PySpark for Dynamic Lakehouse Ingestion

You'll need a Fabric workspace with at least one lakehouse containing bronze, silver, and gold Delta tables. A retail scenario — orders, customers, and products — works well for the examples here.

---

Designing the Quality Check Schema

Before writing a single validation, you need to answer a question that most teams skip: where do the results live, and what shape do they take? Without a consistent result schema, you end up with notebooks that print pass/fail to stdout and leave no audit trail. Six months later, you can't answer "when did null rates on customer_email first start rising?"

Here's a schema that balances completeness with practicality:

from pyspark.sql.types import (
    StructType, StructField, StringType, 
    IntegerType, DoubleType, BooleanType, TimestampType
)

quality_check_schema = StructType([
    StructField("run_id",           StringType(),   nullable=False),
    StructField("pipeline_run_dt",  TimestampType(), nullable=False),
    StructField("layer",            StringType(),   nullable=False),  # bronze/silver/gold
    StructField("table_name",       StringType(),   nullable=False),
    StructField("check_type",       StringType(),   nullable=False),  # row_count/null_threshold/ref_integrity
    StructField("check_name",       StringType(),   nullable=False),
    StructField("expected_value",   DoubleType(),   nullable=True),
    StructField("actual_value",     DoubleType(),   nullable=True),
    StructField("threshold",        DoubleType(),   nullable=True),
    StructField("passed",           BooleanType(),  nullable=False),
    StructField("failure_message",  StringType(),   nullable=True),
])

Each field earns its place. run_id ties all checks from a single pipeline execution together so you can query "show me every check from last night's run." layer tells you where in the medallion stack the check fired. expected_value and actual_value let you trend metrics over time — not just pass/fail, but by how much. failure_message surfaces a human-readable explanation that can feed directly into pipeline alert emails.

Let's create the results table once, at setup time:

from delta.tables import DeltaTable
from pyspark.sql import SparkSession

spark = SparkSession.builder.getOrCreate()

# Create an empty Delta table to receive quality check results
dq_table_path = "Tables/dq_check_results"

if not DeltaTable.isDeltaTable(spark, f"abfss://your-workspace@onelake.dfs.fabric.microsoft.com/dq_lakehouse.Lakehouse/{dq_table_path}"):
    empty_df = spark.createDataFrame([], quality_check_schema)
    (empty_df.write
        .format("delta")
        .mode("overwrite")
        .save(dq_table_path))
    
print("DQ results table ready.")

In practice within Fabric notebooks, you can use the shorthand spark.sql("CREATE TABLE IF NOT EXISTS dq_lakehouse.dq_check_results ...") once your lakehouse is attached. Either approach works — the Delta table is what matters.

---

Building the Check Runner Infrastructure

Rather than writing one-off validation logic scattered across cells, you'll build a small helper class that accumulates results, evaluates them, and writes everything to your audit table in a single batch. This is the core engine of the framework.

import uuid
from datetime import datetime
from pyspark.sql import Row

class DataQualityRunner:
    """
    Accumulates quality check results during a notebook run,
    then writes them to the DQ audit table as a single batch.
    """
    
    def __init__(self, layer: str, pipeline_run_dt: datetime = None):
        self.run_id = str(uuid.uuid4())
        self.layer = layer
        self.pipeline_run_dt = pipeline_run_dt or datetime.utcnow()
        self.results = []
    
    def _record(
        self, 
        table_name: str, 
        check_type: str, 
        check_name: str,
        expected: float, 
        actual: float, 
        threshold: float,
        passed: bool, 
        failure_message: str = None
    ):
        self.results.append(Row(
            run_id=self.run_id,
            pipeline_run_dt=self.pipeline_run_dt,
            layer=self.layer,
            table_name=table_name,
            check_type=check_type,
            check_name=check_name,
            expected_value=float(expected) if expected is not None else None,
            actual_value=float(actual) if actual is not None else None,
            threshold=float(threshold) if threshold is not None else None,
            passed=passed,
            failure_message=failure_message
        ))
    
    def any_failures(self) -> bool:
        return any(not r.passed for r in self.results)
    
    def failed_checks(self) -> list:
        return [r for r in self.results if not r.passed]
    
    def write_results(self, dq_table_name: str = "dq_lakehouse.dq_check_results"):
        if not self.results:
            print("No results to write.")
            return
        results_df = spark.createDataFrame(self.results, schema=quality_check_schema)
        (results_df.write
            .format("delta")
            .mode("append")
            .saveAsTable(dq_table_name))
        print(f"Wrote {len(self.results)} check result(s) for run_id={self.run_id}")
    
    def summary(self):
        total = len(self.results)
        passed = sum(1 for r in self.results if r.passed)
        failed = total - passed
        print(f"\n{'='*50}")
        print(f"Quality Check Summary — Layer: {self.layer.upper()}")
        print(f"Run ID: {self.run_id}")
        print(f"Total checks: {total} | Passed: {passed} | Failed: {failed}")
        if failed > 0:
            print("\nFailed checks:")
            for r in self.failed_checks():
                print(f"  ✗ [{r.table_name}] {r.check_name}: {r.failure_message}")
        print(f"{'='*50}\n")

This class is intentionally lightweight. It doesn't try to be a full Great Expectations replacement — it's a practical, Fabric-native pattern you can own, extend, and debug without external dependencies or complex configuration files.

---

Implementing Row Count Validation

Row count checks are your first line of defense. They catch the most common failure modes: a source extract that returned zero rows because of a network hiccup, a filter condition that accidentally excluded a partition, or a join explosion that multiplied rows unexpectedly.

There are two useful patterns: an absolute minimum row count, and a variance check against the previous load.

def check_row_count_minimum(
    runner: DataQualityRunner,
    table_name: str,
    min_rows: int
):
    """Ensure a table meets a minimum absolute row count."""
    df = spark.table(table_name)
    actual_count = df.count()
    passed = actual_count >= min_rows
    
    runner._record(
        table_name=table_name,
        check_type="row_count",
        check_name=f"min_row_count_{table_name}",
        expected=float(min_rows),
        actual=float(actual_count),
        threshold=float(min_rows),
        passed=passed,
        failure_message=(
            None if passed else
            f"Expected at least {min_rows:,} rows, found {actual_count:,}"
        )
    )
    return actual_count


def check_row_count_variance(
    runner: DataQualityRunner,
    table_name: str,
    reference_table: str,
    max_variance_pct: float = 0.15
):
    """
    Compare current row count against a reference (e.g., previous layer or previous run).
    Fails if variance exceeds max_variance_pct in either direction.
    """
    current_count = spark.table(table_name).count()
    reference_count = spark.table(reference_table).count()
    
    if reference_count == 0:
        # Can't compute variance against zero — treat as warning, not failure
        runner._record(
            table_name=table_name,
            check_type="row_count",
            check_name=f"row_count_variance_{table_name}_vs_{reference_table}",
            expected=0.0,
            actual=float(current_count),
            threshold=max_variance_pct,
            passed=True,
            failure_message="Reference table has zero rows; variance check skipped"
        )
        return current_count
    
    variance = abs(current_count - reference_count) / reference_count
    passed = variance <= max_variance_pct
    
    runner._record(
        table_name=table_name,
        check_type="row_count",
        check_name=f"row_count_variance_{table_name}_vs_{reference_table}",
        expected=float(reference_count),
        actual=float(current_count),
        threshold=max_variance_pct,
        passed=passed,
        failure_message=(
            None if passed else
            f"Row count variance {variance:.1%} exceeds threshold {max_variance_pct:.1%}. "
            f"Current: {current_count:,}, Reference: {reference_count:,}"
        )
    )
    return current_count

In practice, you'd wire these up like this for a silver layer check before promoting to gold:

runner = DataQualityRunner(layer="silver")

# Absolute floor: silver orders must have at least 10,000 rows
check_row_count_minimum(runner, "silver_lakehouse.orders", min_rows=10_000)

# Variance check: silver row count shouldn't deviate more than 20% from bronze
check_row_count_variance(
    runner,
    table_name="silver_lakehouse.orders",
    reference_table="bronze_lakehouse.raw_orders",
    max_variance_pct=0.20
)

---

Implementing Null Threshold Validation

Null checks are where you protect the business logic that downstream consumers depend on. A customer_id that's 5% null in the silver layer means 5% of your fact rows will have no matching dimension — a silent referential integrity problem that shows up as mysterious "Unknown Customer" lines in reports.

def check_null_threshold(
    runner: DataQualityRunner,
    table_name: str,
    column_name: str,
    max_null_pct: float
):
    """
    Ensure that the null rate for a given column does not exceed max_null_pct.
    max_null_pct = 0.0 means no nulls permitted at all.
    max_null_pct = 0.05 means up to 5% nulls are acceptable.
    """
    from pyspark.sql.functions import col, count, when
    
    df = spark.table(table_name)
    total_rows = df.count()
    
    if total_rows == 0:
        runner._record(
            table_name=table_name,
            check_type="null_threshold",
            check_name=f"null_pct_{table_name}.{column_name}",
            expected=0.0,
            actual=0.0,
            threshold=max_null_pct,
            passed=True,
            failure_message="Table is empty; null check skipped"
        )
        return
    
    null_count = df.filter(col(column_name).isNull()).count()
    actual_null_pct = null_count / total_rows
    passed = actual_null_pct <= max_null_pct
    
    runner._record(
        table_name=table_name,
        check_type="null_threshold",
        check_name=f"null_pct_{table_name}.{column_name}",
        expected=float(max_null_pct),
        actual=float(actual_null_pct),
        threshold=float(max_null_pct),
        passed=passed,
        failure_message=(
            None if passed else
            f"Column '{column_name}' null rate {actual_null_pct:.2%} exceeds "
            f"threshold {max_null_pct:.2%} ({null_count:,} of {total_rows:,} rows)"
        )
    )

For a batch of column checks on the same table, it's more efficient to compute all null counts in a single aggregation pass rather than scanning the table once per column:

def check_null_thresholds_batch(
    runner: DataQualityRunner,
    table_name: str,
    column_thresholds: dict  # {"column_name": max_null_pct}
):
    """
    Efficient batch null check: one table scan for multiple columns.
    column_thresholds = {"customer_id": 0.0, "order_date": 0.0, "email": 0.05}
    """
    from pyspark.sql.functions import col, count, when, lit
    
    df = spark.table(table_name)
    total_rows = df.count()
    
    if total_rows == 0:
        for col_name in column_thresholds:
            runner._record(
                table_name=table_name,
                check_type="null_threshold",
                check_name=f"null_pct_{table_name}.{col_name}",
                expected=0.0, actual=0.0,
                threshold=column_thresholds[col_name],
                passed=True,
                failure_message="Table is empty; null check skipped"
            )
        return
    
    # Build aggregation expressions for all columns at once
    agg_exprs = [
        count(when(col(c).isNull(), lit(1))).alias(c)
        for c in column_thresholds.keys()
    ]
    null_counts = df.agg(*agg_exprs).collect()[0]
    
    for col_name, max_null_pct in column_thresholds.items():
        null_count = null_counts[col_name]
        actual_null_pct = null_count / total_rows
        passed = actual_null_pct <= max_null_pct
        
        runner._record(
            table_name=table_name,
            check_type="null_threshold",
            check_name=f"null_pct_{table_name}.{col_name}",
            expected=float(max_null_pct),
            actual=float(actual_null_pct),
            threshold=float(max_null_pct),
            passed=passed,
            failure_message=(
                None if passed else
                f"Column '{col_name}' null rate {actual_null_pct:.2%} exceeds "
                f"threshold {max_null_pct:.2%} ({null_count:,}/{total_rows:,} rows)"
            )
        )

Usage looks like:

check_null_thresholds_batch(
    runner,
    table_name="silver_lakehouse.orders",
    column_thresholds={
        "order_id":    0.00,   # Zero tolerance — primary key equivalent
        "customer_id": 0.00,   # Required for dimension join
        "order_date":  0.00,   # Required for time intelligence
        "product_sku": 0.01,   # Allow up to 1% for legacy data gaps
        "shipping_zip": 0.05,  # Up to 5% for international orders
    }
)

---

Implementing Referential Integrity Checks

Referential integrity failures are the sneakiest data quality problem because the data looks complete — all the rows are there, the null rates are fine — but the relationships between tables are broken. A fact row with a product_sku of "ABC-999" means nothing if the products dimension table doesn't contain "ABC-999."

This matters especially if you're building toward Direct Lake Mode in Power BI reporting, where broken dimension relationships result in blank visuals rather than error messages.

def check_referential_integrity(
    runner: DataQualityRunner,
    fact_table: str,
    fact_key_column: str,
    dim_table: str,
    dim_key_column: str,
    max_orphan_pct: float = 0.0
):
    """
    Check that fact_key_column values in fact_table all exist in dim_table.
    max_orphan_pct = 0.0 means every fact row must have a matching dimension key.
    max_orphan_pct = 0.02 allows up to 2% orphaned fact rows.
    """
    fact_df = spark.table(fact_table).select(fact_key_column).dropDuplicates()
    dim_df = spark.table(dim_table).select(dim_key_column).dropDuplicates()
    
    # Left anti-join: fact keys that have NO match in the dimension
    orphaned_keys = fact_df.join(
        dim_df,
        fact_df[fact_key_column] == dim_df[dim_key_column],
        how="left_anti"
    )
    
    orphan_key_count = orphaned_keys.count()
    total_fact_keys = fact_df.count()
    
    if total_fact_keys == 0:
        runner._record(
            table_name=fact_table,
            check_type="ref_integrity",
            check_name=f"ref_integrity_{fact_table}.{fact_key_column}_vs_{dim_table}",
            expected=0.0, actual=0.0,
            threshold=max_orphan_pct,
            passed=True,
            failure_message="Fact table has zero distinct keys; check skipped"
        )
        return
    
    orphan_pct = orphan_key_count / total_fact_keys
    passed = orphan_pct <= max_orphan_pct
    
    # Capture a sample of orphaned keys for the failure message
    sample_orphans = []
    if not passed:
        sample_orphans = [
            str(row[fact_key_column]) 
            for row in orphaned_keys.limit(5).collect()
        ]
    
    runner._record(
        table_name=fact_table,
        check_type="ref_integrity",
        check_name=f"ref_integrity_{fact_table}.{fact_key_column}_vs_{dim_table}",
        expected=float(max_orphan_pct),
        actual=float(orphan_pct),
        threshold=float(max_orphan_pct),
        passed=passed,
        failure_message=(
            None if passed else
            f"{orphan_key_count:,} orphaned keys ({orphan_pct:.2%}) in "
            f"'{fact_table}.{fact_key_column}' not found in '{dim_table}.{dim_key_column}'. "
            f"Sample orphans: {sample_orphans}"
        )
    )

The left anti-join is the right tool here: it returns only fact-side keys that have no match on the dimension side, which is exactly the set of orphaned records. The .dropDuplicates() calls are important — you want to check the set of distinct key values, not the raw row counts, which could make a single bad key look catastrophically large.

---

Assembling a Complete Layer Validation Notebook

Now let's put this together into a complete notebook for silver-to-gold promotion validation. This is the notebook that runs after silver transformation and before any gold layer write, configured to accept pipeline parameters.

# Cell 1 — Parameters (mark this cell as a "parameter cell" in Fabric)
layer = "silver"
pipeline_run_id = "manual_run"
pipeline_run_dt_str = "2024-01-15T03:00:00"
fail_on_error = True  # Set to False during initial calibration

# Cell 2 — Imports and setup
from datetime import datetime
import sys

pipeline_run_dt = datetime.fromisoformat(pipeline_run_dt_str)
runner = DataQualityRunner(layer=layer, pipeline_run_dt=pipeline_run_dt)

# Cell 3 — Row count checks
print("Running row count checks...")

check_row_count_minimum(
    runner, "silver_lakehouse.orders", min_rows=5_000
)
check_row_count_minimum(
    runner, "silver_lakehouse.customers", min_rows=1_000
)
check_row_count_minimum(
    runner, "silver_lakehouse.products", min_rows=50
)

check_row_count_variance(
    runner,
    table_name="silver_lakehouse.orders",
    reference_table="bronze_lakehouse.raw_orders",
    max_variance_pct=0.20
)

# Cell 4 — Null threshold checks
print("Running null threshold checks...")

check_null_thresholds_batch(
    runner,
    table_name="silver_lakehouse.orders",
    column_thresholds={
        "order_id":    0.00,
        "customer_id": 0.00,
        "order_date":  0.00,
        "product_sku": 0.01,
        "order_total": 0.00,
        "shipping_zip": 0.05,
    }
)

check_null_thresholds_batch(
    runner,
    table_name="silver_lakehouse.customers",
    column_thresholds={
        "customer_id":  0.00,
        "email":        0.02,
        "signup_date":  0.00,
        "country_code": 0.01,
    }
)

# Cell 5 — Referential integrity checks
print("Running referential integrity checks...")

check_referential_integrity(
    runner,
    fact_table="silver_lakehouse.orders",
    fact_key_column="customer_id",
    dim_table="silver_lakehouse.customers",
    dim_key_column="customer_id",
    max_orphan_pct=0.00
)

check_referential_integrity(
    runner,
    fact_table="silver_lakehouse.orders",
    fact_key_column="product_sku",
    dim_table="silver_lakehouse.products",
    dim_key_column="sku",
    max_orphan_pct=0.01  # 1% tolerance for new products not yet in catalog
)

# Cell 6 — Write results and evaluate
runner.write_results("dq_lakehouse.dq_check_results")
runner.summary()

if fail_on_error and runner.any_failures():
    failed = runner.failed_checks()
    failure_summary = "; ".join([f.check_name for f in failed])
    raise ValueError(
        f"Data quality checks failed for layer '{layer}'. "
        f"Failed checks: {failure_summary}. "
        f"Run ID: {runner.run_id}. See dq_check_results for details."
    )

print(f"All quality checks passed. Proceeding with {layer} → gold promotion.")

The raise ValueError at the end is the crucial integration point. When this notebook is called from a Fabric data pipeline as a Notebook activity, a raised exception causes the activity to fail with a non-zero exit code — which the pipeline can then use to stop execution and trigger alert notifications, rather than blithely running the next activity to write bad data into gold.

---

Wiring the Framework Into a Pipeline

The quality notebook sits between your transformation step and your promotion step. In your Fabric pipeline, the structure looks like this:

1. Activity: Transform_Silver — your existing PySpark transformation notebook
2. Activity: Validate_Silver_Quality — the validation notebook above, with dependsOn: Transform_Silver (Succeeded)
3. Activity: Promote_Silver_to_Gold — your gold write notebook, with dependsOn: Validate_Silver_Quality (Succeeded)

This dependency chain means that if Validate_Silver_Quality raises an exception and fails, Promote_Silver_to_Gold never runs. The pipeline fails cleanly, your audit table has a full record of what went wrong, and you can trigger email alerts via Scheduling and Automating Fabric Data Pipeline Runs with Activity-Level Retries, Alerts, and Email Notifications.

Pass the pipeline's built-in run context into the validation notebook as parameters:

pipeline_run_id:  @pipeline().RunId
pipeline_run_dt_str: @formatDateTime(pipeline().TriggerTime, 'yyyy-MM-ddTHH:mm:ss')
layer: silver
fail_on_error: true

---

Querying the Audit Table for Trend Analysis

The real payoff of logging results to Delta is that you can now trend your data quality metrics over time. Open a new notebook or the SQL Analytics Endpoint and run queries like:

-- Show all failed checks in the last 7 days
SELECT 
    pipeline_run_dt,
    layer,
    table_name,
    check_name,
    actual_value,
    threshold,
    failure_message
FROM dq_lakehouse.dq_check_results
WHERE passed = FALSE
  AND pipeline_run_dt >= DATEADD(day, -7, CURRENT_TIMESTAMP())
ORDER BY pipeline_run_dt DESC;

-- Trend null rates for customer_id over time
SELECT 
    DATE(pipeline_run_dt) AS run_date,
    actual_value AS null_pct
FROM dq_lakehouse.dq_check_results
WHERE table_name = 'silver_lakehouse.orders'
  AND check_name LIKE '%customer_id%'
  AND check_type = 'null_threshold'
ORDER BY run_date;

-- Pass rate by layer per week
SELECT 
    layer,
    DATE_TRUNC('week', pipeline_run_dt) AS week_start,
    COUNT(*) AS total_checks,
    SUM(CASE WHEN passed THEN 1 ELSE 0 END) AS passed_checks,
    ROUND(100.0 * SUM(CASE WHEN passed THEN 1 ELSE 0 END) / COUNT(*), 1) AS pass_rate_pct
FROM dq_lakehouse.dq_check_results
GROUP BY layer, week_start
ORDER BY week_start DESC, layer;

This is the kind of visibility that moves data quality from a reactive "we found a problem" discipline to a proactive "we can see quality trending downward" discipline. You can connect this audit table directly to Power BI for a live quality dashboard — the querying lakehouse data with the SQL Analytics Endpoint article covers exactly how to expose it without a separate warehouse.

---

Hands-On Exercise

Build a complete quality framework for a three-table retail medallion setup. Here's the scenario and what to implement:

Setup: You have bronze tables (raw_orders, raw_customers, raw_products) that were loaded from a CSV source. Your silver transformation has run and produced silver_orders, silver_customers, silver_products.

Tasks:

1. Create the dq_check_results Delta table in a new dq_lakehouse in your workspace using the schema defined above.

2. Implement row count minimum checks for all three silver tables. Set thresholds at 80% of the bronze table counts (use a subquery or variable to compute this dynamically rather than hardcoding numbers).

3. Implement null threshold checks for these columns and tolerances:

   • silver_orders.order_id: 0%
   • silver_orders.customer_id: 0%
   • silver_orders.order_total: 0%
   • silver_customers.email: 3% (email is often optional)
   • silver_products.sku: 0%
   • silver_products.category: 2%

4. Implement referential integrity checks:

   • Every order.customer_id must exist in customers.customer_id (0% orphans)
   • Every order.product_sku must exist in products.sku (allow 2% orphans)

5. Run the validation notebook manually first with fail_on_error = False. Query dq_check_results and identify which checks passed and failed.

6. Create a Fabric data pipeline with three notebook activities: bronze-to-silver transformation, quality validation, and silver-to-gold promotion. Wire the dependencies so that the gold promotion only runs if quality validation succeeds.

7. Deliberately break one check (delete 10% of rows from raw_orders before the bronze ingestion to trigger the row count variance check) and verify that the pipeline stops at the validation step rather than writing bad data to gold.

---

Common Mistakes and Troubleshooting

The framework runs but never blocks the pipeline. Check that your validation notebook activity's fail_on_error parameter is actually receiving true from the pipeline. Also verify the notebook activity's "Failure" dependency path is not connected to the next activity — only "Succeeded" should trigger the gold promotion step. If both paths are connected, the pipeline runs regardless of outcome.

Row count checks are taking 10+ minutes on large tables. Delta tables cache metadata about row counts, but Spark still needs to execute the count action. For tables partitioned by date, add a filter to your count expressions to check only the current partition rather than the full history:

df = spark.table(table_name).filter(col("load_date") == current_date)

This transforms a full table scan into a partition-pruned scan. See Optimizing Delta Table Performance in a Fabric Lakehouse: V-Order, OPTIMIZE, VACUUM, and Z-Order for Faster Queries and Smaller Storage for partitioning strategies that make these counts dramatically faster.

Referential integrity check is returning false negatives. This usually means a data type mismatch between the fact key column and the dimension key column. A customer_id stored as LongType in orders and StringType in customers will produce 100% orphan rate even if the values are identical. Add a .cast(StringType()) to both sides of the join before comparing:

fact_df = spark.table(fact_table).select(col(fact_key_column).cast("string").alias("key"))
dim_df = spark.table(dim_table).select(col(dim_key_column).cast("string").alias("key"))
orphaned = fact_df.join(dim_df, on="key", how="left_anti")

write_results() fails because the DQ table doesn't exist yet. The setup step that creates the table must run before any validation notebooks execute. Add it as the first notebook activity in your pipeline, with a dependsOn: None configuration and an IF NOT EXISTS guard so it's idempotent on subsequent runs.

Thresholds that were calibrated in dev are failing in production. This is almost always because production volumes are genuinely different from dev/test volumes. Use the trend query from the audit table section to look at 30 days of production runs after deploying with fail_on_error = false, then set thresholds at the 5th percentile of observed values — giving yourself a buffer that reflects real production variance rather than theoretical ideal values.

Schema changes break the validation notebook. If a column is renamed or dropped in a source table, the null threshold check will throw a AnalysisException rather than a clean failure message. Wrap column-level checks in a try/except that records a structured failure:

try:
    check_null_thresholds_batch(runner, table_name, column_thresholds)
except Exception as e:
    runner._record(
        table_name=table_name,
        check_type="null_threshold",
        check_name=f"batch_null_check_{table_name}",
        expected=None, actual=None, threshold=None,
        passed=False,
        failure_message=f"Check execution error: {str(e)}"
    )

This connects directly to the schema evolution challenges covered in Handling Schema Evolution in Fabric Lakehouse Delta Tables.

---

Summary and Next Steps

You've built a production-grade data quality framework that does four things that matter: it catches silent data failures before they reach consumers, it creates a permanent audit trail in Delta that you can query and trend, it integrates cleanly with Fabric pipeline orchestration to halt execution on failure, and it's entirely owned by your team — no external services, no configuration files in a separate system, no black-box vendor logic.

The framework covers the three most practically valuable check categories: row count validation (protecting against data loss and explosions), null threshold enforcement (ensuring completeness of columns that drive business logic), and referential integrity (verifying that fact-to-dimension relationships will actually join in reports). Together, these catch the vast majority of real-world data quality failures.

Where to take this next:

• Add freshness checks: Extend the DataQualityRunner with a check that validates MAX(load_timestamp) on each table is within an expected window — a table that hasn't loaded in 36 hours when it should load every 24 is a pipeline failure worth catching explicitly.

• Add uniqueness checks: Duplicate primary keys are a common source of join inflation. Add a check that compares COUNT(*) to COUNT(DISTINCT key_column) and fails if they diverge.

• Build a Power BI quality dashboard: Connect your dq_check_results Delta table to a Power BI semantic model in Direct Lake mode. Executives don't need to read pipeline logs — they can see a quality pass-rate tile on their existing dashboard. The building a star schema in a Fabric lakehouse gold layer article shows how to structure Delta tables for exactly this kind of operational reporting.

• Add Delta time travel as a recovery mechanism: When a quality check fails and stops the gold write, you may need to restore the previous gold table state. Implementing Delta Lake Time Travel in a Fabric Lakehouse covers exactly how to roll back to a known-good version if a bad write did sneak through.

• Parameterize your check configurations: Rather than hardcoding thresholds in each notebook, store your check definitions — table names, columns, and thresholds — in a Delta configuration table. Your validation notebook reads the config and runs checks dynamically, making it trivial to add a new table to the quality framework without touching notebook code.

The measure of a quality framework isn't how elegant the code is — it's how many times it catches a real problem before a stakeholder does.