Transforming Data with Spark Notebooks in Microsoft Fabric: PySpark for Lakehouse Tables

Introduction

You've landed raw sales data in your lakehouse — CSV files from three regional systems, each with slightly different column names, inconsistent date formats, and a handful of duplicate records that your source system helpfully generated during a migration. A Dataflow Gen2 can handle some of this, but you need something that can join across large datasets, apply conditional business logic, write partitioned Delta tables, and do it all fast enough to run daily. That's the moment Spark notebooks stop being a "nice to have" and become the right tool.

Spark notebooks in Microsoft Fabric give you a Jupyter-style interactive environment running on Apache Spark, backed by OneLake storage and integrated directly with your lakehouse. You write Python (PySpark) — or SQL, Scala, or R if you prefer — and Spark handles distributed execution across the cluster. The notebook reads and writes Delta tables natively, the results show up instantly in your lakehouse, and Power BI's Direct Lake mode can hit those tables the moment you're done. No ETL pipelines with arcane connectors, no export-and-reimport dance. Just code, transformation, Delta.

By the end of this lesson, you'll have built a working transformation notebook that ingests raw files, cleans and enriches the data, writes partitioned Delta tables to a lakehouse, and follows patterns you can reuse on real production workloads.

What you'll learn:

• How to create and configure a Spark notebook in Fabric and attach it to a lakehouse
• How to read raw files from the Files section and Delta tables from the Tables section of your lakehouse
• Core PySpark DataFrame operations for cleaning, reshaping, and enriching data
• How to write transformed data back as managed Delta tables with partitioning
• How to use %sql magic cells for quick inline SQL on your Spark DataFrames
• How to parameterize notebooks so they can be called from Data Pipelines

---

Prerequisites

This lesson assumes you're comfortable with:

• Basic Python syntax and the concept of DataFrames
• The Microsoft Fabric workspace and lakehouse structure — if you need a refresher, see Building Your First Lakehouse in Microsoft Fabric: Files, Tables, and the SQL Analytics Endpoint
• Why you'd choose a lakehouse over a warehouse for this kind of workload — covered in Fabric Lakehouse vs Warehouse: Choosing the Right Store for Your Workload
• Basic familiarity with Delta Lake concepts and OneLake storage — see OneLake Explained: One Copy of Data, Delta Tables, and Shortcuts

You'll need a Fabric workspace with at least Contributor access, an existing lakehouse, and some raw files uploaded to the Files section. We'll work with a realistic sales dataset throughout.

---

Understanding the Spark Environment in Fabric

Before you write a single line of PySpark, it helps to understand what's actually happening when you run a notebook in Fabric.

When you open a notebook and hit Run, Fabric spins up a Spark session against a managed cluster. That cluster's size is determined by your Fabric capacity (F SKU) — the larger the capacity, the more executor nodes and memory available. For development and moderate data volumes, Fabric's default starter pool (which spins up in about 30-45 seconds) is more than sufficient. For large-scale production jobs, you can configure custom Spark pool settings under the workspace settings.

The key architectural point: Fabric notebooks use the Lakehouse as the default storage layer. When you attach a lakehouse to a notebook, two root paths become available:

• Files/ — the unmanaged file zone where raw data lives (CSV, Parquet, JSON, etc.)
• Tables/ — the managed Delta table zone where structured tables live

Spark in Fabric can read and write both, but only data in Tables/ shows up in the SQL Analytics Endpoint and is accessible to Power BI Direct Lake. This distinction matters enormously in practice: your raw ingested files live in Files/, your transformation output goes to Tables/.

---

Creating a Notebook and Attaching a Lakehouse

Navigate to your Fabric workspace and select New item → Notebook. A blank notebook opens with a single Python cell. The interface should look familiar if you've used Jupyter before — cells, a toolbar, and a kernel status indicator in the top right.

The first thing to do is attach your lakehouse. In the left-hand panel of the notebook, you'll see an Explorer section. Click Add lakehouse, choose your existing lakehouse from the list, and click Add. Once attached, you'll see the lakehouse tree in the left panel — the Files and Tables folders are visible and browsable directly.

You can also set a default lakehouse. When a lakehouse is set as default, PySpark can reference its paths using Files/ and Tables/ relative paths, and the spark.read.table("tablename") shortcut resolves against that lakehouse's metastore. If you're working with multiple lakehouses (say, a raw zone and a curated zone), you'll manage the paths explicitly — more on that shortly.

---

