Deduplicating and Cleansing Lakehouse Delta Tables with PySpark: Drop Duplicates, Fill Nulls, and Enforce Data Quality Rules Across Medallion Layers

Introduction

Picture this: your bronze lakehouse table is loaded with 90 days of customer order data ingested from a REST API. The pipeline ran successfully every night. But when your analyst runs a revenue report, the numbers are off by 12%. After an hour of digging, you find it: duplicate rows, because the API occasionally returns overlapping records on pagination boundaries, and your pipeline dutifully ingested every one of them. Add in a handful of NULL prices where the source system had a data entry gap, and some order statuses that arrived as blank strings instead of a meaningful value, and your "raw" data is quietly poisoning your silver and gold layers.

This is the central problem that data quality work in a medallion architecture solves. The bronze layer is supposed to be your exact landing zone — a faithful copy of source data, warts and all. But the moment you promote data to silver, you're making a promise: this data is fit for analysis. That promise requires active enforcement. You need to deduplicate aggressively, fill or flag nulls intentionally, and enforce business rules that your source systems can't or won't guarantee.

By the end of this lesson, you'll be able to write production-grade PySpark notebooks that clean delta tables at each medallion layer, handle duplicates with surgical precision, apply null-filling strategies appropriate to each column's semantics, and build reusable data quality checks that fail loudly when your data doesn't meet expectations.

What you'll learn:

• How to identify and remove duplicates using dropDuplicates, window functions, and keyed deduplication strategies
• Null-handling techniques: filling defaults, forward-filling, and flagging rather than fixing
• How to write data quality rules as reusable Python functions that raise exceptions on violations
• How to apply these patterns progressively across bronze, silver, and gold medallion layers
• How to use Delta Lake's MERGE to upsert cleansed data without creating additional duplicates downstream

---

Prerequisites

You should be comfortable writing PySpark DataFrames and understand how Delta tables work in a Fabric lakehouse. If you need a refresher on Spark notebooks in Fabric, the lesson on transforming data with Spark Notebooks in Microsoft Fabric covers the essentials. You should also understand the medallion architecture — if that's new, start with implementing the medallion architecture in Microsoft Fabric: Bronze, Silver, and Gold layers before continuing here.

You'll need:

• A Microsoft Fabric workspace with at least one lakehouse (bronze and silver lakehouses are ideal; gold optional)
• A Spark notebook attached to your lakehouse
• Basic familiarity with Delta Lake write modes (append, overwrite, merge)

---

Understanding What "Clean" Actually Means Across Medallion Layers

Before writing a single line of code, you need to be precise about what data quality means at each layer. A lot of teams apply the same cleaning logic everywhere and end up either destroying audit trails in bronze or tolerating dirty data in gold.

Here's the contract each layer should uphold:

Bronze: Land everything exactly as received. No deduplication, no null-filling. Add metadata columns (_ingested_at, _source_file) but don't alter source values. If the source sent a duplicate, bronze has that duplicate. This is your recovery point.

Silver: Apply structural cleansing. This is where deduplication happens, nulls are resolved or explicitly handled, types are cast correctly, and business rules are enforced. Silver is the layer your data analysts and data scientists should be able to trust.

Gold: Apply domain-specific aggregation and business logic. Any remaining data quality issues at this layer indicate a failure in silver. Gold tables typically don't need additional deduplication — they should receive only clean, keyed records from silver.

With that contract established, let's build a realistic dataset and work through the full cleansing pipeline.

---

Setting Up a Realistic Bronze Table

We'll work with a simulated e-commerce orders dataset. In practice, this would arrive via a data pipeline copy activity or a Dataflow Gen2 ingest, but for this lesson we'll create it directly in a notebook so you can follow along without any external dependencies.

from pyspark.sql import SparkSession
from pyspark.sql import functions as F
from pyspark.sql.types import (
    StructType, StructField, StringType, DoubleType,
    IntegerType, TimestampType
)
from datetime import datetime

