Dirty data in bronze silently corrupts silver and gold. Learn how to build a production-grade PySpark cleansing pipeline that removes exact and soft duplicates, handles nulls with intentional strategies, quarantines bad rows, and enforces data quality rules with exceptions — all structured around the medallion architecture in Microsoft Fabric.

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:
dropDuplicates, window functions, and keyed deduplication strategiesYou 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:
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.
Key insight
The most common mistake teams make is performing deduplication inside the bronze ingestion pipeline rather than in a dedicated silver transformation. This looks efficient but destroys your ability to diagnose data quality issues at the source. Keep bronze raw, always.
With that contract established, let's build a realistic dataset and work through the full cleansing pipeline.
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.
Note
In production, your bronze table would be written by a pipeline rather than manually. The write mode is always append — never overwrite bronze. If you need to understand the write mode tradeoffs in depth, the lesson on writing data from a Spark Notebook to a Fabric Lakehouse Delta Table covers append, overwrite, and merge patterns thoroughly.
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.
Warning
If you call df.distinct() or df.dropDuplicates() without a subset, Spark includes every column in the comparison, including your metadata columns. This means a legitimate duplicate with different _ingested_at timestamps will survive deduplication. Always specify the subset parameter using your business key columns.
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.Tip
When your source data has a meaningful updated_at or event_timestamp column, use that for ordering rather than _ingested_at. If the source system sends a correction (updated status), its updated_at will be later than the original record's, making it the natural "latest" winner even if both were ingested in the same pipeline run.
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.
Key insight
Filling a null with 0.0 and adding a boolean flag column is almost always better than simply dropping the row. Dropping loses the information that the order existed. A flagged zero allows downstream consumers to decide whether to include or exclude those records based on their specific use case.
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.
Warning
Never silently filter() out bad rows without logging them somewhere. Data that disappears without a trace is the most dangerous kind of data quality failure — you won't notice it until a stakeholder asks why their customer count dropped.
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")
Tip
If your silver layer receives incremental loads rather than full refreshes, whenMatchedUpdateAll() is usually what you want — it overwrites the entire matched row with the incoming version. But if you only want to update certain columns (e.g., never overwrite created_at), use whenMatchedUpdate(set={"col": "incoming.col"}) with explicit column mappings.
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.
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.
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.
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()
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 columnsrow_number() for keyed deduplication when you need to control which duplicate wins_imputed boolean column alongside any null-filling to preserve downstream optionalityWhere to go next:
Microsoft Fabric Fundamentals
Implementing Incremental Refresh for Direct Lake Semantic Models in Microsoft Fabric: Configuring Delta Table Partitioning, Framing Policies, and Triggering Refresh via the XMLA Endpoint
Implementing End-to-End Pipeline Error Handling in Microsoft Fabric: If Condition Activities, Failed Dependencies, and Email Alerts