Reading Raw Data from the Files Zone

Let's set the scene. Your raw data is three CSV files sitting in Files/raw/sales/:

• apac_sales_2024.csv
• emea_sales_2024.csv
• amer_sales_2024.csv

Each file has slightly different headers (because of course they do). APAC uses sale_date, EMEA uses transaction_date, and AMER uses date. All three need to land in a unified silver_sales Delta table.

Start by reading all three files, handling the schema differences:

from pyspark.sql import functions as F
from pyspark.sql.types import StructType, StructField, StringType, DoubleType, IntegerType

# Read APAC sales
apac = spark.read.option("header", True).option("inferSchema", True).csv(
    "Files/raw/sales/apac_sales_2024.csv"
)

# Read EMEA sales — rename the date column to match our standard
emea = spark.read.option("header", True).option("inferSchema", True).csv(
    "Files/raw/sales/emea_sales_2024.csv"
).withColumnRenamed("transaction_date", "sale_date")

# Read AMER sales — same rename
amer = spark.read.option("header", True).option("inferSchema", True).csv(
    "Files/raw/sales/amer_sales_2024.csv"
).withColumnRenamed("date", "sale_date")

# Tag each with its region before combining
apac = apac.withColumn("region", F.lit("APAC"))
emea = emea.withColumn("region", F.lit("EMEA"))
amer = amer.withColumn("region", F.lit("AMER"))

# Union all three
raw_sales = apac.unionByName(emea, allowMissingColumns=True) \
                .unionByName(amer, allowMissingColumns=True)

print(f"Total raw records: {raw_sales.count()}")
raw_sales.printSchema()

Notice unionByName with allowMissingColumns=True. If one region's CSV has a column the others don't (say EMEA has a vat_amount column), PySpark fills nulls for the missing values rather than throwing an error. This is far more robust than positional union(), which would silently misalign columns if the CSVs have different column orders.

---

Core Data Cleaning Operations

Raw data is almost never clean. Let's work through the transformations you'll apply to the unified raw_sales DataFrame.

Standardizing Date Formats

The sale_date column is currently a string, and its values might be 2024-01-15, 15/01/2024, or Jan 15 2024 depending on the region. Convert everything to a proper date type:

from pyspark.sql.functions import to_date, coalesce, col, trim, upper, when, regexp_replace

# Try multiple date formats and coalesce to the first successful parse
cleaned_sales = raw_sales.withColumn(
    "sale_date_parsed",
    coalesce(
        to_date(col("sale_date"), "yyyy-MM-dd"),
        to_date(col("sale_date"), "dd/MM/yyyy"),
        to_date(col("sale_date"), "MMM dd yyyy")
    )
)

# Flag records where date parsing failed
cleaned_sales = cleaned_sales.withColumn(
    "date_parse_failed",
    col("sale_date_parsed").isNull() & col("sale_date").isNotNull()
)

# Check how many records have unparseable dates
failed_dates = cleaned_sales.filter(col("date_parse_failed")).count()
print(f"Records with unparseable dates: {failed_dates}")

This pattern — try multiple formats, flag failures, quantify the problem — is far more useful in production than blindly dropping bad rows or letting failures silently corrupt downstream analysis.

Removing Duplicates

Your source system generated duplicates during a migration. A "duplicate" in this context isn't just an identical row — it's a record with the same order_id and line_item_id combination. You want to keep only the most recently loaded version:

from pyspark.sql.window import Window
from pyspark.sql.functions import row_number, desc

# Define deduplication window — partition by business key, order by load timestamp
dedup_window = Window.partitionBy("order_id", "line_item_id").orderBy(desc("loaded_at"))

# Assign row numbers and keep only the first (most recent) per business key
deduplicated_sales = cleaned_sales \
    .withColumn("row_num", row_number().over(dedup_window)) \
    .filter(col("row_num") == 1) \
    .drop("row_num")

duplicates_removed = cleaned_sales.count() - deduplicated_sales.count()
print(f"Duplicate records removed: {duplicates_removed}")

The window function approach is more explicit — and safer — than dropDuplicates(), which removes rows that are fully identical. Real duplicates usually differ in some column (like a load timestamp), so you need the row-ranking approach to keep the right version.

Applying Business Logic

Let's add a revenue_usd column by converting regional currencies. Assume you have a simple exchange rate lookup that you'd normally load from a reference table, but for illustration we'll use a hardcoded map:

from pyspark.sql.functions import when