spark = SparkSession.builder.getOrCreate()

# Simulate raw bronze data as it arrives from a source API
# Notice: duplicates, nulls, blank strings, and bad values are all intentional
raw_data = [
    ("ORD-1001", "C-201", "2024-03-01 08:15:00", "laptop",       1,  1299.99, "completed"),
    ("ORD-1002", "C-202", "2024-03-01 09:30:00", "headphones",   2,    89.95, "completed"),
    ("ORD-1003", "C-203", "2024-03-01 10:00:00", "webcam",       1,    55.00, "pending"),
    ("ORD-1001", "C-201", "2024-03-01 08:15:00", "laptop",       1,  1299.99, "completed"),  # exact duplicate
    ("ORD-1004", "C-204", "2024-03-01 11:00:00", "keyboard",     1,   None,   "completed"),  # null price
    ("ORD-1005", "C-205", "2024-03-01 11:30:00", "monitor",      2,   349.99, ""),           # blank status
    ("ORD-1006", "C-201", "2024-03-01 12:00:00", "mouse",        3,    29.99, "completed"),
    ("ORD-1002", "C-202", "2024-03-01 09:30:00", "headphones",   2,    89.95, "refunded"),   # same order_id, different status (soft dup)
    ("ORD-1007", "C-206", "2024-03-01 13:00:00", "usb hub",      1,    24.99, "completed"),
    ("ORD-1008", None,    "2024-03-01 14:00:00", "charger",      1,    19.99, "completed"),  # null customer
    ("ORD-1009", "C-207", "2024-03-01 14:30:00", "tablet",       1,   499.99, "completed"),
    ("ORD-1009", "C-207", "2024-03-01 14:30:00", "tablet",       1,   499.99, "completed"),  # exact duplicate
    ("ORD-1010", "C-208", "2024-03-01 15:00:00", "speaker",      2,    -5.00, "completed"),  # negative price (bad data)
]

schema = StructType([
    StructField("order_id",     StringType(),    True),
    StructField("customer_id",  StringType(),    True),
    StructField("order_ts",     StringType(),    True),  # arrives as string from API
    StructField("product",      StringType(),    True),
    StructField("quantity",     IntegerType(),   True),
    StructField("unit_price",   DoubleType(),    True),
    StructField("status",       StringType(),    True),
])

bronze_df = spark.createDataFrame(raw_data, schema=schema)

# Add standard bronze metadata columns
bronze_df = bronze_df.withColumn("_ingested_at", F.current_timestamp()) \
                     .withColumn("_source", F.lit("orders_api"))

# Write to bronze Delta table (append mode — bronze accumulates everything)
bronze_df.write.format("delta") \
    .mode("append") \
    .saveAsTable("bronze.orders_raw")

print(f"Bronze rows written: {bronze_df.count()}")

Run this cell and you'll have 13 rows in bronze — including two exact duplicates, one soft duplicate, two nulls, one blank string, and one negative price. This is a completely realistic bronze state.

---

Step 1: Removing Exact Duplicates with dropDuplicates

The simplest case is an exact duplicate: every column in row A is identical to every column in row B. PySpark's dropDuplicates() handles this cleanly.

# Read from bronze
df = spark.read.format("delta").table("bronze.orders_raw")

print(f"Total rows in bronze: {df.count()}")  # 13

# Remove exact duplicates across all source columns (exclude metadata cols)
source_cols = ["order_id", "customer_id", "order_ts", "product", "quantity", "unit_price", "status"]

df_deduped = df.dropDuplicates(subset=source_cols)

print(f"Rows after exact dedup: {df_deduped.count()}")  # 11 — two exact dupes removed

dropDuplicates(subset=...) is smarter than a blanket distinct() call. By specifying only source columns, you prevent the metadata columns (_ingested_at, _source) from being included in the comparison — which matters because two identical source records ingested at different times will have different _ingested_at values and distinct() would incorrectly keep both.

