Momentum One Academy · article

Data Cleaning and Transformation with PySpark in Fabric

A production data pipeline does not become trustworthy because a notebook ran without an error. It becomes trustworthy when the team can explain what arrived, what changed, what was rejected, and what a downstream analyst can safely use. That is the practical problem behind Module 4 — Data Cleaning & Transformation with Spark.

In a Fabric delivery, PySpark is useful when the source is large, semi-structured, inconsistent, or expected to grow beyond a comfortable single-machine transformation. The goal is not to make every dataset “perfect.” The goal is to preserve evidence, apply explicit rules, produce a repeatable Delta table, and make failures visible before they reach a report.

Start with the problem, not the notebook

Operational data commonly arrives with blank identifiers, inconsistent casing, mixed date formats, duplicate events, invalid numbers, late updates, and schema drift. A quick cleanup that trims a few columns may make a demo look good while hiding the harder questions: Was a duplicate removed arbitrarily? Did a rerun append the same batch twice? Were rejected rows lost? Can the business explain why a sale is missing?

Momentum One treats cleaning as a controlled boundary between ingestion and consumption. The raw input is retained as evidence. The cleaned output has a documented schema and quality rules. Rows that cannot pass those rules are retained in a quarantine area with a reason, rather than silently discarded. This makes the pipeline debuggable and gives an analyst a clear contract.

The Fabric concepts that matter

A Fabric Lakehouse combines data-lake scale with warehouse-style querying. It stores structured and unstructured data in one place, uses Delta Lake for tables, and supports both Apache Spark and SQL access. The Lakehouse explorer separates Tables from Files: raw files can be landed in Files, while managed Delta tables belong in Tables. Fabric also provides a SQL analytics endpoint for read-only T-SQL queries over Delta tables. 1

That combination changes the design conversation. A PySpark notebook can read a JSON or Parquet landing area, apply typed transformations, and write a Delta table. A later Spark SQL query can inspect the same table during development, while the SQL analytics endpoint and Power BI can consume curated Delta data. The endpoint is not a second write engine for the cleaning job, and non-Delta files do not appear there as queryable tables. 1

Delta Lake is the default table format in a Fabric Lakehouse. It adds transaction logs and table behavior around Parquet, so the team gets transactional consistency, schema-related protections, and a history-aware table format rather than a directory of unrelated files. Microsoft describes Delta as the standard table format across Fabric experiences, but its feature support differs between engines and runtimes. 2 3

For production, this means we should write clean outputs as Delta tables and be deliberate when another engine is involved. If a table uses a Delta feature that a particular reader does not support, a “successful” write can still produce a downstream compatibility problem. Microsoft’s current interoperability guidance specifically recommends using a PySpark notebook when a Python notebook’s Delta reader or writer cannot support the table features required by the workload. 3

Use medallion layers as quality boundaries

Medallion architecture is a useful operating model because each layer has a different promise, not because three folders are automatically better than one. Microsoft defines the layers as bronze for raw data, silver for enriched and validated data, and gold for curated data. The intent is incremental improvement while preserving the original source of truth. 3

A practical Fabric implementation looks like this:

Bronze: land the source with minimal alteration. Keep source fields, arrival metadata, and enough context to replay the load. If a source is already governed in OneLake or another supported store, a shortcut may avoid an unnecessary copy; if the source is unstable or needs an ingestion snapshot, copy it instead.
Silver: standardize names and types, normalize timestamps and codes, enforce required-field rules, remove or isolate duplicates, and record rejected rows. Silver is where a data contract becomes executable.
Gold: shape data for a business question. This may be a fact table, a dimension, a daily metric, or a serving table with carefully chosen grain. Gold should not be the first place where basic type repair happens.

This is also the shape used in Microsoft’s end-to-end Lakehouse tutorial: raw bronze data is transformed into validated and deduplicated silver data, then refined for consumption in gold. The tutorial is useful as a workflow reference, but it explicitly is not a complete reference architecture, so production teams still need to add their own contracts, controls, and operating model. 8

The layer boundary also affects ownership. Microsoft documents two common deployment choices: a lakehouse for every layer, or lakehouses for bronze and silver with a warehouse for gold. It recommends considering separate workspaces for stronger layer-level governance. That is a design decision, not a mandatory topology. A small workshop may use one workspace to keep the exercise understandable; a production estate should align workspace boundaries with access, release, and support responsibilities. 3