# Apply currency conversion based on region
cleaned_sales = deduplicated_sales.withColumn(
    "revenue_usd",
    when(col("region") == "APAC", col("revenue_local") * 0.66)   # AUD → USD
    .when(col("region") == "EMEA", col("revenue_local") * 1.08)  # EUR → USD
    .when(col("region") == "AMER", col("revenue_local") * 1.00)  # Already USD
    .otherwise(None)
).withColumn(
    "revenue_usd",
    F.round(col("revenue_usd"), 2)
)

In a real pipeline, you'd join to an exchange rates Delta table rather than hardcoding — we'll cover that in the next section. But the when().when().otherwise() pattern is the PySpark idiom for conditional column derivation, equivalent to SQL's CASE WHEN.

---

Enriching Data with Joins Against Reference Tables

Hardcoded lookup values belong in reference tables, not in code. Let's load a dim_product table from the lakehouse and join it to enrich the sales data with product category and product manager fields:

# Read a managed Delta table from the Tables zone
dim_product = spark.read.table("dim_product")

# Join on product_id to bring in category and product_manager
enriched_sales = cleaned_sales.join(
    dim_product.select("product_id", "category", "product_manager"),
    on="product_id",
    how="left"
)

# Check join quality — how many sales records didn't match a product?
unmatched = enriched_sales.filter(col("category").isNull()).count()
total = enriched_sales.count()
print(f"Join match rate: {((total - unmatched) / total * 100):.2f}%")

Note the how="left" join. In a transformation pipeline, you almost never want an inner join when enriching fact data — if a product ID is missing from your dimension table (a data quality issue, not a valid business scenario), an inner join silently drops those sales records. A left join surfaces the problem: those records have nulls in the category columns, which you can detect and alert on.

---

Using SQL Magic Cells for Exploration

One of the most productive features of Fabric notebooks is mixing Python and SQL in the same notebook. Any DataFrame you register as a temporary view can be queried with SQL in the same session:

# Register the enriched DataFrame as a temp view
enriched_sales.createOrReplaceTempView("enriched_sales_view")

Then in the next cell, switch to SQL:

%%sql

SELECT
    region,
    category,
    COUNT(*) AS transaction_count,
    ROUND(SUM(revenue_usd), 2) AS total_revenue_usd,
    ROUND(AVG(revenue_usd), 2) AS avg_revenue_usd
FROM enriched_sales_view
WHERE sale_date_parsed >= '2024-01-01'
  AND date_parse_failed = false
GROUP BY region, category
ORDER BY total_revenue_usd DESC
LIMIT 20

The %%sql magic cell executes the entire cell as Spark SQL against the session's registered views and managed tables. Results render as an interactive table directly in the notebook. This is invaluable for validation mid-pipeline — you can spot that APAC's electronics category is showing implausibly low revenue before you write bad data to your Delta table.

---

Writing Transformed Data to Delta Tables

This is where your transformation work becomes durable and useful downstream. Fabric writes Delta tables to the Tables/ path in your lakehouse, and they're immediately queryable via the SQL Analytics Endpoint.

Basic Write

# Select and rename columns for the final silver table
silver_sales = enriched_sales.select(
    col("order_id"),
    col("line_item_id"),
    col("sale_date_parsed").alias("sale_date"),
    col("region"),
    col("category"),
    col("product_id"),
    col("product_manager"),
    col("quantity"),
    col("revenue_local"),
    col("revenue_usd"),
    col("customer_id"),
    col("channel")
).filter(col("date_parse_failed") == False)

# Write to Delta format in the Tables zone
silver_sales.write \
    .format("delta") \
    .mode("overwrite") \
    .option("overwriteSchema", "true") \
    .save("Tables/silver_sales")

The .mode("overwrite") replaces the entire table each run. For a daily full refresh of a sales table, this is fine. For append-only logging or incremental loads, you'd use .mode("append").

Partitioned Writes for Performance

If your analysts frequently filter sales data by region or by year/month, partitioning the Delta table by those columns dramatically reduces how much data Spark (and Direct Lake) scans per query:

from pyspark.sql.functions import year, month

# Add partition columns
silver_sales_partitioned = silver_sales \
    .withColumn("year", year(col("sale_date"))) \
    .withColumn("month", month(col("sale_date")))

# Write with partitioning
silver_sales_partitioned.write \
    .format("delta") \
    .mode("overwrite") \
    .partitionBy("region", "year", "month") \
    .option("overwriteSchema", "true") \
    .save("Tables/silver_sales")