---

Step 2: Handling Soft Duplicates with Window Functions

Exact duplicates are easy. The harder case is the soft duplicate: ORD-1002 appears twice with the same order_id and order_ts but different status values ("completed" vs "refunded"). These are not identical rows, so dropDuplicates won't remove them. You need a strategy.

The most common approach for ordered event data is to keep the latest record per business key. This is where window functions come in.

from pyspark.sql.window import Window

# Define a window partitioned by our business key, ordered by ingestion time
# For source data that has a meaningful timestamp, use that instead of _ingested_at
window_spec = Window.partitionBy("order_id").orderBy(F.col("_ingested_at").desc())

df_ranked = df_deduped.withColumn("_row_rank", F.row_number().over(window_spec))

# Keep only rank 1 — the most recently ingested version of each order_id
df_keyed_dedup = df_ranked.filter(F.col("_row_rank") == 1).drop("_row_rank")

print(f"Rows after keyed dedup: {df_keyed_dedup.count()}")  # 10 — ORD-1002 soft dup resolved

Let's talk about why row_number() is the right choice here rather than rank() or dense_rank():

• row_number() always assigns a unique sequential number — even if two rows tie on the order column, one gets rank 1 and the other gets rank 2. This is what you want for deduplication: exactly one winner.
• rank() gives tied rows the same rank, meaning both would survive your filter. That defeats the purpose.
• dense_rank() has the same tying behavior as rank() for this use case.

---

Step 3: Null Handling — Fix, Flag, or Fail?

Not all nulls deserve the same treatment. Before you reach for fillna(), ask: why is this null, and what does null mean for this column?

Column	Null Scenario	Right Strategy
unit_price	Source system data entry error	Fill with product-level median, or flag for review
customer_id	Guest checkout (legitimate)	Fill with "GUEST" literal
status	Blank string from API	Normalize to "unknown"
quantity	Shouldn't be null per business rules	Raise a data quality exception

Let's implement all four patterns:

# --- Pattern 1: Fill null customer_id with GUEST literal ---
df_clean = df_keyed_dedup.withColumn(
    "customer_id",
    F.when(F.col("customer_id").isNull(), F.lit("GUEST"))
     .otherwise(F.col("customer_id"))
)

# --- Pattern 2: Normalize blank/empty status strings ---
# Empty string and null should be treated the same way
df_clean = df_clean.withColumn(
    "status",
    F.when(
        F.col("status").isNull() | (F.trim(F.col("status")) == ""),
        F.lit("unknown")
    ).otherwise(F.lower(F.trim(F.col("status"))))
)

# --- Pattern 3: Fill null prices with a sentinel and add a flag column ---
# We don't want to silently invent a price — flag it for downstream awareness
df_clean = df_clean.withColumn(
    "price_imputed",
    F.col("unit_price").isNull().cast("boolean")
).withColumn(
    "unit_price",
    F.when(F.col("unit_price").isNull(), F.lit(0.0))
     .otherwise(F.col("unit_price"))
)

# --- Pattern 4: Cast order_ts from string to timestamp ---
df_clean = df_clean.withColumn(
    "order_ts",
    F.to_timestamp(F.col("order_ts"), "yyyy-MM-dd HH:mm:ss")
)

# Verify
df_clean.select("order_id", "customer_id", "status", "unit_price", "price_imputed", "order_ts").show(15, truncate=False)

The price_imputed flag column is a powerful pattern. Rather than silently setting a price to 0 or some computed average, you're being explicit: this value was not in the source data. Downstream consumers — including Power BI reports and gold-layer aggregations — can filter out imputed records if they need true revenue figures, or include them if they just need order counts. You've preserved optionality.

---

Step 4: Enforcing Business Rules with Data Quality Functions