Keep bronze close to the original format when that helps traceability. For silver and gold, prefer Delta tables so Spark, SQL, pipelines, and BI can work from a consistent table representation. Avoid adding partitions or clustering by habit. Choose layout and maintenance settings from measured query patterns, table size, and the consuming engine; high-cardinality partitioning can create more file and metadata work instead of improving reads. 2 3

A production-shaped PySpark pattern

The following example assumes a Fabric notebook has a default Lakehouse attached and that JSON order files are in Files/bronze/orders. It uses a static schema, built-in Spark functions, an explicit reject reason, and a window to select one record per order key. Static schemas are important for large JSON inputs because full schema inference can scan the dataset and consume substantial memory. 5

from pyspark.sql import functions as F
from pyspark.sql.types import (
    StructType, StructField, StringType
)
from pyspark.sql.window import Window

order_schema = StructType([
    StructField("order_id", StringType(), True),
    StructField("customer_id", StringType(), True),
    StructField("order_ts", StringType(), True),
    StructField("amount", StringType(), True),
    StructField("status", StringType(), True),
    StructField("source_updated_at", StringType(), True),
])

raw = (
    spark.read
        .format("json")
        .schema(order_schema)
        .load("Files/bronze/orders")
)

typed = (
    raw
      .withColumn("order_id", F.trim("order_id"))
      .withColumn("customer_id", F.trim("customer_id"))
      .withColumn(
          "order_ts",
          F.to_timestamp("order_ts", "yyyy-MM-dd'T'HH:mm:ssXXX")
      )
      .withColumn("amount", F.col("amount").cast("decimal(18,2)"))
      .withColumn("status", F.upper(F.trim("status")))
      .withColumn(
          "source_updated_at",
          F.to_timestamp("source_updated_at", "yyyy-MM-dd'T'HH:mm:ssXXX")
      )
      .withColumn("ingested_at", F.current_timestamp())
      .withColumn(
          "record_hash",
          F.sha2(
              F.to_json(F.struct(
                  "order_id", "customer_id", "order_ts", "amount",
                  "status", "source_updated_at"
              )),
              256
          )
      )
)

rule = (
    F.when(F.col("order_id").isNull() | (F.col("order_id") == ""),
           F.lit("missing order_id"))
     .when(F.col("order_ts").isNull(), F.lit("invalid order_ts"))
     .when(F.col("amount").isNull() | (F.col("amount") < 0),
           F.lit("invalid amount"))
     .when(F.col("status").isNull() |
           ~F.col("status").isin("NEW", "PAID", "CANCELLED"),
           F.lit("unknown status"))
)

checked = typed.withColumn("reject_reason", rule)
rejected = checked.filter(F.col("reject_reason").isNotNull())
accepted = checked.filter(F.col("reject_reason").isNull()).drop("reject_reason")

latest = Window.partitionBy("order_id").orderBy(
    F.col("source_updated_at").desc_nulls_last(),
    F.col("record_hash").asc()
)

silver_orders = (
    accepted
      .withColumn("row_number", F.row_number().over(latest))
      .filter(F.col("row_number") == 1)
      .drop("row_number", "record_hash")
)

rejected.write.format("delta").mode("append").saveAsTable("quarantine_orders")
silver_orders.write.format("delta").mode("append").saveAsTable("silver_orders")

This is deliberately ordinary PySpark. trim, to_timestamp, cast, when, filter, and row_number are built-in DataFrame operations that Spark can optimize. The deduplication order should use a source sequence or another deterministic tie-breaker when the source can produce equal timestamps; do not assume that “latest” is meaningful unless the source supplies an ordering rule. In an incremental pipeline, also define how a batch is identified and how a rerun is made idempotent. Blindly appending the same input twice is not a cleaning strategy.

The example writes both accepted and rejected data as Delta tables in the attached default Lakehouse. In a real project, add a batch identifier, source file or event metadata, row counts, and a data-quality summary to an operational table. Keep sensitive payloads out of the quarantine table if they are not needed for diagnosis. The notebook should fail when the input cannot be read or the contract is structurally broken; it should not necessarily fail the whole batch because ten rows have a known business-rule violation.

Choose Spark, Spark SQL, and KQL intentionally

Use PySpark when the transformation benefits from Python control flow, reusable functions, schema handling, or multi-step DataFrame logic. Use Spark SQL when the rule is clearer as a relational expression and the team wants analysts and engineers to review the same SQL. Fabric notebooks support PySpark, Spark SQL, Scala, and SparkR cells, so a single notebook can mix languages when that improves readability. 6