Physically, Delta stores the data in subfolders like Tables/silver_sales/region=APAC/year=2024/month=1/. When a query filters on region = 'APAC' AND year = 2024 AND month = 3, Spark reads only that subfolder rather than scanning the whole table. On a table with hundreds of millions of rows across three years of data, this can reduce query time from minutes to seconds.

---

Incremental Loads with Delta Merge (UPSERT)

A full overwrite works for smaller tables or initial loads, but for large fact tables you'll want to merge only new or changed records. Delta Lake's MERGE operation (also called UPSERT) handles this efficiently:

from delta.tables import DeltaTable

# Check whether the target table already exists
target_table_path = "Tables/silver_sales"

if DeltaTable.isDeltaTable(spark, target_table_path):
    # Target exists — run a merge
    target = DeltaTable.forPath(spark, target_table_path)
    
    target.alias("target").merge(
        silver_sales.alias("source"),
        "target.order_id = source.order_id AND target.line_item_id = source.line_item_id"
    ).whenMatchedUpdateAll() \
     .whenNotMatchedInsertAll() \
     .execute()
    
    print("Merge complete — records updated or inserted.")
else:
    # First run — create the table
    silver_sales.write \
        .format("delta") \
        .mode("overwrite") \
        .save(target_table_path)
    
    print("Initial table created.")

The merge condition identifies records as "the same" when both order_id and line_item_id match. Matched records get updated (if anything changed), unmatched source records get inserted as new rows, and target records with no corresponding source record are left alone (no deletes here — add .whenNotMatchedBySourceDelete() if you want full sync behavior).

This pattern is how most production lakehouses handle their daily loads. The fabric ecosystem makes it particularly clean because OneLake stores all of this as a single logical Delta table regardless of how many times you've merged into it.

---

Parameterizing Notebooks for Pipeline Execution

A notebook that only runs interactively is half the story. To make your transformation notebook part of an automated pipeline — called from a Data Pipeline's notebook activity — it needs to accept parameters.

Fabric uses a special cell type called a parameter cell to define defaults that can be overridden at runtime. To create one: in the cell you want to use as the parameter cell, click the three-dot menu on the right side of the cell and select Toggle parameter cell. You'll see a small "Parameters" tag appear at the bottom of the cell.

Inside that parameter cell:

# Parameter cell — these values are overridden by the pipeline at runtime
load_date = "2024-01-15"
region_filter = "ALL"
overwrite_mode = "merge"

Now in the rest of the notebook, use these variables:

from datetime import datetime

# Parse the load_date parameter
load_dt = datetime.strptime(load_date, "%Y-%m-%d").date()

# Apply region filter if specified
if region_filter != "ALL":
    raw_sales = raw_sales.filter(col("region") == region_filter)
    print(f"Filtered to region: {region_filter}")

# Use overwrite_mode parameter for the write strategy
if overwrite_mode == "overwrite":
    silver_sales.write.format("delta").mode("overwrite").save("Tables/silver_sales")
elif overwrite_mode == "merge":
    # ... run the merge logic from the previous section
    pass

When you call this notebook from a Data Pipeline, you pass values for load_date, region_filter, and overwrite_mode as pipeline parameters. The notebook receives them, overrides the parameter cell defaults, and runs with exactly the configuration the pipeline specifies. This turns your notebook into a reusable, testable transformation unit rather than a one-time script.

---

Hands-On Exercise

Work through this exercise to cement everything from the lesson. You'll build a complete transformation pipeline from raw files to a queryable Delta table.

Setup: Upload the following raw data to Files/raw/orders/ in your lakehouse. If you don't have real data available, create two simple CSV files manually — one representing "online" channel orders and one representing "in-store" channel orders, with a customer_id, product_id, order_date, quantity, and unit_price column. Introduce some variation: different date formats between the two files, a few duplicate order_id values, and one or two rows with an unparseable date.

Part 1 — Ingest and combine:

• Read both CSV files using spark.read.csv()
• Rename columns to a consistent standard using withColumnRenamed()
• Add a channel tag column using withColumn(F.lit(...))
• Union the two DataFrames using unionByName(allowMissingColumns=True)

Part 2 — Clean:

• Parse the order_date string to a proper date using coalesce() and multiple to_date() formats
• Flag any rows where date parsing failed
• Deduplicate on order_id keeping the row with the highest unit_price (use a window function with desc("unit_price"))
• Derive a total_price column: quantity * unit_price