Now for the most important part: rules that fail loudly. Silently swallowing bad data is how you get wrong numbers in reports. You want your notebook to raise an exception — and thereby fail the pipeline — when data violates rules that should never be violated.

def enforce_data_quality(df, table_name: str) -> None:
    """
    Runs a suite of data quality checks on a DataFrame.
    Raises a ValueError with a detailed report if any check fails.
    """
    failures = []
    total_rows = df.count()

    # Rule 1: No null order_ids
    null_order_ids = df.filter(F.col("order_id").isNull()).count()
    if null_order_ids > 0:
        failures.append(f"CRITICAL: {null_order_ids} rows have null order_id")

    # Rule 2: No null quantities
    null_quantities = df.filter(F.col("quantity").isNull()).count()
    if null_quantities > 0:
        failures.append(f"CRITICAL: {null_quantities} rows have null quantity")

    # Rule 3: No negative prices (imputed zeros are OK, negatives are not)
    negative_prices = df.filter(
        (F.col("unit_price") < 0) & (F.col("price_imputed") == False)
    ).count()
    if negative_prices > 0:
        failures.append(f"CRITICAL: {negative_prices} rows have negative unit_price")

    # Rule 4: Status must be in allowed set
    allowed_statuses = ["completed", "pending", "refunded", "cancelled", "unknown"]
    invalid_statuses = df.filter(
        ~F.col("status").isin(allowed_statuses)
    ).count()
    if invalid_statuses > 0:
        failures.append(f"WARNING: {invalid_statuses} rows have unexpected status values")

    # Rule 5: Quantity must be positive
    non_positive_qty = df.filter(F.col("quantity") <= 0).count()
    if non_positive_qty > 0:
        failures.append(f"CRITICAL: {non_positive_qty} rows have quantity <= 0")

    # Rule 6: Completeness threshold — no more than 5% of prices should be imputed
    imputed_pct = (df.filter(F.col("price_imputed") == True).count() / total_rows) * 100
    if imputed_pct > 5.0:
        failures.append(
            f"WARNING: {imputed_pct:.1f}% of rows have imputed prices (threshold: 5%)"
        )

    # Report results
    critical_failures = [f for f in failures if f.startswith("CRITICAL")]
    warnings = [f for f in failures if f.startswith("WARNING")]

    print(f"\n=== Data Quality Report: {table_name} ===")
    print(f"Total rows evaluated: {total_rows}")

    if warnings:
        for w in warnings:
            print(f"  ⚠️  {w}")

    if critical_failures:
        for c in critical_failures:
            print(f"  ❌ {c}")
        raise ValueError(
            f"Data quality check FAILED for {table_name}. "
            f"{len(critical_failures)} critical rule(s) violated. "
            f"Review the report above before promoting to silver."
        )

    print(f"  ✅ All critical checks passed ({len(warnings)} warning(s))")

Now apply it:

# This will raise a ValueError because ORD-1010 has a negative price
try:
    enforce_data_quality(df_clean, "silver.orders_clean")
except ValueError as e:
    print(f"\nPipeline halted: {e}")

You'll see the report fire on the negative price in ORD-1010. This is the right behavior — you want to know about it before it reaches silver. Now fix it:

# Quarantine rows that fail critical rules rather than silently dropping them
df_quarantine = df_clean.filter(
    (F.col("unit_price") < 0) & (F.col("price_imputed") == False)
).withColumn("_quarantine_reason", F.lit("negative_unit_price")) \
 .withColumn("_quarantined_at", F.current_timestamp())

# Write bad rows to a quarantine table for investigation
df_quarantine.write.format("delta") \
    .mode("append") \
    .saveAsTable("silver.orders_quarantine")

# Exclude quarantined rows from the clean set
df_silver = df_clean.filter(
    ~((F.col("unit_price") < 0) & (F.col("price_imputed") == False))
)

# Re-run quality check — should pass now
enforce_data_quality(df_silver, "silver.orders_clean")