Do not default to an RDD or a Python UDF for work that built-in DataFrame functions can express. Microsoft’s Spark guidance recommends DataFrames because Catalyst and the Tungsten execution engine can optimize them. It also recommends filtering and selecting columns before joins, using adaptive query execution where appropriate, and considering a broadcast join for a genuinely small lookup table. Row-by-row Python UDFs add serialization overhead; use a built-in expression, a vectorized Pandas UDF, or a JVM-language UDF only when the logic truly requires it. 5

KQL has a different center of gravity in Fabric: the Lakehouse experience exposes an Eventhouse endpoint for high-performance real-time analytics. It can complement a batch-cleaning design, but it is not a reason to force streaming or event analytics into a PySpark silver transformation. Select the engine from the workload and latency requirement rather than from the availability of a query language. 1

Production trade-offs and failure modes

Schema inference versus an explicit contract. Inference is convenient for exploration, but it can be slow on large JSON or XML inputs and can infer an unintended type from a bad sample. Start with a representative sample during discovery, then persist and review the schema used by scheduled runs. Treat an unexpected new column or incompatible type as a controlled change, not a silent cast.

Append versus upsert. Append is straightforward for immutable events. It is unsafe for a source that sends corrections or repeats a delivery. Define the natural key, update ordering, and late-arrival policy before implementing an upsert. Delta supports MERGE, but the key and source semantics still belong to the project team; a merge with a non-unique source can update the same target row unpredictably. 3

Large joins and skew. A job can be logically correct and operationally unusable. Watch for hot keys, oversized shuffles, and a small number of tasks that run much longer than the rest. Filter early, project only needed columns, broadcast only small stable dimensions, and use adaptive query execution. Avoid collect(), countByKey(), or large toPandas() calls that pull data into the driver. 5

Small files and uncontrolled history. Many tiny writes increase metadata and file-management work. Batch writes where possible, inspect table health, and use the supported Fabric maintenance approach for the workload. Delta history and deleted-file retention also have storage implications, so retention should reflect recovery and audit needs rather than be copied from a sample notebook. 2 3

“It worked in my session.” Starter pools are a best-effort fast-start option, not a promise that every production run starts in seconds. Custom Spark pools provide more control over node sizing and scaling. Library installation and Spark properties can also change startup behavior. For repeatable jobs, manage dependencies through a Fabric environment and test the same runtime and configuration used by the scheduled path. 4 6

The wrong identity ran the job. Fabric notebook execution has different security contexts. A pipeline-triggered notebook runs under the pipeline’s last modified user, while a scheduled run uses the user who created or last updated the schedule. If a job can read interactively but fails in automation, inspect identity and permissions before changing transformation code. 6

Operational checklist

Before promoting a PySpark cleaning job, the Momentum One delivery review asks:

Is the bronze input replayable, and are source location, arrival time, and batch identifiers retained?
Is the silver schema explicit, documented, and tested against nulls, malformed values, and expected drift?
Does every rejected row have a reason and a defined retention or review path?
Is deduplication based on a documented key and deterministic ordering?
Can a rerun avoid duplicating a previously accepted batch?
Are transformations expressed with DataFrame functions before reaching for UDFs?
Have join sizes, skew, driver actions, and output file patterns been checked on representative volume?
Are Delta tables used where downstream Spark, SQL, or BI consumers require them?
Are the runtime, environment libraries, notebook parameters, and target Lakehouse explicit?
Has the automated identity been tested with least-privilege access?
Is there a run monitor, an error log, and a row-count or quality signal that an operator can act on?

Fabric’s Monitor hub provides a central place to browse Spark activities, while notebook contextual monitoring exposes tasks, executors, logs, and execution details during development. Use those views to turn “the pipeline is slow” into a concrete observation such as a skewed stage, an executor memory failure, or an unexpectedly large scan. 7

Workshop connection

In Module 4, the useful outcome is not memorizing withColumn. It is being able to defend a transformation boundary. Build a bronze input with a few deliberately bad orders, write a silver table and a quarantine table, rerun the same batch, and explain what prevents duplicate business records. Then inspect the Spark run, change the input schema, and trace the failure or reject path. Finally, compare the same rule in PySpark and Spark SQL and decide which version a production reviewer can maintain.

That exercise mirrors a Momentum One production conversation: what is the data contract, what is the recovery path, who owns the layer, and how will we know tomorrow’s run is still healthy?

Ready to turn the pattern into a working Fabric delivery? Explore the Momentum One Academy Fabric Production Workshop.

References

Plan your workshop