Part 3 — Validate with SQL:

• Register the cleaned DataFrame as a temp view
• Write a %%sql cell that shows total revenue by channel and verifies no dates are null in the cleaned output

Part 4 — Write to Delta:

• Write the cleaned DataFrame to Tables/silver_orders with partitionBy("channel")
• Check the lakehouse explorer to confirm the table appears under Tables
• Query it via the SQL analytics endpoint using the Query tab in the lakehouse UI

Part 5 — Parameterize:

• Add a parameter cell with a target_table parameter defaulting to "Tables/silver_orders"
• Modify your write step to use the parameter value for the output path

---

Common Mistakes & Troubleshooting

The session takes 5 minutes to start, then my first cell is slow

Fabric's starter Spark pool takes 30-90 seconds to initialize — that's normal. The "slow first cell" is usually Spark's lazy evaluation resolving: Spark doesn't execute transformations until you call an action like .count(), .show(), or .write. If every cell feels slow, check whether you're calling multiple count actions unnecessarily — each one triggers a full Spark job.

AnalysisException: Path does not exist

This almost always means you're using the wrong relative path. When reading from a default lakehouse, Files/raw/sales/ is the correct relative path. If you've attached multiple lakehouses or have no default lakehouse set, you need the full abfss:// path. You can find the full path by hovering over a file in the lakehouse explorer — it shows the full ADLS Gen2 path.

Columns show up as StringType after reading CSV, even with inferSchema=True

inferSchema samples a subset of the file. If the first thousand rows of a numeric column contain values like "N/A" or empty strings, Spark infers the column as StringType. Fix this by casting explicitly after reading:

df = df.withColumn("quantity", col("quantity").cast("integer")) \
       .withColumn("unit_price", col("unit_price").cast("double"))

The Delta table appears in the lakehouse Files but not under Tables

This happens when you write to a path inside Files/ instead of Tables/. Double-check your save path. Only files written to the Tables/ path (or registered in the metastore explicitly) show up as lakehouse tables. This also means they won't be accessible to the SQL Analytics Endpoint or Power BI Direct Lake.

merge fails with ConcurrentAppendException

This occurs when two Spark jobs are writing to the same Delta table simultaneously. In production, ensure your pipeline orchestration doesn't trigger overlapping notebook runs against the same target table. Use pipeline concurrency settings to enforce sequential execution on tables that can't handle concurrent writes.

OutOfMemoryError on a large join

Large joins that exceed executor memory spill to disk or fail. Before joining large DataFrames, check whether you can filter them first to reduce size. If one side of the join is small (under ~100 MB), use a broadcast join to avoid shuffling the large DataFrame across the cluster:

from pyspark.sql.functions import broadcast

enriched = large_df.join(broadcast(small_lookup_df), on="product_id", how="left")

---

Summary & Next Steps

You've gone from raw CSV files to clean, partitioned, queryable Delta tables using PySpark in a Fabric notebook. Along the way, you handled the reality of messy source data — inconsistent schemas, bad dates, duplicates — with patterns that are production-grade, not demo-grade: multi-format date parsing with failure flagging, window-function deduplication, left joins with match-rate monitoring, and parameterized notebooks that plug cleanly into automated pipelines.

The key mental model to carry forward: notebooks sit in the transformation layer of your lakehouse architecture. Dataflow Gen2 handles ingestion and simpler transformations with a no-code interface, pipelines handle orchestration, but when you need distributed processing power, join-heavy logic, or fine-grained control over how data is written to Delta, notebooks are the right tool.

Here are your logical next steps within the Microsoft Fabric learning path:

• Orchestrate your notebook — Call it from a Data Pipeline on a schedule, passing load_date as a dynamic parameter based on @utcnow(). The lesson on Orchestrating Loads with Fabric Data Pipelines covers exactly this.
• Surface the data in Power BI — Your silver_sales Delta table is now Direct Lake-ready. Explore how Direct Lake reporting works against lakehouse tables without any import step.
• Understand what you built in broader context — If you want to see how notebooks, pipelines, Dataflow Gen2, and lakehouses fit into the Fabric platform holistically, What Is Microsoft Fabric? Workloads, OneLake, and How It Fits with Power BI is worth a read.

The patterns in this lesson — ingest from Files, clean and enrich, write partitioned Delta to Tables — are the foundation of every medallion architecture you'll build in Fabric. Once they're automatic, you can focus on the hard parts: business logic, data quality rules, and making the downstream consumers genuinely useful.