The quarantine table is critical. Rather than dropping bad rows (which loses information) or keeping them in the main table (which corrupts your analysis), you park them in a dedicated table with a reason column. This lets your data engineering team investigate, fix, and re-ingest.

---

Step 5: Writing Clean Data to Silver with MERGE

Now that df_silver is deduplicated and validated, you need to write it to the silver table. The correct pattern is an upsert (MERGE), not an append. If you use append, re-running your notebook will create more duplicates in silver — exactly what you were trying to prevent.

from delta.tables import DeltaTable

silver_table_path = "Tables/silver/orders_clean"

# Create the silver table if it doesn't exist yet
df_silver.write.format("delta") \
    .mode("overwrite") \
    .option("overwriteSchema", "true") \
    .saveAsTable("silver.orders_clean")

print("Silver table initialized.")

For subsequent runs, use MERGE:

def upsert_to_silver(df_new, silver_table_name: str, key_col: str) -> None:
    """
    Upserts a cleaned DataFrame into a silver Delta table.
    Matches on key_col and updates all columns if source differs.
    """
    silver_table = DeltaTable.forName(spark, silver_table_name)

    silver_table.alias("silver").merge(
        df_new.alias("incoming"),
        condition=f"silver.{key_col} = incoming.{key_col}"
    ).whenMatchedUpdateAll() \
     .whenNotMatchedInsertAll() \
     .execute()

    print(f"Upsert complete into {silver_table_name}")

# On subsequent runs, use this instead of the initial overwrite
# upsert_to_silver(df_silver, "silver.orders_clean", "order_id")

For a deeper dive into MERGE patterns — including how to handle slowly changing dimensions — the lesson on implementing slowly changing dimensions in a Fabric lakehouse using PySpark and Delta Lake MERGE covers the full SCD Type 1 and Type 2 patterns.

---

Step 6: Promoting to Gold — Aggregation with Quality Guarantees

The gold layer is where you build the tables that Power BI reports, analysts, and ML pipelines consume directly. At this layer, you're not cleansing anymore — you're aggregating. But you can still embed quality assertions to catch problems that slipped through silver.

# Read from silver
df_silver_read = spark.read.format("delta").table("silver.orders_clean")

# Build a gold-layer daily revenue summary
# Exclude imputed-price orders from revenue figures (include them in order counts)
df_gold = df_silver_read.groupBy(
    F.to_date(F.col("order_ts")).alias("order_date"),
    "status"
).agg(
    F.count("order_id").alias("total_orders"),
    F.sum(
        F.when(F.col("price_imputed") == False, F.col("unit_price") * F.col("quantity"))
         .otherwise(F.lit(0.0))
    ).alias("confirmed_revenue"),
    F.sum(F.col("unit_price") * F.col("quantity")).alias("total_revenue_incl_imputed"),
    F.countDistinct("customer_id").alias("unique_customers")
)

# Gold-level assertion: confirmed revenue should always be <= total revenue
assertion_failed = df_gold.filter(
    F.col("confirmed_revenue") > F.col("total_revenue_incl_imputed")
).count()

if assertion_failed > 0:
    raise ValueError("Gold assertion failed: confirmed_revenue exceeds total_revenue_incl_imputed")

# Write gold table (full overwrite is typical for aggregated gold tables)
df_gold.write.format("delta") \
    .mode("overwrite") \
    .saveAsTable("gold.daily_order_revenue")

print("Gold table written successfully.")
df_gold.show()

Notice the inline assertion between the transformation and the write. This is your last line of defense. A gold assertion failure means something unexpected happened in the math — maybe a null slipped through, maybe there's a sign error in a calculation. Failing here prevents a bad number from landing in a Power BI report.

This table is now ready to serve as a source for Direct Lake reporting. As discussed in the Direct Lake mode in Power BI article, Power BI reads these Delta tables directly without an import step — which means any issue in your gold table shows up immediately in reports. That's why the quality gates matter so much.

---

Hands-On Exercise: Build a Reusable Cleansing Pipeline

Now pull everything together. Your task is to build a self-contained notebook that accepts a bronze table name and outputs a clean silver table, parameterized so it can be called from a data pipeline.

Setup: Add this cell at the top of your notebook to accept pipeline parameters:

# Notebook parameters — these can be overridden by a Fabric Data Pipeline
# Use the Fabric notebook "parameters" cell tag for pipeline integration
dbutils.widgets.text("bronze_table", "bronze.orders_raw")
dbutils.widgets.text("silver_table", "silver.orders_clean")
dbutils.widgets.text("quarantine_table", "silver.orders_quarantine")
dbutils.widgets.text("key_column", "order_id")

bronze_table     = dbutils.widgets.get("bronze_table")
silver_table     = dbutils.widgets.get("silver_table")
quarantine_table = dbutils.widgets.get("quarantine_table")
key_column       = dbutils.widgets.get("key_column")

Full pipeline cell:

def run_silver_cleansing_pipeline(
    bronze_table: str,
    silver_table: str,
    quarantine_table: str,
    key_column: str
):
    print(f"Starting cleansing pipeline: {bronze_table} → {silver_table}")

    # 1. Read bronze
    df = spark.read.format("delta").table(bronze_table)
    print(f"  Bronze rows: {df.count()}")

    # 2. Exact dedup
    source_cols = [c for c in df.columns if not c.startswith("_")]
    df = df.dropDuplicates(subset=source_cols)

    # 3. Keyed dedup (keep latest per business key)
    window_spec = Window.partitionBy(key_column).orderBy(F.col("_ingested_at").desc())
    df = df.withColumn("_row_rank", F.row_number().over(window_spec)) \
           .filter(F.col("_row_rank") == 1) \
           .drop("_row_rank")

    print(f"  After dedup: {df.count()}")

    # 4. Null handling
    df = df.withColumn(
        "customer_id",
        F.when(F.col("customer_id").isNull(), F.lit("GUEST")).otherwise(F.col("customer_id"))
    ).withColumn(
        "status",
        F.when(F.col("status").isNull() | (F.trim(F.col("status")) == ""), F.lit("unknown"))
         .otherwise(F.lower(F.trim(F.col("status"))))
    ).withColumn(
        "price_imputed",
        F.col("unit_price").isNull().cast("boolean")
    ).withColumn(
        "unit_price",
        F.when(F.col("unit_price").isNull(), F.lit(0.0)).otherwise(F.col("unit_price"))
    ).withColumn(
        "order_ts",
        F.to_timestamp(F.col("order_ts"), "yyyy-MM-dd HH:mm:ss")
    )

    # 5. Quarantine bad rows
    bad_rows = df.filter((F.col("unit_price") < 0) & (F.col("price_imputed") == False)) \
                 .withColumn("_quarantine_reason", F.lit("negative_unit_price")) \
                 .withColumn("_quarantined_at", F.current_timestamp())

    quarantine_count = bad_rows.count()
    if quarantine_count > 0:
        bad_rows.write.format("delta").mode("append").saveAsTable(quarantine_table)
        print(f"  Quarantined {quarantine_count} rows → {quarantine_table}")

    df_clean = df.filter(
        ~((F.col("unit_price") < 0) & (F.col("price_imputed") == False))
    )

    # 6. Data quality gate
    enforce_data_quality(df_clean, silver_table)

    # 7. Upsert to silver
    try:
        upsert_to_silver(df_clean, silver_table, key_column)
    except Exception:
        # Table doesn't exist yet — initial load
        df_clean.write.format("delta").mode("overwrite") \
            .option("overwriteSchema", "true") \
            .saveAsTable(silver_table)
        print(f"  Silver table created: {silver_table}")

    print(f"Pipeline complete. Clean rows in silver: {df_clean.count()}")

# Run it
run_silver_cleansing_pipeline(
    bronze_table=bronze_table,
    silver_table=silver_table,
    quarantine_table=quarantine_table,
    key_column=key_column
)

Once this notebook works end-to-end, wire it into a Fabric Data Pipeline as a Notebook activity. The pipeline passes parameters into the notebook, so the same notebook handles cleansing for any table in your bronze layer — just pass different table names. The lesson on using notebook variables and parameters in Microsoft Fabric explains the exact parameter-passing mechanism in detail.

---

Common Mistakes & Troubleshooting

Mistake 1: Using distinct() instead of dropDuplicates(subset=...)

distinct() compares every column. With metadata columns like _ingested_at, duplicates with different ingestion timestamps survive. Always use dropDuplicates(subset=source_columns).

Mistake 2: Deduplicating before partitioning

If you call dropDuplicates on a very large DataFrame without an appropriate orderBy, Spark chooses which duplicate to keep arbitrarily. Combine it with a window function rank pattern when you need to control which record survives.

Mistake 3: Writing silver with mode("append") in a re-runnable notebook

This is the most common cause of downstream duplicate issues. The notebook runs twice (maybe due to a retry), and silver has double the rows. Always use MERGE for idempotent silver writes.

Mistake 4: Running df.count() repeatedly on the same DataFrame

Each count() triggers a full Spark job. In a production pipeline with many quality checks, this adds significant runtime. Cache the DataFrame first:

df_clean.cache()
df_clean.count()  # triggers caching
# now subsequent count() calls are fast
# ...
df_clean.unpersist()  # release when done

Mistake 5: Forgetting to handle schema evolution

When your source adds a new column, your dropDuplicates(subset=source_cols) list is out of date. Build your source column list dynamically:

source_cols = [c for c in df.columns if not c.startswith("_")]

This automatically includes new columns without requiring notebook changes. The broader topic of handling schema evolution in Fabric Lakehouse Delta Tables covers more advanced scenarios.

Debugging tip: Check the quarantine table first

When your pipeline fails data quality checks, your first stop should be the quarantine table — not the bronze table. The quarantine table tells you exactly which rows failed and why, which narrows your investigation dramatically.

# Quick diagnostic query
spark.read.format("delta").table("silver.orders_quarantine") \
    .groupBy("_quarantine_reason") \
    .count() \
    .orderBy(F.col("count").desc()) \
    .show()

---

Summary & Next Steps

You've built a complete data quality pipeline that handles the full spectrum of real-world data problems: exact duplicates, soft duplicates with business key conflicts, null values with semantically appropriate resolutions, blank strings, type mismatches, and business rule violations. More importantly, you've structured it around the medallion contract — bronze stays raw, silver is clean and trusted, gold is aggregated and assertion-protected.

The key patterns to take away:

• dropDuplicates(subset=source_cols) for exact deduplication, always excluding metadata columns
• Window function with row_number() for keyed deduplication when you need to control which duplicate wins
• Flag + fill, don't just fill — add a _imputed boolean column alongside any null-filling to preserve downstream optionality
• Quarantine, don't drop — bad rows go to a dedicated table with a reason column, never silently discarded
• MERGE for idempotent silver writes — re-runnable notebooks that don't create duplicates on retry
• Inline assertions at gold — your last quality gate before data reaches reports

Where to go next:

• To understand how these clean Delta tables translate into faster query performance, read about optimizing Delta Table performance in a Fabric Lakehouse — V-Order, OPTIMIZE, VACUUM, and Z-Order work on top of the clean tables you've built here.
• If you need to recover from a bad write to silver (a quality gate that failed after the write, for example), Delta Lake time travel in a Fabric Lakehouse shows you how to roll back.
• If your silver tables feed a Power BI semantic model, the guide on connecting a Power BI semantic model to a Fabric Lakehouse in Direct Lake mode explains what your Delta table needs to look like for optimal Direct Lake performance.