Learning Objectives

By the end of this lesson, you will be able to:

• Understand what Apache Spark is.
• Understand why Spark is used for big-data processing.
• Understand Spark architecture.
• Explain Driver, Executors, Jobs, Stages, and Tasks.
• Understand transformations and actions.
• Understand lazy evaluation.
• Understand Spark in Microsoft Fabric.
• Run basic PySpark code in a Fabric Notebook.

What is Apache Spark?

Apache Spark is an open-source distributed data processing engine designed to process large volumes of data quickly across multiple compute resources. Instead of processing a huge dataset on a single machine:

Large Dataset
     │
     ▼
Single Computer
     │
     ▼
Slow Processing

Spark distributes the workload:

                  Large Dataset
                       │
                       ▼
                  Apache Spark
                       │
          ┌────────────┼────────────┐
          ▼            ▼            ▼
       Executor 1   Executor 2   Executor 3
          │            │            │
       Partition 1  Partition 2  Partition 3
          │            │            │
          └────────────┼────────────┘
                       ▼
                  Final Result

This allows Spark to process large datasets efficiently.

Why Apache Spark?

Traditional data processing often runs on a single server. For example:

1 TB Data
   │
   ▼
One Server
   │
   ▼
Long Processing Time

Spark can distribute the work:

1 TB Data
   │
   ▼
Spark Cluster
   │
   ├── Worker 1
   ├── Worker 2
   ├── Worker 3
   ├── Worker 4
   └── Worker 5

Each worker processes part of the dataset.

Key Characteristics of Spark

Apache Spark provides:

• Distributed processing
• Parallel execution
• In-memory processing
• Fault tolerance
• SQL support
• Python support
• Scala support
• Java support
• Streaming capabilities
• Machine learning capabilities
• Integration with data lake storage

Apache Spark in Microsoft Fabric

Microsoft Fabric provides Spark compute for:

• Fabric Notebooks
• Spark Jobs
• Data Engineering
• Data Science
• Lakehouse processing

A typical Fabric architecture is:

                 Microsoft Fabric
                        │
                     OneLake
                        │
                    Lakehouse
                        │
                     Notebook
                        │
                   Spark Engine
                        │
              ┌─────────┴─────────┐
              │                   │
           Driver             Executors
                                  │
                         ┌────────┼────────┐
                         ▼        ▼        ▼
                      Task 1   Task 2   Task 3

Spark Architecture

The most important components are:

1. Driver
2. Executors
3. Cluster Manager
4. Jobs
5. Stages
6. Tasks
7. Partitions

1. Spark Driver

The Driver is the coordinator of a Spark application. It:

• Creates the Spark application.
• Creates the SparkSession.
• Converts operations into execution plans.
• Coordinates executors.
• Schedules tasks.
• Tracks execution.

Conceptually:

                Driver
                  │
       ┌──────────┼──────────┐
       ▼          ▼          ▼
   Executor 1 Executor 2 Executor 3

SparkSession

In PySpark, the SparkSession is the main entry point for working with Spark. In a Fabric Notebook, you can use:

spark

You can also access:

spark.version

to inspect the Spark version available to your session.

2. Spark Executors

Executors are processes that perform the actual data processing. They:

• Execute tasks.
• Process data partitions.
• Store intermediate results.
• Return results to the driver.

Example:

Driver
  │
  ├───────────────┐
  ▼               ▼
Executor 1     Executor 2
  │               │
Partition 1     Partition 2

3. Cluster Manager

A cluster manager is responsible for allocating compute resources. In general Spark deployments, Spark can work with cluster managers such as:

• Standalone
• Kubernetes
• YARN

Microsoft Fabric manages the Spark compute environment for Fabric workloads, so users typically focus on configuring and using the Spark session rather than manually managing the underlying cluster infrastructure.

4. Jobs

A Job represents a Spark computation triggered by an action. For example:

df.count()

The action causes Spark to execute the required computation.

5. Stages

A Spark job is divided into stages. A stage represents a group of operations that can be executed together without requiring a shuffle boundary. Conceptually:

Spark Job
    │
    ├── Stage 1
    │
    ├── Stage 2
    │
    └── Stage 3

6. Tasks

A Task is the smallest unit of work sent to an executor. If a DataFrame has multiple partitions:

DataFrame
│
├── Partition 1 → Task 1
├── Partition 2 → Task 2
├── Partition 3 → Task 3
└── Partition 4 → Task 4

The tasks can run in parallel.

7. Partitions

A partition is a logical chunk of distributed data. Example:

10 Million Records
       │
       ▼
    Spark
       │
 ┌─────┼─────┬─────┐
 ▼     ▼     ▼     ▼
P1    P2    P3    P4

Partitions allow Spark to distribute processing across executors.

Spark Execution Flow

Suppose we execute:

df.filter(df.Amount > 1000).count()

Spark roughly follows this process:

PySpark Code
     │
     ▼
Spark Driver
     │
     ▼
Execution Plan
     │
     ▼
Job
     │
     ▼
Stages
     │
     ▼
Tasks
     │
     ▼
Executors
     │
     ▼
Result

Transformations

A transformation creates a new DataFrame based on an existing DataFrame. Examples:

df.select("CustomerID", "Amount")
df.filter(df.Amount > 1000)
df.dropDuplicates()

Common transformations include:

• select()
• filter()
• where()
• withColumn()
• drop()
• dropDuplicates()
• groupBy()
• join()
• orderBy()

Actions

An action triggers execution of a Spark computation. Examples:

df.show()
df.count()
df.collect()
df.write.format("delta").saveAsTable("Sales")

Transformations vs Actions

Important Transformations generally build a computation plan. Actions cause Spark to execute that plan.

Lazy Evaluation

Spark uses lazy evaluation. Consider:

df2 = df.filter(df.Amount > 1000)

Spark generally does not immediately process all the data. It builds a logical execution plan. When you execute:

df2.count()

Spark performs the required computation.

filter()
   │
   │ Transformation
   ▼
Execution Plan
   │
   │ count()
   ▼
Actual Execution

Why Lazy Evaluation?

Lazy evaluation allows Spark to optimize the execution plan before processing the data. For example:

df.filter(df.Country == "India") \
  .select("CustomerID", "Amount") \
  .count()

Spark can optimize the execution rather than blindly executing every operation independently.

Narrow Transformations

A narrow transformation can process each partition without requiring data to be moved between partitions. Examples: filter() select() withColumn() Conceptually:

Partition 1 ──► Partition 1
Partition 2 ──► Partition 2
Partition 3 ──► Partition 3

No major redistribution is required.

Wide Transformations

A wide transformation requires data to be redistributed between partitions. Examples include: groupBy() join() orderBy() Conceptually:

Partition 1 ──┐
Partition 2 ──┼──► Shuffle ──► New Partitions
Partition 3 ──┤
Partition 4 ──┘

This redistribution is called a shuffle.

What is a Shuffle?

A shuffle occurs when Spark needs to redistribute data across partitions. For example:

df.groupBy("Country").count()

Records belonging to the same country may need to be brought together.

Partition 1 ──┐
Partition 2 ──┼──► Shuffle
Partition 3 ──┤
Partition 4 ──┘
                  │
                  ▼
             Country Groups

Shuffles can be expensive because they involve data movement.

Example: Spark DataFrame

Create sample data:

data = [
    (1, "John", "USA", 2500),
    (2, "Anita", "India", 3200),
    (3, "David", "UK", 1800),
    (4, "Priya", "India", 4100)
]

columns = [
    "CustomerID",
    "CustomerName",
    "Country",
    "Amount"
]

df = spark.createDataFrame(data, columns)

Display:

display(df)

Filtering Data

india_df = df.filter(
    df.Country == "India"
)

display(india_df)

Selecting Columns

df.select(
    "CustomerID",
    "CustomerName",
    "Amount"
).show()

Aggregating Data

df.groupBy("Country") \
  .sum("Amount") \
  .show()

Example result: India 7300 USA 2500 UK 1800

Spark SQL

Create a temporary view:

df.createOrReplaceTempView("sales")

Query it:

SELECT
    Country,
    SUM(Amount) AS TotalSales
FROM sales
GROUP BY Country;

Reading CSV

Spark can read files directly from the Lakehouse.

df = spark.read \
    .option("header", "true") \
    .option("inferSchema", "true") \
    .csv("Files/Sales.csv")

Display:

display(df)

Writing Delta

Write the DataFrame as a Delta table:

df.write \
    .format("delta") \
    .mode("overwrite") \
    .saveAsTable("Sales")

Then read:

sales_df = spark.read.table("Sales")

display(sales_df)

Spark and Delta Lake

In Microsoft Fabric, Spark works closely with Delta Lake.

Spark
  │
  ├── Read CSV
  ├── Transform
  ├── Clean
  └── Aggregate
       │
       ▼
    Delta Lake
       │
       ▼
    Lakehouse

Delta provides capabilities such as:

• ACID transactions
• Schema enforcement
• Schema evolution
• Updates
• Deletes
• MERGE
• Version history

Spark vs SQL

In Fabric, Spark and SQL complement each other.

Spark Use Cases

Apache Spark is commonly used for: Data Engineering

Raw Data
   ↓
Spark
   ↓
Clean
   ↓
Transform
   ↓
Delta
Data Science
Large Dataset
    ↓
Spark
    ↓
Feature Engineering
    ↓
Machine Learning
Log Processing
Application Logs
      ↓
Spark
      ↓
Parse
      ↓
Aggregate
      ↓
Analytics
IoT
IoT Data
   ↓
Spark
   ↓
Transform
   ↓
Delta
   ↓
Analytics

Advantages of Spark

1. Distributed Processing

Large datasets can be processed across multiple compute resources.

1. Scalability

Spark can process datasets much larger than a single machine's memory.

1. Multiple APIs

Spark supports:

• Python
• SQL
• Scala
• Java

1. Data Lake Integration

Spark works well with:

• Delta Lake
• Parquet
• CSV
• JSON
• OneLake
• ADLS

1. Unified Processing

Spark supports:

• Batch processing
• Streaming
• SQL
• Machine learning

Limitations of Spark

Spark is powerful, but it isn't always the best choice. Potential disadvantages include:

• Cluster startup overhead.
• Memory consumption.
• Shuffle costs.
• Poorly designed jobs can be expensive.
• Small datasets may not benefit from distributed processing.
• Collecting huge datasets to the driver can cause failures.

For example, avoid:

df.collect()
on a very large DataFrame because it attempts to bring all records to the driver.

Prefer:

df.limit(100).show()

for inspection.

Spark Best Practices

Select Only Required Columns Instead of:

df.select("*")

prefer:

df.select(
    "CustomerID",
    "Country",
    "Amount"
)

Filter Early

df.filter(df.Country == "India")

Reducing data early can reduce downstream processing. Avoid Unnecessary Shuffles Be careful with: groupBy join orderBy distinct on very large datasets. Avoid collect() on Large DataFrames Use:

df.limit(10).show()

instead. Use Appropriate Partitioning Good partitioning helps Spark process data efficiently.

Spark Execution Example

Consider:

result = (

df

    .filter(df.Amount > 1000)
    .groupBy("Country")
    .sum("Amount")
)

result.show()

Execution conceptually becomes:

DataFrame
   │
   ▼
Filter
   │
   ▼
GroupBy
   │
   ▼
Shuffle
   │
   ▼
Aggregation
   │
   ▼
show()
   │
   ▼
Result

The filter() and groupBy() build the plan; show() triggers execution.

Hands-On Lab

Objective Use a Fabric Notebook to read, transform, and write data using Spark. Step 1 — Create Notebook Create a new Notebook in your Fabric workspace. Step 2 — Create DataFrame

data = [
    (1, "John", "USA", 2500),
    (2, "Anita", "India", 3200),
    (3, "David", "UK", 1800),
    (4, "Priya", "India", 4100)
]

columns = [
    "CustomerID",
    "CustomerName",
    "Country",
    "Amount"
]

df = spark.createDataFrame(data, columns)

Step 3 — Display Data

display(df)

Step 4 — Filter

df_india = df.filter(
    df.Country == "India"
)

display(df_india)

Step 5 — Aggregate

df.groupBy("Country") \
  .sum("Amount") \
  .show()

Step 6 — Write Delta

df.write \
    .format("delta") \
    .mode("overwrite") \
    .saveAsTable("CustomerSales")

Step 7 — Query Delta

SELECT *
FROM CustomerSales;

Interview Questions

              Spark Application
                     │
                   Driver
                     │
          ┌──────────┼──────────┐
          ▼          ▼          ▼
      Executor    Executor    Executor
          │          │          │
      Partition   Partition   Partition
          │          │          │
          └──────────┼──────────┘
                     ▼
                  Result

Learning Objectives

By the end of this lesson, you will be able to:

• Understand what a Spark cluster is.
• Understand Driver and Executor architecture.
• Understand Spark nodes, cores, memory, jobs, stages, and tasks.
• Understand how Fabric manages Spark compute.
• Understand Spark sessions in Microsoft Fabric.
• Understand autoscaling and dynamic resource allocation.
• Choose appropriate compute for different workloads.
• Identify common Spark performance problems.

What is a Spark Cluster?

A Spark cluster is a group of compute resources that work together to process large datasets using Apache Spark. Instead of processing everything on one machine:

Large Dataset
     │
     ▼
One Computer
     │
     ▼
Processing

Spark distributes the workload:

                    Large Dataset
                         │
                         ▼
                   Spark Cluster
                         │
          ┌──────────────┼──────────────┐
          ▼              ▼              ▼
      Executor 1     Executor 2     Executor 3
          │              │              │

Partition 1 Partition 2 Partition 3

          │              │              │
          └──────────────┼──────────────┘
                         ▼
                       Result

This allows multiple pieces of a dataset to be processed in parallel.

Spark Cluster Architecture

A Spark application typically consists of:

                 Spark Application
                        │
                        ▼
                  Spark Driver
                        │
             ┌──────────┼──────────┐
             ▼          ▼          ▼
         Executor 1  Executor 2  Executor 3
             │          │          │
           Tasks      Tasks       Tasks
             │          │          │
             └──────────┼──────────┘
                        ▼
                     Storage
                  OneLake / ADLS

The two most important components are:

• Driver
• Executors

1. Driver

The Driver is the coordinator of a Spark application. It is responsible for:

• Creating the SparkSession.
• Understanding the application code.
• Building the execution plan.
• Scheduling tasks.
• Coordinating executors.
• Tracking application progress.

Example:

df = spark.read.table("Sales")

result = df.filter(
    df.Amount > 1000
)

result.count()

The Driver coordinates the execution of this operation.

2. Executors

Executors perform the actual data processing. Each executor can:

• Run tasks.
• Process data partitions.
• Store intermediate results.
• Return results to the Driver.

Example:

Driver
  │
  ├───────────────┐
  ▼               ▼
Executor 1     Executor 2
  │               │
Partition 1     Partition 2
Partition 3     Partition 4

Driver vs Executor

Simple analogy Think of a construction project: Project Manager = Driver

Workers = Executors

Work assignments = Tasks

Materials = Data Partitions The manager doesn't build every component personally. The manager coordinates the workers.

What is a Node?

A node is a compute machine or compute resource participating in the cluster. Conceptually:

Spark Cluster
│
├── Driver Node
│
├── Worker Node
│
├── Worker Node
└── Worker Node

The exact infrastructure details are managed by the Fabric Spark environment.

CPU Cores

Each compute resource can have multiple CPU cores. For example:

Executor
│
├── Core 1 → Task
├── Core 2 → Task
├── Core 3 → Task
└── Core 4 → Task

More available cores can allow more tasks to execute concurrently, assuming there is enough parallelism in the workload.

Memory

Spark workloads require memory for:

• Data processing
• Joins
• Aggregations
• Caching
• Shuffle operations
• Intermediate results

Example:

Executor
│
├── CPU Cores
└── Memory
      │
      ├── Execution
      └── Storage

Insufficient memory can result in failures or excessive spilling to disk.

Spark Partitions

Spark divides datasets into partitions. Example:

100 GB Dataset
       │
       ▼
Spark
       │
 ┌─────┼─────┬─────┐
 ▼     ▼     ▼     ▼
P1    P2    P3    P4
25GB  25GB  25GB  25GB

Executors process these partitions.

Tasks

A task is a unit of work performed on one partition. Example:

Partition 1 → Task 1
Partition 2 → Task 2
Partition 3 → Task 3
Partition 4 → Task 4

If there are enough executor resources, these tasks can run concurrently.

Jobs

A Spark job is created when an action triggers execution. For example:

df.count()
The count() action can create a Spark job.

Stages

A Spark job is divided into stages. Example:

Job
 │
 ├── Stage 1
 │
 ├── Stage 2
 │
 └── Stage 3

Stages are separated by operations that require data redistribution, commonly called shuffle boundaries.

Complete Execution Hierarchy

A useful hierarchy to remember is:

Application
    │
    ▼
   Job
    │
    ▼
  Stages
    │
    ▼
  Tasks
    │
    ▼
Partitions

And:

Driver
   │
   ▼
Jobs
   │
   ▼
Stages
   │
   ▼
Tasks
   │
   ▼
Executors

Spark Cluster in Microsoft Fabric

Microsoft Fabric provides a managed Spark environment. A Fabric Notebook can use Spark without requiring you to manually install and configure Apache Spark. Conceptually:

              Microsoft Fabric
                     │
                  Workspace
                     │
                  Notebook
                     │
                Spark Session
                     │
             ┌───────┴───────┐
             ▼               ▼
          Driver          Executors
                             │
                             ▼
                           OneLake

Fabric handles much of the underlying infrastructure management.

Spark Session

A Spark application works through a SparkSession. In a Fabric Notebook:

spark

You can inspect information about the Spark environment using:

spark.version

You can also create DataFrames:

data = [
    (1, "John"),
    (2, "Anita"),
    (3, "David")
]

df = spark.createDataFrame(
    data,
    ["ID", "Name"]
)

display(df)

Spark Session Lifecycle

A typical Notebook workflow is:

Open Notebook
     │
     ▼
Start Spark Session
     │
     ▼
Run PySpark Code
     │
     ▼
Execute Jobs
     │
     ▼
Process Data
     │
     ▼
Session Ends

In Fabric, the Spark session and its resources are managed by the platform according to the configured environment and workload.

Fabric Spark Compute

Fabric provides different compute configurations and capacities depending on the workspace and environment. When designing a workload, consider:

• Dataset size
• Number of transformations
• Join complexity
• Number of concurrent users/jobs
• Memory requirements
• Execution duration
• Cost/performance requirements

A small development workload does not necessarily require the same resources as a production ETL workload.

Autoscaling

Autoscaling allows compute resources to adjust according to workload requirements where supported by the Fabric Spark configuration. Conceptually:

Low Workload
     │
     ▼
Fewer Resources

When workload increases:

High Workload
     │
     ▼
More Resources

Example:

             Workload
                │
       ┌────────┴────────┐
       ▼                 ▼
     Low               High
       │                 │
       ▼                 ▼
   Small Compute    More Compute

The exact behavior depends on the Fabric Spark settings and capacity configuration.

Dynamic Resource Allocation

Dynamic resource allocation allows Spark to adjust executor resources based on workload. Conceptually:

Workload
   │
   ├── Low  → Fewer Executors
   │
   ├── Medium → More Executors
   │
   └── High → More Executors

This can help avoid allocating unnecessary compute for periods of low activity.

Example: Small Workload

Suppose you process: 100 MB CSV A very large cluster would be unnecessary.

100 MB
  │
  ▼
Small Spark Compute
  │
  ▼
Processing

Example: Large Workload

Suppose you process: 5 TB Sales Data The workload may benefit from distributed execution:

5 TB
 │
 ▼
Spark
 │
 ├── Executor 1
 ├── Executor 2
 ├── Executor 3
 ├── Executor 4
 ├── Executor 5
 └── ...

Parallel Processing

Suppose there are four partitions: P1 P2 P3 P4 With sufficient executor resources:

Executor 1 → P1
Executor 2 → P2
Executor 3 → P3
Executor 4 → P4

They can be processed concurrently. This is one of the fundamental reasons Spark can process large datasets efficiently.

Partition Count Matters

Suppose you have: 100 partitions but only: 4 available task slots Spark processes them in multiple waves:

Wave 1 → P1 P2 P3 P4
Wave 2 → P5 P6 P7 P8
Wave 3 → P9 P10 P11 P12

... Therefore, parallelism depends on both:

• Number of partitions
• Available compute resources

Too Few Partitions

Example: 4 partitions + 100 available cores Only a small number of tasks can execute simultaneously.

100 Cores
     │
     ▼
4 Tasks

Many cores may remain underutilized.

Too Many Partitions

The opposite problem can also occur. For example: 10 million tiny partitions Spark may spend excessive time managing tasks. Therefore, partitioning should be balanced.

Spark Shuffle

Some operations require data to move between executors. Example:

df.groupBy("Country").count()

Conceptually:

Executor 1 ──┐
Executor 2 ──┤
Executor 3 ──┼──► Shuffle ──► Aggregation
Executor 4 ──┘

Shuffle can be expensive because of:

• Network transfer
• Disk I/O
• Serialization
• Memory pressure

Join Example

Consider:

orders.join(
    customers,
    orders.CustomerID == customers.CustomerID
)

Spark may need to redistribute data to perform the join.

Orders
  │
  ├────────┐
  │        │
  ▼        ▼
Executor 1 Executor 2
       │
       ▼
    Shuffle
       │
       ▼
    Join

Large joins should be designed carefully.

Caching

Spark can cache data that will be reused. Example:

df.cache()

Then:

df.count()
and subsequent operations may reuse the cached representation.
However, do not cache everything.

Caching consumes memory and is most useful when:

• The same DataFrame is reused multiple times.
• Recomputing it is expensive.
• Sufficient memory is available.

Example

Suppose:

df = spark.read.table("LargeSales")

You perform:

df.filter(df.Country == "India").count()
and then:
df.groupBy("Country").sum("Amount")
If the same expensive base DataFrame is reused repeatedly, caching may sometimes help.
df.cache()

But always measure performance rather than caching automatically.

Driver Memory Problem

A common mistake is:

df.collect()
If the DataFrame contains millions of records, Spark attempts to bring them to the Driver.

Millions of Records
       │
       ▼
Executors
       │
       ▼
    collect()
       │
       ▼
     Driver
       │
       ▼
   Memory Error

For inspection, use:

df.limit(100).show()
or:
display(df.limit(100))

Executor Memory Problem

A large transformation may require more executor memory than available. For example:

Large Join
    │
    ▼
Large Shuffle
    │
    ▼
Executor Memory Pressure

Possible solutions include:

• Optimize joins.
• Reduce unnecessary columns.
• Filter data earlier.
• Improve partitioning.
• Avoid unnecessary caching.
• Increase appropriate compute resources when justified.

Cluster Sizing Considerations

Before choosing compute, consider:

Development vs Production

Development You may have:

Small Dataset
     │
     ▼
Small Compute
     │
     ▼
Notebook Testing
Production

You may have:

Large Dataset
     │
     ▼
Production Spark Environment
     │
     ▼
Scheduled Pipeline
     │
     ▼
Delta Tables

Production workloads require proper monitoring, resource planning, and failure handling.

Spark Cluster Monitoring

When troubleshooting a Spark job, examine:

• Job duration
• Stage duration
• Task duration
• Number of tasks
• Failed tasks
• Shuffle read
• Shuffle write
• Executor memory
• Executor CPU
• Data skew
• Input/output size

The Spark monitoring interfaces available in Fabric can help identify bottlenecks.

Data Skew

Data skew occurs when one or a few partitions contain significantly more data than others. Example:

Partition 1 → 1 GB
Partition 2 → 1 GB
Partition 3 → 1 GB
Partition 4 → 100 GB   ← Skew

The first three tasks may finish quickly while the fourth takes much longer.

P1 ───── Done
P2 ───── Done
P3 ───── Done
P4 ───────────────────────── Long Running

This can make the entire stage wait for the slow partition.

Common Causes of Data Skew

• Highly uneven key distribution.
• A very common join key.
• Poor partitioning.
• Large numbers of records associated with one key.

Example:

Country
India → 95 million
USA   → 2 million
UK    → 1 million

Partitioning by Country may create an uneven workload.

Cluster Performance Optimization

1. Filter Early

df.filter(df.Amount > 1000)

before expensive transformations.

1. Select Required Columns

df.select(
    "CustomerID",
    "Amount"
)

1. Avoid Unnecessary Shuffles

Be careful with: groupBy() join() distinct() orderBy()

1. Avoid collect()

Do not bring large datasets to the Driver.

1. Don't Over-Partition

Too many tiny partitions create unnecessary task overhead.

1. Don't Under-Partition

Too few partitions prevent effective parallel processing.

Spark Cluster Example

Suppose we have: 1 TB Sales Data The data is divided into: 100 partitions And the cluster provides: 10 executor task slots Spark processes approximately:

Wave 1 → 10 partitions
Wave 2 → 10 partitions
Wave 3 → 10 partitions

...

Wave 10 → 10 partitions

Conceptually:

              1 TB
               │
               ▼
        100 Partitions
               │
               ▼
        ┌──────┴──────┐
        │ Spark Cluster│
        └──────┬──────┘
               │
     ┌─────────┼─────────┐
     ▼         ▼         ▼
 Executor 1 Executor 2 Executor 3
     │         │         │
    Tasks     Tasks      Tasks

Hands-On Lab

Objective Understand Spark execution in a Fabric Notebook. Step 1 — Create a DataFrame

data = [
    (1, "India", 1000),
    (2, "USA", 2000),
    (3, "India", 3000),
    (4, "UK", 1500)
]

df = spark.createDataFrame(
    data,
    ["ID", "Country", "Amount"]
)

Step 2 — Inspect Partitions

df.rdd.getNumPartitions()

Step 3 — Filter

filtered_df = df.filter(
    df.Amount > 1500
)

Step 4 — Trigger an Action

filtered_df.count()
The count() action causes Spark to execute the required computation.

Step 5 — Group Data

df.groupBy("Country").sum("Amount").show()

This introduces an aggregation and potentially a shuffle.

Interview Questions

Spark Application
       │
       ▼
     Driver
       │
       ▼
      Job
       │
       ▼
     Stages
       │
       ▼
     Tasks
       │
       ▼
   Partitions
       │
       ▼
   Executors
Fabric Data Engineering Pattern
OneLake
   │
   ▼
Lakehouse
   │
   ▼
Fabric Notebook
   │
   ▼
Spark Session
   │
   ├── Driver
   │
   └── Executors
          │
          ▼
      Data Processing
          │
          ▼
      Delta Tables

Learning Objectives

By the end of this lesson, you will be able to:

• Understand PySpark and its role in Microsoft Fabric.
• Create and use a Spark DataFrame.
• Select and filter data.
• Add, modify, and remove columns.
• Handle NULL values.
• Sort and aggregate data.
• Use basic PySpark functions.
• Understand transformations and actions.
• Write PySpark data to Delta tables.

What is PySpark?

PySpark is the Python API for Apache Spark. It allows data engineers to use Python to perform distributed data processing. Instead of writing Spark applications in Scala or Java, you can use Python:

Python
   │
   ▼
PySpark
   │
   ▼
Apache Spark
   │
   ▼
Distributed Processing

In Microsoft Fabric, PySpark is commonly used inside Fabric Notebooks.

PySpark in Microsoft Fabric

A typical architecture is:

                Microsoft Fabric
                       │
                    Notebook
                       │
                    PySpark
                       │
                  Spark Engine
                       │
            ┌──────────┴──────────┐
            │                     │
         Driver               Executors
                                  │
                                  ▼
                              OneLake
                                  │
                              Lakehouse

PySpark can be used to:

• Read CSV files.
• Read Parquet files.
• Read Delta tables.
• Clean data.
• Transform data.
• Join datasets.
• Aggregate data.
• Write Delta tables.
• Perform incremental processing.

Creating a Fabric Notebook

In your Fabric workspace:

New Item → Notebook

You can start writing PySpark code in a notebook cell. In Fabric, the spark variable is typically already available.

spark

You can check the Spark version:

spark.version

4.3.1 Creating Your First DataFrame

A DataFrame is one of the most important PySpark concepts. Create sample data:

data = [
    (1, "John", "USA", 2500),
    (2, "Anita", "India", 3200),
    (3, "David", "UK", 1800),
    (4, "Priya", "India", 4100)
]

Define the column names:

columns = [
    "CustomerID",
    "CustomerName",
    "Country",
    "Amount"
]

Create the DataFrame:

df = spark.createDataFrame(
    data,

columns

)

Display it:

display(df)

Result:

4.3.2 Inspecting a DataFrame

Display Data

display(df)

Show Rows

df.show()

Show First 2 Rows

df.show(2)

Get Row Count

df.count()

Get Column Names

df.columns

Get Schema

df.printSchema()

Example: root |-- CustomerID: long |-- CustomerName: string |-- Country: string |-- Amount: long

4.3.3 Selecting Columns

Select one column:

df.select("CustomerName").show()

Select multiple columns:

df.select(
    "CustomerID",
    "CustomerName",
    "Amount"
).show()

You can also use:

df.select(
    df.CustomerID,
    df.CustomerName
).show()

4.3.4 Filtering Data

Filter customers from India:

india_df = df.filter(
    df.Country == "India"
)

display(india_df)

You can also use:

india_df = df.where(
    df.Country == "India"
)
Both filter() and where() are commonly used.

Multiple Conditions

Find customers from India with Amount greater than 3000:

result = df.filter(
    (df.Country == "India") &
    (df.Amount > 3000)
)

display(result)

Important Use: & for AND. Use: | for OR. Do not use Python's normal and / or for Spark column expressions.

OR Condition

result = df.filter(
    (df.Country == "India") |
    (df.Country == "USA")
)

display(result)

NOT Condition

result = df.filter(
    df.Country != "India"
)

display(result)

4.3.5 Adding Columns

Use withColumn() to create a new column. Suppose we want to calculate a 10% tax:

from pyspark.sql.functions import col

df2 = df.withColumn(
    "Tax",
    col("Amount") * 0.10
)

display(df2)

Result:

Creating a Derived Column

Calculate total amount including tax:

df2 = df2.withColumn(
    "TotalAmount",
    col("Amount") + col("Tax")
)

display(df2)

4.3.6 Renaming Columns

Use withColumnRenamed():

df2 = df.withColumnRenamed(
    "CustomerName",
    "Customer_Name"
)

You can rename multiple columns:

df2 = df \
    .withColumnRenamed("CustomerID", "Customer_ID") \
    .withColumnRenamed("CustomerName", "Customer_Name")

4.3.7 Dropping Columns

Remove a column:

df2 = df.drop("Amount")

Multiple columns:

df2 = df.drop(
    "Amount",
    "Country"
)

4.3.8 Sorting Data

Sort by Amount:

df.orderBy("Amount").show()

Descending order:

from pyspark.sql.functions import desc

df.orderBy(
    desc("Amount")
).show()

4.3.9 Handling NULL Values

Suppose: CustomerName = NULL Check NULL records:

df.filter(
    col("CustomerName").isNull()
).show()

Check non-NULL:

df.filter(
    col("CustomerName").isNotNull()
).show()

Filling NULL Values

df.fillna({
    "Country": "Unknown"
})

For numeric values:

df.fillna({

"Amount": 0

})

Removing NULL Records

df.dropna()
Or only when a particular column is NULL:
df.dropna(
    subset=["CustomerID"]
)

4.3.10 Removing Duplicates

Remove complete duplicate rows:

df.dropDuplicates()

Remove duplicates based on a key:

df.dropDuplicates(
    ["CustomerID"]
)

This is particularly useful when preparing data for Delta tables.

4.3.11 String Functions

PySpark provides many functions for string manipulation. Import:

from pyspark.sql.functions import trim, upper, lower

Trim spaces:

df.withColumn(
    "CustomerName",
    trim(col("CustomerName"))
)

Convert to uppercase:

df.withColumn(
    "Country",
    upper(col("Country"))
)

Convert to lowercase:

df.withColumn(
    "CustomerName",
    lower(col("CustomerName"))
)

4.3.12 Data Type Conversion

Use cast().

df2 = df.withColumn(
    "Amount",
    col("Amount").cast("double")
)

Check:

df2.printSchema()

Common Data Types

4.3.13 Aggregations

Calculate total sales:

from pyspark.sql.functions import sum

df.select(
    sum("Amount").alias("TotalSales")
).show()

Calculate average:

from pyspark.sql.functions import avg

df.select(
    avg("Amount").alias("AverageSales")
).show()

Calculate maximum:

from pyspark.sql.functions import max

df.select(
    max("Amount").alias("MaximumSale")
).show()

4.3.14 GroupBy

Calculate sales by country:

df.groupBy("Country") \
  .sum("Amount") \
  .show()

A cleaner approach is:

from pyspark.sql.functions import sum

result = df.groupBy("Country").agg(
    sum("Amount").alias("TotalSales")
)

display(result)

Result:

Multiple Aggregations

from pyspark.sql.functions import (
    sum,
    avg,

count

)

result = df.groupBy("Country").agg(
    count("*").alias("CustomerCount"),
    sum("Amount").alias("TotalSales"),
    avg("Amount").alias("AverageSales")
)

display(result)

4.3.15 PySpark Joins

Suppose we have: Customers

Orders

Join them:

result = orders.join(
    customers,
    orders.CustomerID == customers.CustomerID,
    "inner"
)

Select required columns:

result.select(
    orders.OrderID,
    customers.CustomerName,

orders.Amount

).show()

Types of Joins

PySpark supports: inner left right full left_semi left_anti cross Example:

df1.join(
    df2,
    "CustomerID",
    "left"
)

4.3.16 Reading CSV

Read a CSV file from the Lakehouse:

df = spark.read \
    .option("header", "true") \
    .option("inferSchema", "true") \
    .csv("Files/Sales.csv")

Display:

display(df)

Explicit Schema

For production pipelines, explicitly defining the schema is often preferable to relying on schema inference.

from pyspark.sql.types import (
    StructType,
    StructField,
    IntegerType,
    StringType,

DoubleType

)

schema = StructType([
    StructField("SaleID", IntegerType(), True),
    StructField("CustomerName", StringType(), True),
    StructField("Amount", DoubleType(), True)
])

Read:

df = spark.read \
    .option("header", "true") \
    .schema(schema) \
    .csv("Files/Sales.csv")

4.3.17 Reading Parquet

df = spark.read.parquet(
    "Files/Sales.parquet"
)

4.3.18 Reading Delta

df = spark.read.format("delta").load(
    "Tables/Sales"
)
Or, if registered as a table:
df = spark.read.table("Sales")

4.3.19 Writing Data

Write as Parquet:

df.write \
    .mode("overwrite") \
    .parquet("Files/Output")

Write as Delta:

df.write \
    .format("delta") \
    .mode("overwrite") \
    .saveAsTable("Sales")

Write Modes

Common modes are:

Example:

df.write \
    .format("delta") \
    .mode("append") \
    .saveAsTable("Sales")

4.3.20 Transformations vs Actions

This is an important PySpark concept. Transformations Examples:

df.select(...)
df.filter(...)
df.withColumn(...)
df.groupBy(...)
df.join(...)

They build a logical execution plan. Actions Examples:

df.show()
df.count()
df.collect()
df.first()

They trigger execution.

Example

result = df.filter(
    col("Amount") > 2000
)

This creates a transformation. Spark doesn't necessarily execute immediately. Then:

result.count()
is an action and triggers execution.

4.3.21 PySpark Functions

Common functions you'll use frequently:

from pyspark.sql.functions import (
    col,
    lit,
    when,
    sum,
    avg,
    count,
    max,
    min,
    trim,
    upper,
    lower,
    regexp_replace,
    to_date,

current_date

)

when() Function

Create conditional logic.

df2 = df.withColumn(
    "CustomerType",
    when(
        col("Amount") >= 3000,
        "Premium"
    ).otherwise("Standard")
)

display(df2)

Result:

lit() Function

Add a constant value:

df2 = df.withColumn(
    "SourceSystem",
    lit("SQLServer")
)

Result: SourceSystem ------------ SQLServer SQLServer SQLServer SQLServer

Date Functions

Example:

from pyspark.sql.functions import to_date

df2 = df.withColumn(
    "SaleDate",
    to_date(col("SaleDate"))
)

Current date:

from pyspark.sql.functions import current_date

df2 = df.withColumn(
    "LoadDate",
    current_date()
)

4.3.22 Practical Data Cleaning Example

Suppose the source contains:

CustomerName = "  JOHN  "
Country       = "india"
Amount        = "2500"

Clean it:

from pyspark.sql.functions import (
    col,
    trim,

upper

)

clean_df = df \
    .withColumn(
        "CustomerName",
        trim(col("CustomerName"))
    ) \
    .withColumn(
        "Country",
        upper(col("Country"))
    ) \
    .withColumn(
        "Amount",
        col("Amount").cast("double")
    )

4.3.23 End-to-End PySpark Example

Suppose: Files/Sales.csv contains sales data. Step 1 — Read

df = spark.read \
    .option("header", "true") \
    .option("inferSchema", "true") \
    .csv("Files/Sales.csv")

Step 2 — Inspect

df.printSchema()
display(df)

Step 3 — Clean

from pyspark.sql.functions import (
    col,
    trim,

upper

)

clean_df = df \
    .dropDuplicates() \
    .withColumn(
        "CustomerName",
        trim(col("CustomerName"))
    ) \
    .withColumn(
        "Country",
        upper(col("Country"))
    ) \
    .withColumn(
        "Amount",
        col("Amount").cast("double")
    )

Step 4 — Filter

clean_df = clean_df.filter(
    col("Amount") > 0
)

Step 5 — Add Load Date

from pyspark.sql.functions import current_date

clean_df = clean_df.withColumn(
    "LoadDate",
    current_date()
)

Step 6 — Write Delta

clean_df.write \
    .format("delta") \
    .mode("overwrite") \
    .saveAsTable("Sales")

Step 7 — Query

SELECT
    Country,
    SUM(Amount) AS TotalSales
FROM Sales
GROUP BY Country
ORDER BY TotalSales DESC;

PySpark Coding Best Practices

1. Avoid collect() on Large Data

Avoid:

df.collect()

for large datasets. Use:

df.limit(100).show()

instead.

2. Select Required Columns

Instead of:

df.select("*")

prefer:

df.select(
    "CustomerID",
    "Country",
    "Amount"
)
when you don't need every column.

3. Filter Early

df = df.filter(
    col("Amount") > 0
)

Reducing data early can improve downstream processing.

4. Avoid Excessive withColumn()

When many transformations are required, consider organizing the logic clearly rather than creating a long chain of unnecessary intermediate columns.

5. Define Schemas for Production

Prefer:

.schema(schema)
over relying entirely on:
.inferSchema

for stable production pipelines.

Common PySpark Errors

AnalysisException Often caused by:

• Incorrect column name
• Missing table
• Invalid SQL expression
• Schema mismatch

Example:

df.select("CustmerID")
when the actual column is:

CustomerID

TypeError

Often caused by incorrect Python/Spark expressions. For example, use: col("Amount") > 1000 rather than trying to treat a Spark column like a normal Python value.

Out of Memory

Possible causes: collect() Large joins Large aggregations Excessive caching Data skew

Interview Questions

df.filter(col("Amount") > 1000)
and:
df.where(col("Amount") > 1000)

df.write \
    .format("delta") \
    .mode("overwrite") \
    .saveAsTable("Sales")

Hands-On Lab

Project: Customer Sales Transformation Input Files/Customers.csv Columns: CustomerID CustomerName Country Amount Requirements Build a PySpark Notebook that:

1. Reads the CSV.
2. Displays the data.
3. Prints the schema.
4. Removes duplicate records.
5. Trims customer names.
6. Converts country to uppercase.
7. Converts Amount to double.
8. Removes records where Amount ≤ 0.
9. Adds CustomerType.
10. Adds LoadDate.
11. Calculates sales by country.
12. Writes the cleaned data to a Delta table.

Expected architecture

Customers.csv
      │
      ▼
Fabric Notebook
      │
      ▼
    PySpark
      │
      ▼
   DataFrame
      │
      ├── Clean
      ├── Filter
      ├── Transform
      └── Validate
             │
             ▼
        Delta Table
             │
             ▼
          Power BI

Module 4.3 Summary

The Core PySpark Pattern

Read
  ↓
Inspect
  ↓
Clean
  ↓
Transform
  ↓
Filter
  ↓
Aggregate / Join
  ↓
Validate
  ↓
Write Delta

Key idea: PySpark gives you the ability to use Python to build distributed data-processing pipelines that run on Spark and integrate naturally with Fabric Lakehouses and Delta tables.

Learning Objectives

By the end of this lesson, you will be able to:

• Understand Spark SQL and why it is used.
• Create temporary and global views.
• Query DataFrames using SQL.
• Filter, aggregate, sort, and join data.
• Use SQL functions and window functions.
• Query Delta tables from Spark SQL.
• Compare Spark SQL with PySpark DataFrame operations.
• Build practical Fabric Notebook transformations using Spark SQL.

What is Spark SQL?

Spark SQL is the SQL interface of Apache Spark. It allows you to use SQL statements to query and transform data that Spark can access. In Microsoft Fabric, Spark SQL is particularly useful for working with:

• Lakehouse tables
• Delta tables
• DataFrames
• CSV files
• Parquet files
• Views

The basic architecture is:

             Microsoft Fabric
                    │
                 Notebook
                    │
              Spark Session
                    │
                Spark SQL
                    │
          ┌─────────┴─────────┐
          │                   │
      DataFrames          Delta Tables
          │                   │
          └─────────┬─────────┘
                    ▼
                 OneLake

Why Spark SQL?

Data engineers who already know SQL can use Spark without having to write everything in Python. For example, instead of:

df.groupBy("Country").sum("Amount")

you can write:

SELECT
    Country,
    SUM(Amount) AS TotalSales
FROM sales
GROUP BY Country;

This makes Spark accessible to both:

• SQL developers
• Python/data engineers

Spark SQL vs Traditional SQL

Spark SQL looks similar to standard SQL, but it operates on Spark's distributed processing engine.

Traditional SQL
       │
       ▼
Database Engine
       │
       ▼
Server / Warehouse

Spark SQL:

Spark SQL
    │
    ▼
Spark Engine
    │
 ┌──┴──────────────┐
 ▼                 ▼
Executor          Executor
 │                 │
Partition         Partition

The SQL query can therefore be executed across distributed data.

4.4.1 Creating a DataFrame

Let's create sample sales data.

data = [
    (1, "John", "USA", 2500),
    (2, "Anita", "India", 3200),
    (3, "David", "UK", 1800),
    (4, "Priya", "India", 4100),
    (5, "Mike", "USA", 2900)
]

columns = [
    "CustomerID",
    "CustomerName",
    "Country",
    "Amount"
]

df = spark.createDataFrame(data, columns)

display(df)

4.4.2 Creating a Temporary View

Spark SQL works with tables and views. Register the DataFrame as a temporary view:

df.createOrReplaceTempView("sales")

Now Spark SQL can query it.

SELECT *
FROM sales;

Temporary View

A temporary view:

• Exists within the Spark session.
• Is based on a DataFrame or query.
• Is useful for Notebook transformations.
• Does not represent a permanent Lakehouse table.

Example:

df.createOrReplaceTempView("sales")

Then:

SELECT *
FROM sales;

4.4.3 Selecting Columns

SELECT
    CustomerID,
    CustomerName,
    Country,

Amount FROM sales; Select specific columns:

SELECT
    CustomerName,

Amount FROM sales;

Using Aliases

SELECT
    CustomerName AS Customer,

Amount AS SalesAmount FROM sales;

4.4.4 Filtering Data

Find customers from India:

SELECT *
FROM sales
WHERE Country = 'India';
Find sales greater than 3,000:
SELECT *
FROM sales
WHERE Amount > 3000;

Multiple Conditions

SELECT *
FROM sales
WHERE Country = 'India'
  AND Amount > 3000;
OR:
SELECT *
FROM sales
WHERE Country = 'India'
   OR Country = 'USA';

IN Operator

Instead of: WHERE Country = 'India' OR Country = 'USA' you can use:

WHERE Country IN ('India', 'USA');

BETWEEN

SELECT *
FROM sales
WHERE Amount BETWEEN 2000 AND 4000;

LIKE

Find names beginning with A:

SELECT *
FROM sales
WHERE CustomerName LIKE 'A%';

4.4.5 Sorting Data

Ascending:

SELECT *
FROM sales
ORDER BY Amount;

Descending:

SELECT *
FROM sales
ORDER BY Amount DESC;

Multiple columns:

SELECT *
FROM sales
ORDER BY Country, Amount DESC;

4.4.6 Aggregations

Calculate total sales:

SELECT
    SUM(Amount) AS TotalSales
FROM sales;

Average:

SELECT
    AVG(Amount) AS AverageSales
FROM sales;

Maximum:

SELECT
    MAX(Amount) AS MaximumSale
FROM sales;

Minimum:

SELECT
    MIN(Amount) AS MinimumSale
FROM sales;

Count:

SELECT
    COUNT(*) AS TotalCustomers
FROM sales;

4.4.7 GROUP BY

Calculate sales by country:

SELECT
    Country,
    SUM(Amount) AS TotalSales
FROM sales
GROUP BY Country;

Result:

Multiple Aggregations

SELECT
    Country,
    COUNT(*) AS CustomerCount,
    SUM(Amount) AS TotalSales,
    AVG(Amount) AS AverageSales,
    MAX(Amount) AS MaximumSale
FROM sales
GROUP BY Country;

HAVING

HAVING filters aggregated results. Example:

SELECT
    Country,
    SUM(Amount) AS TotalSales
FROM sales
GROUP BY Country
HAVING SUM(Amount) > 5000;
This returns countries whose total sales exceed 5,000.

WHERE vs HAVING

Example:

SELECT
    Country,
    SUM(Amount) AS TotalSales
FROM sales
WHERE Amount > 1000
GROUP BY Country
HAVING SUM(Amount) > 5000;

4.4.8 Calculated Columns

You can create derived values in SQL.

SELECT
    CustomerID,
    CustomerName,
    Amount,

Amount * 0.10 AS Tax FROM sales; Calculate total amount:

SELECT
    CustomerID,
    Amount,

Amount * 1.10 AS TotalAmount FROM sales;

CASE Expression

Create customer categories:

SELECT
    CustomerName,
    Amount,
    CASE
        WHEN Amount >= 3000 THEN 'Premium'
        WHEN Amount >= 2000 THEN 'Standard'
        ELSE 'Basic'
    END AS CustomerType
FROM sales;

Result:

4.4.9 DISTINCT

Find unique countries:

SELECT DISTINCT Country
FROM sales;

Count unique countries:

SELECT COUNT(DISTINCT Country) AS CountryCount
FROM sales;

4.4.10 NULL Handling

Find NULL values:

SELECT *
FROM sales
WHERE Country IS NULL;

Find non-NULL values:

SELECT *
FROM sales
WHERE Country IS NOT NULL;

Use COALESCE:

SELECT
    CustomerName,
    COALESCE(Country, 'Unknown') AS Country
FROM sales;

4.4.11 String Functions

Convert to uppercase:

SELECT
    UPPER(CustomerName) AS CustomerName
FROM sales;

Lowercase:

SELECT
    LOWER(CustomerName) AS CustomerName
FROM sales;

Remove spaces:

SELECT
    TRIM(CustomerName) AS CustomerName
FROM sales;

String length:

SELECT
    CustomerName,
    LENGTH(CustomerName) AS NameLength
FROM sales;

4.4.12 Date Functions

Suppose we have: SaleDate Convert it to a date:

SELECT
    CAST(SaleDate AS DATE) AS SaleDate
FROM sales;

Extract year:

SELECT
    YEAR(SaleDate) AS SaleYear
FROM sales;

Extract month:

SELECT
    MONTH(SaleDate) AS SaleMonth
FROM sales;

4.4.13 JOIN Operations

Suppose we have: Customers CustomerID CustomerName Country Orders OrderID CustomerID Amount Create two temporary views:

customers.createOrReplaceTempView("customers")
orders.createOrReplaceTempView("orders")

Now perform an INNER JOIN:

SELECT
    o.OrderID,
    c.CustomerName,
    c.Country,

o.Amount FROM orders o INNER JOIN customers c ON o.CustomerID = c.CustomerID;

LEFT JOIN

SELECT
    c.CustomerID,
    c.CustomerName,
    o.OrderID,

o.Amount FROM customers c LEFT JOIN orders o ON c.CustomerID = o.CustomerID; This keeps all customers even if they have no orders.

Multiple Joins

SELECT
    o.OrderID,
    c.CustomerName,
    p.ProductName,

o.Amount FROM orders o JOIN customers c ON o.CustomerID = c.CustomerID JOIN products p ON o.ProductID = p.ProductID;

4.4.14 Subqueries

Example:

SELECT *
FROM sales
WHERE Amount > (
    SELECT AVG(Amount)
    FROM sales
);

This returns customers whose sales are above the average.

Common Table Expressions (CTEs)

CTEs make complex queries easier to read.

WITH CountrySales AS
(
    SELECT
        Country,
        SUM(Amount) AS TotalSales
    FROM sales
    GROUP BY Country
)
SELECT *
FROM CountrySales
WHERE TotalSales > 5000;

4.4.15 Window Functions

Window functions are extremely important in data engineering. They perform calculations across related rows without collapsing them into one row. Example:

SELECT
    CustomerName,
    Country,
    Amount,
    SUM(Amount) OVER (
        PARTITION BY Country
    ) AS CountryTotal
FROM sales;

Result:

ROW_NUMBER()

Assign a sequence number:

SELECT
    CustomerName,
    Country,
    Amount,
    ROW_NUMBER() OVER (
        PARTITION BY Country
        ORDER BY Amount DESC
    ) AS RowNum
FROM sales;

This is useful for finding the highest-selling customer within each country.

RANK()

SELECT
    CustomerName,
    Country,
    Amount,
    RANK() OVER (
        PARTITION BY Country
        ORDER BY Amount DESC
    ) AS SalesRank
FROM sales;

Top Customer Per Country

Using ROW_NUMBER():

WITH RankedSales AS
(
    SELECT
        CustomerName,
        Country,
        Amount,
        ROW_NUMBER() OVER (
            PARTITION BY Country
            ORDER BY Amount DESC
        ) AS rn
    FROM sales
)
SELECT
    CustomerName,
    Country,

Amount FROM RankedSales WHERE rn = 1; This is a very common data engineering pattern.

4.4.16 Querying Delta Tables

Suppose a Delta table called Sales exists in the Lakehouse. You can query it:

SELECT *
FROM Sales;

Aggregation:

SELECT
    Country,
    SUM(Amount) AS TotalSales
FROM Sales
GROUP BY Country;

Creating a Delta Table with SQL

You can create a table:

CREATE TABLE SalesDelta
USING DELTA
AS
SELECT *
FROM sales;

The exact table behavior and catalog semantics depend on the Fabric Lakehouse context.

Inserting Data

INSERT INTO SalesDelta
SELECT *
FROM sales;

Updating Delta Data

For Delta tables, updates can be performed with SQL:

UPDATE SalesDelta
SET Amount = Amount * 1.10
WHERE Country = 'India';

Deleting Data

DELETE FROM SalesDelta
WHERE Amount <= 0;

MERGE in Spark SQL
MERGE is extremely important for incremental pipelines.

Suppose: Target: Sales

Source: SalesStage Use:

MERGE INTO Sales AS target
USING SalesStage AS source
ON target.SaleID = source.SaleID

WHEN MATCHED THEN

    UPDATE SET *

WHEN NOT MATCHED THEN

    INSERT *;

This performs:

Existing SaleID → UPDATE
New SaleID      → INSERT

Spark SQL and Incremental Loading

Suppose the source contains: ModifiedDate Use:

SELECT *
FROM SourceSales
WHERE ModifiedDate > '2026-08-21 00:00:00';

Then merge:

Source
  │
  ▼
Incremental Data
  │
  ▼
SalesStage
  │
  ▼

MERGE

  │
  ▼
Sales Delta Table

4.4.17 Creating Views

Temporary view:

df.createOrReplaceTempView("sales")

Then:

SELECT *
FROM sales;

You can also create a SQL view over a table when supported by the Lakehouse SQL environment:

CREATE VIEW HighValueSales AS
SELECT *
FROM Sales
WHERE Amount > 3000;

4.4.18 Spark SQL vs PySpark

The same operation can often be written in two ways. PySpark

df.filter(
    col("Amount") > 3000
).select(
    "CustomerName",
    "Amount"
)

Spark SQL

SELECT
    CustomerName,

Amount FROM sales WHERE Amount > 3000;

Comparison

Both are executed by Spark's distributed processing engine.

4.4.19 Query Execution

Consider:

SELECT
    Country,
    SUM(Amount)
FROM sales
GROUP BY Country;

Conceptually:

SQL Query
    │
    ▼
Spark SQL Parser
    │
    ▼
Logical Plan
    │
    ▼
Optimized Plan
    │
    ▼
Physical Plan
    │
    ▼
Spark Jobs
    │
    ▼
Stages
    │
    ▼
Tasks
    │
    ▼
Executors

Spark's optimizer can optimize the execution plan before processing the data.

Catalyst Optimizer

Spark SQL uses the Catalyst Optimizer to optimize query execution plans. It can perform optimizations such as:

• Predicate pushdown
• Column pruning
• Expression optimization
• Join optimization
• Logical plan optimization

Example:

SELECT

CustomerID FROM Sales WHERE Amount > 1000; Spark can avoid processing unnecessary columns when possible.

Predicate Pushdown

Suppose a table has: 100 million records Query:

SELECT *
FROM Sales
WHERE Country = 'India';

Spark can push the filter closer to the data source when supported.

100M Records
     │
     ▼
Filter Early
     │
     ▼
Relevant Records

This reduces unnecessary processing.

Column Pruning

If the table has 50 columns but the query requires only:

SELECT
    CustomerID,

Amount FROM Sales; Spark can avoid reading unnecessary columns where the storage format supports it. This is especially beneficial with columnar formats such as Parquet and Delta.

4.4.20 Explain Query Plan

You can inspect the execution plan:

EXPLAIN
SELECT
    Country,
    SUM(Amount)
FROM sales
GROUP BY Country;

Or from PySpark:

df.groupBy("Country").sum("Amount").explain()

This helps identify:

• Scans
• Filters
• Joins
• Aggregations
• Exchanges/shuffles
• Execution strategies

Performance Best Practices

1. Filter Early

SELECT *
FROM Sales
WHERE SaleDate >= '2026-08-01';

2. Select Required Columns

Avoid:

SELECT *
FROM Sales;
when you need only a few columns.

Prefer:

SELECT
    SaleID,
    CustomerID,

Amount FROM Sales;

3. Be Careful with Large Joins

Large joins can cause significant shuffle.

4. Avoid Unnecessary DISTINCT

SELECT DISTINCT *
FROM Sales;

can be expensive on large datasets. Use DISTINCT only when required.

5. Use Appropriate Partitioning

Partition large datasets according to useful query patterns, but avoid creating excessive small partitions.

Real-World Example

Suppose your Lakehouse contains:

SalesLakehouse
│
├── Customers
├── Products
├── Orders
└── OrderItems

You can create an analytical query:

SELECT
    c.Country,
    COUNT(DISTINCT o.OrderID) AS TotalOrders,
    SUM(o.Amount) AS TotalSales,
    AVG(o.Amount) AS AverageOrderValue
FROM Orders o
JOIN Customers c
    ON o.CustomerID = c.CustomerID
GROUP BY c.Country
ORDER BY TotalSales DESC;

This produces a country-level sales summary.

End-to-End Fabric Notebook Example

Cell 1 — Create DataFrame

data = [
    (1, "John", "USA", 2500),
    (2, "Anita", "India", 3200),
    (3, "David", "UK", 1800),
    (4, "Priya", "India", 4100),
    (5, "Mike", "USA", 2900)
]

df = spark.createDataFrame(
    data,
    ["CustomerID", "CustomerName", "Country", "Amount"]
)

Cell 2 — Create View

df.createOrReplaceTempView("sales")

Cell 3 — Query

SELECT
    Country,
    COUNT(*) AS Customers,
    SUM(Amount) AS TotalSales,
    AVG(Amount) AS AverageSales
FROM sales
GROUP BY Country
ORDER BY TotalSales DESC;

Cell 4 — Create Customer Classification

SELECT
    CustomerID,
    CustomerName,
    Amount,
    CASE
        WHEN Amount >= 3000 THEN 'Premium'
        ELSE 'Standard'
    END AS CustomerType
FROM sales;

Hands-On Lab

Project: Sales Analytics Using Spark SQL Dataset Create a Sales DataFrame with: SaleID CustomerID ProductID SaleDate Country Quantity Amount Tasks Task 1 — Create DataFrame

df = spark.createDataFrame(...)

Task 2 — Create Temporary View

df.createOrReplaceTempView("sales")

Task 3 — Basic Query

SELECT *
FROM sales;

Task 4 — Filter

SELECT *
FROM sales
WHERE Amount > 1000;

Task 5 — Aggregate

SELECT
    Country,
    SUM(Amount) AS TotalSales
FROM sales
GROUP BY Country;

Task 6 — Rank Find the highest-value sale in each country:

WITH RankedSales AS
(
    SELECT
        *,
        ROW_NUMBER() OVER (
            PARTITION BY Country
            ORDER BY Amount DESC
        ) AS rn
    FROM sales
)
SELECT *
FROM RankedSales
WHERE rn = 1;

Task 7 — Write Delta

CREATE TABLE SalesDelta
USING DELTA
AS
SELECT *
FROM sales;

Interview Questions

df.createOrReplaceTempView("sales")

SELECT *
FROM sales;

A shuffle redistributes data between partitions, commonly caused by GROUP BY, joins, and sorting operations.

Yes. In appropriate Delta/Lakehouse contexts, Spark SQL supports operations such as INSERT, UPDATE, DELETE, and MERGE.

Lakehouse / OneLake
        │
        ▼
   Fabric Notebook
        │
        ▼
    Spark SQL
        │
   ┌────┼────┐
   │    │    │
Filter Join Aggregate
   │    │    │
   └────┼────┘
        ▼
   Delta Table
        │
        ▼
     Power BI

Learning Objectives

By the end of this lesson, you will be able to:

• Understand what a Spark DataFrame is.
• Create DataFrames using PySpark.
• Understand DataFrame schemas.
• Select, filter, and transform columns.
• Add, rename, and remove columns.
• Sort and aggregate data.
• Join DataFrames.
• Handle NULLs and duplicates.
• Read and write DataFrames.
• Understand DataFrame transformations and actions.
• Work with DataFrames in Microsoft Fabric Lakehouses.

4.5.1 What is a DataFrame?

A DataFrame is a distributed collection of structured data organized into rows and named columns. It is conceptually similar to:

• A SQL table
• A relational database table
• A pandas DataFrame

But a Spark DataFrame is distributed across Spark executors and is designed to process large datasets.

                  Spark DataFrame
                        │
              ┌─────────┴─────────┐
              │                   │
          Partition 1         Partition 2
              │                   │
         Executor 1           Executor 2

Example:

DataFrame vs SQL Table

A DataFrame: df is an object in a Spark application. A table: Sales is a persistent database/Lakehouse object. You can create a DataFrame from a table:

df = spark.read.table("Sales")
And you can write a DataFrame as a Delta table:
df.write \
    .format("delta") \
    .mode("overwrite") \
    .saveAsTable("Sales")

4.5.2 DataFrames in Microsoft Fabric

In Fabric, a common architecture is:

             OneLake
                │
                ▼
            Lakehouse
                │
         ┌──────┴──────┐
         │             │
       Files         Tables
         │             │
         └──────┬──────┘
                ▼
          Fabric Notebook
                │
             PySpark
                │
            DataFrame
                │
         ┌──────┼──────┐
         ▼      ▼      ▼
       Clean Transform Join
                │
                ▼
           Delta Table

4.5.3 Creating a DataFrame

Create sample data:

data = [
    (1, "John", "USA", 2500),
    (2, "Anita", "India", 3200),
    (3, "David", "UK", 1800),
    (4, "Priya", "India", 4100)
]

Define column names:

columns = [
    "CustomerID",
    "CustomerName",
    "Country",
    "Amount"
]

Create DataFrame:

df = spark.createDataFrame(
    data,

columns

)

Display:

display(df)

4.5.4 Inspecting a DataFrame

Display Data In Fabric:

display(df)
Or:
df.show()

Show First 10 Rows

df.show(10)

Count Rows

df.count()

Get Columns

df.columns

Example: [ 'CustomerID', 'CustomerName', 'Country', 'Amount'

]

Print Schema

df.printSchema()

Example: root |-- CustomerID: long |-- CustomerName: string |-- Country: string |-- Amount: long

4.5.5 Understanding Schema

A DataFrame schema defines:

• Column name
• Data type
• Nullable property

Example:

CustomerID → long
CustomerName → string
Country → string
Amount → long

Schema is extremely important in production data engineering.

Common Spark Data Types

4.5.6 Creating a DataFrame with Explicit Schema

For production pipelines, explicitly defining the schema can provide better control.

from pyspark.sql.types import (
    StructType,
    StructField,
    IntegerType,
    StringType,

DoubleType

)

Create schema:

schema = StructType([
    StructField(
        "CustomerID",
        IntegerType(),

True

    ),
    StructField(
        "CustomerName",
        StringType(),

True

    ),
    StructField(
        "Country",
        StringType(),

True

    ),
    StructField(
        "Amount",
        DoubleType(),

True

    )
])

Create DataFrame:

df = spark.createDataFrame(
    data,

schema

)

4.5.7 Selecting Columns

Select one column:

df.select("CustomerName").show()

Multiple columns:

df.select(
    "CustomerID",
    "CustomerName",
    "Amount"
).show()
Using col():
from pyspark.sql.functions import col

df.select(
    col("CustomerID"),
    col("Amount")
).show()

Selecting All Columns Except One

You can use:

df.drop("Country")

to create a DataFrame without the Country column.

4.5.8 Filtering Data

Find customers from India:

india_df = df.filter(
    col("Country") == "India"
)

display(india_df)

You can also use:

india_df = df.where(
    col("Country") == "India"
)

Multiple Conditions

result = df.filter(
    (col("Country") == "India") &
    (col("Amount") > 3000)
)

display(result)
AND

& OR | Example:

result = df.filter(
    (col("Country") == "India") |
    (col("Country") == "USA")
)

4.5.9 Adding Columns

Use withColumn().

from pyspark.sql.functions import col

df2 = df.withColumn(
    "Tax",
    col("Amount") * 0.10
)

display(df2)

Add total amount:

df2 = df2.withColumn(
    "TotalAmount",
    col("Amount") + col("Tax")
)

4.5.10 Conditional Columns

Use when().

from pyspark.sql.functions import when

df2 = df.withColumn(
    "CustomerType",
    when(
        col("Amount") >= 3000,
        "Premium"
    ).otherwise("Standard")
)

display(df2)

Result:

Multiple Conditions

df2 = df.withColumn(
    "CustomerType",
    when(
        col("Amount") >= 4000,
        "Platinum"
    )
    .when(
        col("Amount") >= 3000,
        "Premium"
    )
    .when(
        col("Amount") >= 2000,
        "Standard"
    )
    .otherwise("Basic")
)

4.5.11 Renaming Columns

Use:

df2 = df.withColumnRenamed(
    "CustomerName",
    "Customer_Name"
)

Multiple columns:

df2 = df \
    .withColumnRenamed(
        "CustomerID",
        "Customer_ID"
    ) \
    .withColumnRenamed(
        "CustomerName",
        "Customer_Name"
    )

4.5.12 Dropping Columns

Drop one:

df2 = df.drop("Amount")

Drop multiple:

df2 = df.drop(
    "Amount",
    "Country"
)

4.5.13 Sorting Data

Ascending:

df.orderBy("Amount").show()

Descending:

from pyspark.sql.functions import desc

df.orderBy(
    desc("Amount")
).show()

Multiple sort columns:

df.orderBy(
    "Country",
    desc("Amount")
).show()

4.5.14 Removing Duplicates

Remove complete duplicate rows:

df2 = df.dropDuplicates()

Remove duplicates based on CustomerID:

df2 = df.dropDuplicates(
    ["CustomerID"]
)

This is particularly useful in ingestion pipelines.

4.5.15 Handling NULL Values

Check NULL values:

df.filter(
    col("Country").isNull()
).show()

Check non-NULL:

df.filter(
    col("Country").isNotNull()
).show()

Fill NULL Values

df2 = df.fillna({
    "Country": "Unknown"
})

Numeric NULL:

df2 = df.fillna({

"Amount": 0

})

Drop NULL Records

Remove rows containing NULL:

df2 = df.dropna()

Only consider a specific column:

df2 = df.dropna(
    subset=["CustomerID"]
)

4.5.16 Data Type Conversion

Suppose Amount is a string. Convert it:

df2 = df.withColumn(
    "Amount",
    col("Amount").cast("double")
)

Check:

df2.printSchema()

4.5.17 String Transformations

Import functions:

from pyspark.sql.functions import (
    trim,
    upper,

lower

)

Trim spaces:

df2 = df.withColumn(
    "CustomerName",
    trim(col("CustomerName"))
)

Uppercase:

df2 = df.withColumn(
    "Country",
    upper(col("Country"))
)

Lowercase:

df2 = df.withColumn(
    "CustomerName",
    lower(col("CustomerName"))
)

Replace Text

Use regexp_replace():

from pyspark.sql.functions import regexp_replace

df2 = df.withColumn(
    "CustomerName",
    regexp_replace(
        col("CustomerName"),
        "John",
        "Jonathan"
    )
)

4.5.18 Aggregations

Calculate total:

from pyspark.sql.functions import sum

df.select(
    sum("Amount").alias("TotalSales")
).show()

Average:

from pyspark.sql.functions import avg

df.select(
    avg("Amount").alias("AverageSales")
).show()

Count:

from pyspark.sql.functions import count

df.select(
    count("*").alias("RecordCount")
).show()

4.5.19 GroupBy

Sales by country:

result = df.groupBy(
    "Country"
).agg(
    sum("Amount").alias("TotalSales")
)

display(result)

Multiple aggregations:

result = df.groupBy("Country").agg(
    count("*").alias("CustomerCount"),
    sum("Amount").alias("TotalSales"),
    avg("Amount").alias("AverageSales")
)

display(result)

4.5.20 Joining DataFrames

Suppose: Customer DataFrame CustomerID CustomerName Country Orders DataFrame OrderID CustomerID Amount Join:

result = orders.join(
    customers,
    orders.CustomerID == customers.CustomerID,
    "inner"
)

Select:

result = result.select(
    orders.OrderID,
    customers.CustomerName,
    customers.Country,

orders.Amount

)

display(result)

Types of Joins

# Inner
df1.join(df2, "CustomerID", "inner")

# Left
df1.join(df2, "CustomerID", "left")

# Right
df1.join(df2, "CustomerID", "right")

# Full
df1.join(df2, "CustomerID", "full")

Other join types include: left_semi left_anti cross

4.5.21 Reading CSV into DataFrame

df = spark.read \
    .option("header", "true") \
    .option("inferSchema", "true") \
    .csv("Files/Sales.csv")

Display:

display(df)

Reading Multiple CSV Files

df = spark.read \
    .option("header", "true") \
    .option("inferSchema", "true") \
    .csv("Files/Sales/")

Spark can read multiple matching files from the specified location.

4.5.22 Reading Parquet

df = spark.read.parquet(
    "Files/Sales.parquet"
)

4.5.23 Reading Delta

Read a Delta table:

df = spark.read \
    .format("delta") \
    .load("Tables/Sales")

If it is registered as a table:

df = spark.read.table("Sales")

4.5.24 Writing DataFrames

Write to Delta:

df.write \
    .format("delta") \
    .mode("overwrite") \
    .saveAsTable("Sales")

Append:

df.write \
    .format("delta") \
    .mode("append") \
    .saveAsTable("Sales")

Write Modes

4.5.25 DataFrame Transformations

DataFrame operations such as:

df.select(...)
df.filter(...)
df.withColumn(...)
df.join(...)
df.groupBy(...)

are transformations. Spark builds an execution plan rather than immediately processing all data.

DataFrame Actions

Actions trigger execution:

df.show()
df.count()
df.collect()
df.first()
df.write.save(...)

Example

filtered_df = df.filter(
    col("Amount") > 3000
)

No result needs to be computed immediately. Then:

filtered_df.count()

triggers execution.

4.5.26 DataFrame Lineage

Spark maintains the logical lineage of transformations. Example:

CSV
 │
 ▼
Read
 │
 ▼
DataFrame
 │
 ▼
Filter
 │
 ▼
Clean
 │
 ▼
Join
 │
 ▼
Aggregate
 │
 ▼
Write Delta

This allows Spark to build an optimized execution plan.

4.5.27 Explain Execution Plan

You can inspect the plan:

df.explain()

For more details:

df.explain(True)

This can help identify:

• Filters
• Scans
• Joins
• Aggregations
• Shuffles
• Execution strategies

4.5.28 DataFrame vs Pandas DataFrame

This is an important interview topic.

Example:

import pandas as pd

creates a Pandas DataFrame. Whereas:

df = spark.createDataFrame(data)

creates a Spark DataFrame.

4.5.29 DataFrame vs RDD

RDD stands for Resilient Distributed Dataset.

For modern Fabric data engineering, DataFrames are generally preferred for structured data processing.

4.5.30 Performance Considerations

Select Only Required Columns Instead of:

df.select("*")

prefer:

df.select(
    "CustomerID",
    "Amount"
)
when only those columns are needed.

Filter Early

df.filter(
    col("Amount") > 1000
)

Reducing data early can reduce downstream processing.

Avoid collect() on Large DataFrames

Avoid:

df.collect()

for millions of rows. Prefer:

df.limit(100).show()

for inspection.

Be Careful with Large Joins

Large joins may cause expensive shuffles.

Avoid Unnecessary distinct()

df.distinct()

can require significant data movement on large datasets. Use it only when required.

4.5.31 Caching DataFrames

If the same expensive DataFrame is reused multiple times, you can consider:

df.cache()

Example:

df.cache()

df.filter(
    col("Country") == "India"
).count()

df.groupBy(
    "Country"
).sum("Amount")
Caching consumes resources, so it should be used when there is a real reuse benefit.

4.5.32 Repartitioning

You can change the number of partitions:

df2 = df.repartition(10)

Partition by a column:

df2 = df.repartition(
    "Country"
)
This can cause a shuffle, so it should be used intentionally.

Coalesce

coalesce() can reduce the number of partitions without a full shuffle in common cases:

df2 = df.coalesce(4)

A simplified comparison:

4.5.33 Practical Data Cleaning Pipeline

Suppose your source contains: CustomerID | CustomerName | Country | Amount 1 | " John " | india | "2500" 2 | " Anita " | India | "3200" 3 | NULL | UK | "1800" Clean it:

from pyspark.sql.functions import (
    col,
    trim,

upper

)

clean_df = df \
    .withColumn(
        "CustomerName",
        trim(col("CustomerName"))
    ) \
    .withColumn(
        "Country",
        upper(col("Country"))
    ) \
    .withColumn(
        "Amount",
        col("Amount").cast("double")
    ) \
    .dropDuplicates()

Handle NULLs:

clean_df = clean_df.fillna({
    "CustomerName": "Unknown",
    "Country": "UNKNOWN",

"Amount": 0

})

4.5.34 End-to-End Fabric Example

Source

OneLake
   │
   ▼
Files/Sales.csv
Read

df = spark.read \
    .option("header", "true") \
    .option("inferSchema", "true") \
    .csv("Files/Sales.csv")

Clean

from pyspark.sql.functions import (
    col,
    trim,

upper

)

clean_df = df \
    .dropDuplicates() \
    .withColumn(
        "CustomerName",
        trim(col("CustomerName"))
    ) \
    .withColumn(
        "Country",
        upper(col("Country"))
    ) \
    .withColumn(
        "Amount",
        col("Amount").cast("double")
    )

Transform

from pyspark.sql.functions import when

clean_df = clean_df.withColumn(
    "CustomerType",
    when(
        col("Amount") >= 3000,
        "Premium"
    ).otherwise("Standard")
)

Aggregate

country_sales = clean_df.groupBy(
    "Country"
).agg(
    sum("Amount").alias("TotalSales")
)

Write

clean_df.write \
    .format("delta") \
    .mode("overwrite") \
    .saveAsTable("Sales")

4.5.35 Complete DataFrame Pipeline

             CSV
              │
              ▼

        spark.read.csv()

              │
              ▼
          DataFrame
              │
       ┌──────┼──────┐
       ▼      ▼      ▼
    Filter   Clean  Cast
       │      │      │
       └──────┼──────┘
              ▼
           Join
              │
              ▼
         Aggregation
              │
              ▼
         Data Validation
              │
              ▼
        Delta DataFrame
              │
              ▼
        saveAsTable()
              │
              ▼
        Lakehouse Table

Hands-On Lab

Project: Customer Sales DataFrame Processing Input Upload: Sales.csv with columns: SaleID CustomerID CustomerName Country SaleDate Quantity Amount Step 1 — Read CSV

df = spark.read \
    .option("header", "true") \
    .option("inferSchema", "true") \
    .csv("Files/Sales.csv")

Step 2 — Inspect

display(df)
df.printSchema()

Step 3 — Remove Duplicates

df = df.dropDuplicates(
    ["SaleID"]
)

Step 4 — Clean Names

df = df.withColumn(
    "CustomerName",
    trim(col("CustomerName"))
)

Step 5 — Standardize Country

df = df.withColumn(
    "Country",
    upper(col("Country"))
)

Step 6 — Convert Amount

df = df.withColumn(
    "Amount",
    col("Amount").cast("double")
)

Step 7 — Create Customer Type

df = df.withColumn(
    "CustomerType",
    when(
        col("Amount") >= 3000,
        "Premium"
    ).otherwise("Standard")
)

Step 8 — Aggregate

country_sales = df.groupBy(
    "Country"
).agg(
    sum("Amount").alias("TotalSales")
)

display(country_sales)

Step 9 — Write Delta

df.write \
    .format("delta") \
    .mode("overwrite") \
    .saveAsTable("Sales")

Interview Questions

df = spark.createDataFrame(
    data,

)

df.filter(
    col("Amount") > 1000
)

df.dropDuplicates()
or:
df.dropDuplicates(["CustomerID"])

df.fillna(...)
or:
df.dropna(...)

CREATE

   ↓

SELECT

   ↓
FILTER
   ↓
WITHCOLUMN
   ↓
JOIN
   ↓
GROUPBY
   ↓
SORT
   ↓
VALIDATE
   ↓
WRITE

OneLake
│
▼
Lakehouse Files / Tables
│
▼
Fabric Notebook
│
▼
PySpark DataFrame
│
├── Read
├── Clean
├── Transform
├── Join
├── Aggregate
└── Validate
│
▼
Delta Table
│
▼
Power BI / Analytics

Learning Objectives

By the end of this lesson, you will be able to:

• Understand Delta Lake and Delta tables.
• Understand why Delta is important in Microsoft Fabric.
• Create Delta tables using PySpark and Spark SQL.
• Read and write Delta data.
• Perform INSERT, UPDATE, and DELETE.
• Use MERGE for upserts.
• Understand ACID transactions.
• Understand schema enforcement and schema evolution.
• Understand Delta transaction logs and table versions.
• Use Delta tables for incremental data pipelines.

4.6.1 What is a Delta Table?

A Delta table is a table stored using the Delta Lake storage format. Delta Lake builds additional data-management capabilities on top of Parquet files.

                  Delta Table
                       │
          ┌────────────┴────────────┐
          │                         │
      Parquet Data              _delta_log
       Files                   Transaction Log

The combination of: Parquet files + Delta transaction log provides capabilities that plain Parquet files do not provide by themselves.

Why Delta Tables?

Consider a normal Parquet dataset:

Sales
│
├── part-0001.parquet
├── part-0002.parquet
└── part-0003.parquet

Suppose you need to update one customer's record. With plain files, reliably managing updates and concurrent changes can become complicated. Delta adds a transaction log:

Sales
│
├── part-0001.parquet
├── part-0002.parquet
├── part-0003.parquet
│
└── _delta_log
      ├── 00000000000000000000.json
      ├── 00000000000000000001.json
      └── ...

The transaction log records changes to the table.

Delta Tables in Microsoft Fabric

Delta is central to Fabric Lakehouse data engineering. Typical architecture:

                 OneLake
                    │
                Lakehouse
                    │
              Delta Tables
                    │
        ┌───────────┼───────────┐
        ▼           ▼           ▼
      Spark        SQL        Power BI

For example:

SalesLakehouse
│
├── Tables
│    ├── Customers
│    ├── Products
│    ├── Orders
│    └── Sales
│
└── Files

Lakehouse tables are commonly stored using Delta format.

4.6.2 Delta vs Parquet

4.6.3 Delta Table Architecture

A Delta table consists primarily of:

                Delta Table
                    │
          ┌─────────┴─────────┐
          ▼                   ▼
     Data Files          Transaction Log
     Parquet                _delta_log
          │                   │
          │                   ├── Add files
          │                   ├── Remove files
          │                   ├── Metadata
          │                   └── Table versions

4.6.4 Transaction Log

The _delta_log is one of the most important Delta concepts. It records changes made to the table. Conceptually:

_version 0
    │
    ▼
Initial table

_version 1
    │
    ▼
New data added

_version 2
    │
    ▼
Records updated

_version 3
    │
    ▼
Records deleted

This allows Delta to understand the state of the table at different versions.

4.6.5 ACID Transactions

Delta provides ACID transaction capabilities. ACID stands for:

• Atomicity
• Consistency
• Isolation
• Durability

These properties help ensure reliable data operations.

Atomicity

An operation should complete successfully as a unit or not be partially committed.

Transaction
    │
    ├── Step 1 ✓
    ├── Step 2 ✓
    └── Step 3 ✓
          │
          ▼
       COMMIT

Consistency

The table should remain in a valid state after a transaction.

Isolation

Concurrent operations should not incorrectly interfere with one another.

Durability

Once a transaction is committed, the committed state is persisted.

4.6.6 Creating a Delta Table Using PySpark

Create sample data:

data = [
    (1, "John", "USA", 2500),
    (2, "Anita", "India", 3200),
    (3, "David", "UK", 1800),
    (4, "Priya", "India", 4100)
]

columns = [
    "CustomerID",
    "CustomerName",
    "Country",
    "Amount"
]

df = spark.createDataFrame(
    data,

columns

)

Write as Delta:

df.write \
    .format("delta") \
    .mode("overwrite") \
    .saveAsTable("Sales")

This creates a Delta table called: Sales

4.6.7 Reading a Delta Table

Using Spark:

df = spark.read \
    .format("delta") \
    .table("Sales")
Or:
df = spark.read.table("Sales")

Display:

display(df)

4.6.8 Querying a Delta Table with SQL

SELECT *
FROM Sales;

Filter:

SELECT *
FROM Sales
WHERE Country = 'India';

Aggregate:

SELECT
    Country,
    SUM(Amount) AS TotalSales
FROM Sales
GROUP BY Country;

4.6.9 Creating Delta Tables Using SQL

You can create a Delta table using Spark SQL:

CREATE TABLE SalesDelta
USING DELTA
AS
SELECT *
FROM Sales;

You can also define a schema explicitly:

CREATE TABLE CustomerSales
(
    CustomerID INT,
    CustomerName STRING,
    Country STRING,

Amount DOUBLE

)
USING DELTA;

4.6.10 INSERT

Insert records:

INSERT INTO CustomerSales
VALUES
    (1, 'John', 'USA', 2500),
    (2, 'Anita', 'India', 3200);
Or insert from another table:
INSERT INTO CustomerSales
SELECT *
FROM SalesStage;

4.6.11 UPDATE

Delta supports updates. Example:

UPDATE CustomerSales
SET Amount = 3500
WHERE CustomerID = 2;

Before: CustomerID = 2 Amount = 3200 After: CustomerID = 2 Amount = 3500

4.6.12 DELETE

Delete records:

DELETE FROM CustomerSales
WHERE CustomerID = 2;

You can also delete based on a condition:

DELETE FROM CustomerSales
WHERE Amount <= 0;

4.6.13 MERGE

MERGE is one of the most important Delta features for data engineering.

It is used to implement upsert logic. Upsert means:

UPDATE existing records

+

INSERT new records

Suppose: Target: CustomerSales

Source: SalesStage Use:

MERGE INTO CustomerSales AS target
USING SalesStage AS source
ON target.CustomerID = source.CustomerID

WHEN MATCHED THEN

    UPDATE SET *

WHEN NOT MATCHED THEN

    INSERT *;

MERGE Architecture

             Source Data
                  │
                  ▼
             SalesStage
                  │
                  ▼

               MERGE

/ \ / \

      Existing       New
          │             │
          ▼             ▼

        UPDATE        INSERT

\ /

              \       /
               ▼     ▼
              Delta Table

4.6.14 Why MERGE is Important

Imagine your source system sends: CustomerID Amount 101 5000 102 7000 103 2500 Target already contains: CustomerID Amount 101 4500 102 7000 After MERGE:

101 → UPDATE 4500 → 5000
102 → UPDATE 7000 → 7000
103 → INSERT 2500

Final:

This is fundamental to incremental data engineering.

4.6.15 Schema Enforcement

Delta can help prevent incompatible data from being written to a table schema. For example, suppose:

Amount → DOUBLE

but incoming data attempts to write an incompatible value. Schema validation can identify the mismatch instead of silently accepting an invalid table structure.

4.6.16 Schema Evolution

Sometimes new columns need to be added. Existing table: CustomerID CustomerName Amount New source: CustomerID CustomerName Amount Country Schema evolution allows the table schema to be expanded when appropriately configured. In PySpark, an example is:

df.write \
    .format("delta") \
    .option("mergeSchema", "true") \
    .mode("append") \
    .saveAsTable("CustomerSales")

Use schema evolution deliberately in production; uncontrolled schema changes can create downstream issues.

4.6.17 Delta Time Travel

Delta maintains table versions. Conceptually:

Version 0
   │
   ▼
Version 1
   │
   ▼
Version 2
   │
   ▼
Version 3

This allows historical table states to be queried where supported. For example:

SELECT *
FROM Sales VERSION AS OF 2;

This can be useful for:

• Auditing
• Debugging
• Recovering from incorrect changes
• Comparing table versions

4.6.18 Table History

You can inspect Delta table history using Spark SQL:

DESCRIBE HISTORY Sales;

The history can provide information such as:

• Version
• Timestamp
• Operation
• Operation parameters
• User/context information

4.6.19 Restore

Delta supports mechanisms for restoring a table to an earlier state in environments where the corresponding Delta operation is supported. Conceptually:

Current Version
      │
      ▼
Incorrect Update
      │
      ▼
Identify Previous Version
      │
      ▼
Restore
      │
      ▼
Correct Table State

Always consider retention and operational policies before relying on historical versions for recovery.

4.6.20 Delta Table Maintenance

Over time, Delta tables can accumulate files. For example:

Sales
│
├── old-file-1.parquet
├── old-file-2.parquet
├── new-file-1.parquet
└── new-file-2.parquet

Maintenance operations can optimize storage and query performance. Important concepts include:

• Compaction
• File optimization
• Vacuum
• Small-file management

The exact commands and maintenance capabilities depend on the Fabric/Lakehouse environment and Delta runtime version.

Small Files Problem

Suppose a pipeline writes: 1 million records but creates: 50,000 tiny files This can create overhead when reading the data.

Many Small Files
       │
       ▼
More File Metadata
       │
       ▼
Slower Queries

Good ingestion design should avoid excessive small-file creation.

4.6.21 Partitioning Delta Tables

Delta tables can be partitioned. Example:

df.write \
    .format("delta") \
    .partitionBy("Country") \
    .mode("overwrite") \
    .saveAsTable("Sales")

This creates a layout conceptually like:

Sales
│
├── Country=India
├── Country=USA
└── Country=UK

Partitioning can help when queries frequently filter on a suitable partition column. However, avoid partitioning on columns with extremely high cardinality.

4.6.22 Delta and Incremental Loads

Delta is excellent for incremental data engineering. Example:

Source System
      │
      ▼
New / Changed Records
      │
      ▼
Staging Delta
      │
      ▼

MERGE

      │
      ▼
Production Delta

Watermark-Based Incremental Processing

Suppose your source contains: ModifiedDate Get records after the previous watermark:

SELECT *
FROM SourceSales
WHERE ModifiedDate > '2026-08-21 00:00:00';

Load them into staging:

Source
  ↓
Incremental Records
  ↓
SalesStage

Then:

MERGE INTO Sales AS target
USING SalesStage AS source
ON target.SaleID = source.SaleID

WHEN MATCHED THEN

    UPDATE SET *

WHEN NOT MATCHED THEN

    INSERT *;

4.6.23 Slowly Changing Dimensions

Delta tables can also be used to implement Slowly Changing Dimensions (SCD). For example, Customer 101 changes country: CustomerID = 101

Old: Country = India

New: Country = USA An SCD Type 2 implementation could preserve history: CustomerID | Country | StartDate | EndDate | Current 101 | India | 2025-01-01 | 2026-08-01 | N 101 | USA | 2026-08-02 | NULL | Y Delta MERGE is commonly used as part of SCD processing.

4.6.24 Delta Table Security and Governance

In a production Fabric environment, Delta tables should be managed with appropriate:

• Workspace permissions
• Lakehouse permissions
• Data access controls
• Governance policies
• Data lineage
• Auditing

Delta storage itself is not a replacement for access control.

4.6.25 Delta Table Lifecycle

A typical production lifecycle is:

Source
  │
  ▼
Raw Data
  │
  ▼
Bronze Delta
  │
  ▼
Cleaning
  │
  ▼
Silver Delta
  │
  ▼
Business Transformations
  │
  ▼
Gold Delta
  │
  ▼
Power BI

Medallion Architecture

Delta tables are commonly used in a Medallion Architecture. Bronze Raw data:

API / SQL / Files
       ↓
Bronze Delta
Silver

Cleaned and standardized:

Bronze
  ↓
Clean
  ↓
Silver Delta
Gold

Business-ready:

Silver
  ↓
Business Logic
  ↓
Gold Delta

4.6.26 PySpark Delta Operations

Write

df.write \
    .format("delta") \
    .mode("overwrite") \
    .saveAsTable("Sales")

Read

df = spark.read.table("Sales")

Append

df.write \
    .format("delta") \
    .mode("append") \
    .saveAsTable("Sales")

DeltaTable API

PySpark can also use the Delta Lake API.

from delta.tables import DeltaTable

target = DeltaTable.forName(
    spark,
    "Sales"
)

Then perform a merge:

target.alias("target") \
    .merge(
        source_df.alias("source"),
        "target.SaleID = source.SaleID"
    ) \
    .whenMatchedUpdateAll() \
    .whenNotMatchedInsertAll() \
    .execute()

4.6.27 Delta Table Example

Suppose the source provides: Sales.csv Columns: SaleID CustomerID ProductID SaleDate Quantity Amount Pipeline:

Sales.csv
    │
    ▼
Fabric Notebook
    │
    ▼
Spark DataFrame
    │
    ▼
Clean Data
    │
    ▼
Bronze Delta
    │
    ▼
Transform
    │
    ▼
Silver Delta
    │
    ▼

MERGE

    │
    ▼
Gold Delta

4.6.28 Practical Example — Incremental MERGE

Target Sales Incoming Data SalesStage Python

from delta.tables import DeltaTable

target = DeltaTable.forName(
    spark,
    "Sales"
)

target.alias("target") \
    .merge(
        sales_stage.alias("source"),
        "target.SaleID = source.SaleID"
    ) \
    .whenMatchedUpdateAll() \
    .whenNotMatchedInsertAll() \
    .execute()

4.6.29 Delta Table Monitoring

For production Delta tables, monitor:

• Number of records
• Table size
• Number of files
• Number of small files
• Load duration
• MERGE duration
• Failed transactions
• Schema changes
• Data quality
• Table history

Example validation:

SELECT COUNT(*) AS RecordCount
FROM Sales;

Check duplicates:

SELECT
    SaleID,
    COUNT(*) AS RecordCount
FROM Sales
GROUP BY SaleID
HAVING COUNT(*) > 1;

Common Delta Problems

1. Duplicate Records

Possible cause: Incorrect MERGE key Always use a stable business key.

2. Schema Mismatch

Source: Amount = STRING Target: Amount = DOUBLE Solution: validate/cast the source schema before writing.

3. Too Many Small Files

Possible cause: Frequent small writes Solution:

• Batch writes appropriately.
• Optimize/compact when appropriate.
• Avoid unnecessarily frequent tiny transactions.

4. Incorrect MERGE Condition

Bad: ON target.CustomerName = source.CustomerName Names may not be unique. Better: ON target.CustomerID = source.CustomerID Use a stable business key.

Hands-On Lab

Project: Incremental Sales Delta Pipeline Source Sales.csv Requirements Create: SalesLakehouse and a Delta table: Sales Step 1 — Read CSV

df = spark.read \
    .option("header", "true") \
    .option("inferSchema", "true") \
    .csv("Files/Sales.csv")

Step 2 — Clean

df = df.dropDuplicates(
    ["SaleID"]
)

Step 3 — Write Initial Delta Table

df.write \
    .format("delta") \
    .mode("overwrite") \
    .saveAsTable("Sales")

Step 4 — Create New Batch Create a second DataFrame:

new_data = [
    (1, 101, 1001, 5000),
    (5, 105, 1005, 3500)
]

new_df = spark.createDataFrame(
    new_data,
    [
        "SaleID",
        "CustomerID",
        "ProductID",
        "Amount"
    ]
)

Step 5 — Merge

from delta.tables import DeltaTable

target = DeltaTable.forName(
    spark,
    "Sales"
)

target.alias("target") \
    .merge(
        new_df.alias("source"),
        "target.SaleID = source.SaleID"
    ) \
    .whenMatchedUpdateAll() \
    .whenNotMatchedInsertAll() \
    .execute()

Step 6 — Validate

SELECT *
FROM Sales
ORDER BY SaleID;

Interview Questions

MERGE allows you to perform conditional updates and inserts, making it ideal for upsert and incremental-loading scenarios.

                 Delta Table
                     │
        ┌────────────┴────────────┐
        ▼                         ▼
   Parquet Files             _delta_log
        │                         │
        └────────────┬────────────┘
                     ▼
              ACID Transactions
                     │
       ┌─────────────┼─────────────┐
       ▼             ▼             ▼

    INSERT         UPDATE        DELETE

                     │
                     ▼

                   MERGE

                     │
                     ▼
             Incremental Loads
Most Important Delta Concepts

SQL Server / API / Files
          │
          ▼
       Ingestion
          │
          ▼
      Bronze Delta
          │
          ▼
    Spark / PySpark
          │
          ▼
      Silver Delta
          │
          ▼

       MERGE / SCD

          │
          ▼
       Gold Delta
          │
          ▼
        Power BI

Learning Objectives

By the end of this lesson, you will be able to:

• Understand notebook utilities in Microsoft Fabric.
• Use notebookutils for common data-engineering tasks.
• Work with files and folders in OneLake.
• Pass parameters between notebooks.
• Run child notebooks.
• Manage notebook execution flow.
• Use utilities for orchestration and automation.
• Build reusable Fabric data-engineering notebooks.

4.7.1 What are Notebook Utilities?

Notebook Utilities are built-in capabilities available in Microsoft Fabric notebooks that help you perform tasks beyond normal PySpark processing. They are useful for:

• File management
• Folder management
• Notebook orchestration
• Parameter passing
• Environment interaction
• Pipeline integration
• Session-related operations

In Fabric, these capabilities are commonly accessed through:

notebookutils

Conceptually:

Fabric Notebook
      │
      ├── PySpark
      │
      ├── Spark SQL
      │
      └── notebookutils
             │
       ┌─────┼─────────┐
       ▼     ▼         ▼
     Files Notebooks Parameters

4.7.2 Why Notebook Utilities Are Important

A data-engineering notebook often needs to do more than transform data. For example: 1 Find today's files 2 Read the files 3 Transform data 4 Write Delta table 5 Call another notebook 6 Pass parameters 7 Archive the source file 8 Return a result PySpark handles the transformation, while notebook utilities can help with the orchestration and file-management parts.

4.7.3 notebookutils

In Fabric, you can use:

notebookutils

For example:

print(notebookutils)

The exact available utilities can depend on the Fabric runtime and notebook context.

4.7.4 File Utilities

Notebook utilities can be used to work with files in supported storage locations. A common pattern is:

notebookutils.fs

For example:

notebookutils.fs.ls("Files")

This can be used to inspect files and folders available to the notebook.

Listing Files

Example:

files = notebookutils.fs.ls("Files")

for file in files:
    print(file.name)

Conceptually:

Files/
│
├── Sales.csv
├── Customers.csv
├── Products.csv
└── Orders.csv

The notebook can inspect the contents before processing them.

4.7.5 Checking File Information

When listing files, the returned file information can include properties such as:

• Name
• Path
• Size
• Whether it is a directory

Example:

files = notebookutils.fs.ls("Files")

for file in files:
    print(
        file.name,
        file.path,

file.size

    )

This can be useful for ingestion validation.

4.7.6 Creating Directories

You can create a directory using the filesystem utility:

notebookutils.fs.mkdirs(
    "Files/archive"
)

Conceptually:

Files/
│
├── Sales.csv
├── Customers.csv
│
└── archive/

This is useful when organizing raw files.

4.7.7 Copying Files

You can copy files using the filesystem utilities. Conceptually:

notebookutils.fs.cp(
    "Files/Sales.csv",
    "Files/archive/Sales.csv"
)

This is useful for:

Raw
 │
 ├── Processing
 │
 └── Archive

4.7.8 Moving Files

A common pattern is to move processed files into an archive location.

notebookutils.fs.mv(
    "Files/Sales.csv",
    "Files/archive/Sales.csv"
)

For example: Before:

Files/
└── Sales.csv

After:

Files/
└── archive/
    └── Sales.csv

This can prevent the same file from being processed repeatedly.

4.7.9 Deleting Files

You can remove a file:

notebookutils.fs.rm(
    "Files/Sales.csv"
)
For folders containing data, recursive deletion may be required depending on the operation.

Be careful with delete operations in production.

4.7.10 OneLake Paths

Fabric notebooks frequently work with Lakehouse paths. For example: Files/Sales/Sales.csv and: Tables/Sales Conceptually:

OneLake
   │
   ▼
Lakehouse
   │
   ├── Files
   │    ├── Raw
   │    ├── Processed
   │    └── Archive
   │
   └── Tables
        ├── Sales
        ├── Customers
        └── Products

4.7.11 Notebook Parameters

Parameters allow the same notebook to be reused for different inputs. Instead of hardcoding:

file_path = "Files/Sales.csv"

you can make the path configurable. Example:

file_path = "Files/Sales.csv"

df = spark.read \
    .option("header", "true") \
    .csv(file_path)

A parameterized notebook can instead receive: file_path = Files/Customers.csv or: file_path = Files/Orders.csv

Why Parameters Matter

Without parameters:

Notebook 1 → Sales
Notebook 2 → Customers
Notebook 3 → Orders

With parameters:

             Generic Notebook
                    │
       ┌────────────┼────────────┐
       ▼            ▼            ▼
    Sales.csv   Customers.csv  Orders.csv

This promotes reuse.

4.7.12 Parameter Cell

Fabric notebooks provide a way to designate a cell as a parameter cell. For example:

file_path = "Files/Sales.csv"
load_type = "incremental"

The notebook can then receive different parameter values when executed through supported orchestration mechanisms.

Example Parameterized Notebook

file_path = "Files/Sales.csv"
load_type = "full"

Then:

df = spark.read \
    .option("header", "true") \
    .option("inferSchema", "true") \
    .csv(file_path)

Now the same notebook can process another file by changing the parameters.

4.7.13 Running Another Notebook

Notebook utilities can be used to orchestrate notebooks. Conceptually:

notebookutils.notebook.run(
    "TransformSales",
    600,
    {
        "load_date": "2026-08-22"
    }
)

The exact method signature and timeout behavior should be verified against the Fabric runtime available in your workspace. The idea is:

Master Notebook
      │
      ├── Ingestion Notebook
      │
      ├── Cleaning Notebook
      │
      ├── Transformation Notebook
      │
      └── Validation Notebook

4.7.14 Parent and Child Notebooks

A common architecture is:

             Master Notebook
                    │
       ┌────────────┼────────────┐
       ▼            ▼            ▼
   Notebook A   Notebook B   Notebook C
   Ingestion     Cleaning     Validation

The parent notebook controls the workflow. The child notebooks perform individual tasks.

Why Use Child Notebooks?

Instead of creating one huge notebook: 10,000 lines split the workflow: 01_Ingestion 02_Cleansing 03_Transformation 04_Load 05_Validation Benefits:

• Reusability
• Easier debugging
• Easier testing
• Better organization
• Separation of responsibilities

4.7.15 Passing Parameters to Child Notebooks

Suppose the parent has:

load_date = "2026-08-22"

It can pass this to a child notebook. Conceptually:

notebookutils.notebook.run(
    "SalesTransform",
    600,
    {

"load_date": load_date

    }
)

Child notebook can then use: load_date for filtering or partitioning.

4.7.16 Returning Values

A child notebook may need to communicate a result back to its parent. For example:

Child Notebook
      │
      ▼
Processed 1,250 records
      │
      ▼
Parent Notebook

A common pattern is to return a structured result containing: status record_count message For example: {

    "status": "SUCCESS",

"record_count": 1250

}

The exact return mechanism should follow the Fabric notebook runtime's supported API.

4.7.17 Notebook Workflow

A practical Fabric pipeline might look like:

             Master Notebook
                    │
                    ▼
            Check Input Files
                    │
                    ▼
             Ingestion Notebook
                    │
                    ▼
              Bronze Delta
                    │
                    ▼
             Transform Notebook
                    │
                    ▼
              Silver Delta
                    │
                    ▼
              Load Notebook
                    │
                    ▼
               Gold Delta
                    │
                    ▼
             Validation Notebook
                    │
                    ▼
                  SUCCESS

4.7.18 Getting Notebook Context

Notebook environments can expose context information through supported utilities. This can be useful when you need information such as:

• Workspace
• Lakehouse
• Notebook
• Execution context
• Run information

The exact context API depends on the Fabric runtime version. For production code, prefer documented Fabric APIs rather than relying on undocumented internal objects.

4.7.19 Using mssparkutils vs notebookutils

You may encounter examples using: mssparkutils in older Microsoft Fabric/Synapse/Spark documentation and examples. Modern Fabric documentation uses:

notebookutils

for notebook utility functionality. For new Fabric development, prefer the current notebookutils APIs supported by your Fabric runtime.

4.7.20 Notebook Utilities for File-Based Ingestion

Consider a folder:

Files/Raw/Sales/
│
├── Sales_20260820.csv
├── Sales_20260821.csv
└── Sales_20260822.csv

The notebook can list the files:

files = notebookutils.fs.ls(
    "Files/Raw/Sales"
)

for file in files:
    print(file.name)
Then identify candidate files and process them.

Example Processing Pattern

files = notebookutils.fs.ls(
    "Files/Raw/Sales"
)

for file in files:

    if file.name.endswith(".csv"):

        print(
            f"Processing {file.name}"
        )

        df = spark.read \
            .option("header", "true") \
            .option("inferSchema", "true") \
            .csv(file.path)

        # Transform data here

This provides a basic file-driven ingestion pattern.

4.7.21 Archive Processed Files

After successful processing: Before:

Raw/
└── Sales_20260822.csv

After:

Raw/
└──

Archive/
└── Sales_20260822.csv

Conceptually:

notebookutils.fs.mv(
    file.path,

"Files/Archive/" + file.name

)
Important: In production, move/archive only after the Delta write and validation succeed.

4.7.22 Error Handling

A robust notebook should handle errors. Example:

try:

    df = spark.read.csv(
        "Files/Sales.csv",
        header=True,
        inferSchema=True
    )

    print("File loaded successfully")

except Exception as e:

    print(
        f"File processing failed: {e}"
    )

    raise

The raise ensures that orchestration systems can detect the failure.

4.7.23 Validation Before Archive

A safer pattern:

Read File
    │
    ▼
Transform
    │
    ▼
Write Delta
    │
    ▼
Validate
    │
    ├── FAILED → Keep Source
    │
    └── SUCCESS
            │
            ▼
          Archive

This prevents a failed file from being moved away before successful processing.

4.7.24 Record Count Validation

Example:

source_count = df.count()

target_count = spark.table(
    "Sales"
).count()

print(
    f"Source: {source_count}"
)

print(
    f"Target: {target_count}"
)

For incremental pipelines, validation should normally compare the appropriate batch counts rather than blindly comparing the entire source and target table counts.

4.7.25 Notebook Utilities + Delta

A complete ingestion pattern could be:

                Raw CSV
                   │
                   ▼

          notebookutils.fs.ls()

                   │
                   ▼
             Select File
                   │
                   ▼
              Read CSV
                   │
                   ▼
            Spark DataFrame
                   │
                   ▼
              Transform
                   │
                   ▼
             Delta MERGE
                   │
                   ▼
              Validation
                   │
              ┌────┴────┐
              ▼         ▼
           Success    Failure
              │         │
              ▼         ▼
           Archive    Keep Raw

4.7.26 Notebook Utilities + Parameters

Imagine a reusable notebook: Notebook: LoadToDelta Parameters: source_path target_table load_type load_date Execution 1: source_path = Files/Sales.csv target_table = Sales load_type = incremental Execution 2: source_path = Files/Customers.csv target_table = Customers load_type = full The same notebook can handle both workloads.

4.7.27 Building a Reusable Notebook

A good reusable notebook can follow:

# Parameters

source_path = "Files/Sales.csv"
target_table = "Sales"
load_type = "full"

Then:

# Read

df = spark.read \
    .option("header", "true") \
    .option("inferSchema", "true") \
    .csv(source_path)

Transform:

# Transform

clean_df = df.dropDuplicates()

Load:

# Load

clean_df.write \
    .format("delta") \
    .mode("overwrite") \
    .saveAsTable(target_table)

Validate:

# Validate

record_count = spark.table(

target_table

).count()

print(
    f"Loaded {record_count} records"
)

4.7.28 Notebook Utilities Best Practices

1. Don't Hardcode Paths

Avoid: "Files/Sales.csv" throughout the notebook. Use parameters: source_path

2. Separate Configuration from Logic

Example:

source_path = "Files/Sales"
target_table = "Sales"
load_type = "incremental"
Then processing logic follows.

3. Make Notebooks Idempotent

An idempotent notebook produces the correct result even if the same execution is accidentally repeated. For example:

Run 1 → 10,000 records
Run 2 → same input
       ↓
No duplicate records

Delta MERGE is often useful for this.

4. Validate Before Archiving

Never move a source file to the archive until: Read ✓ Transform ✓ Write ✓ Validate ✓

5. Fail Clearly

Avoid:

except:

pass This hides failures. Prefer:

except Exception as e:
    print(e)
    raise

4.7.29 Practical Project

Project: File-to-Lakehouse Ingestion Notebook Input Files/Raw/Sales/ Output Tables/Sales Requirements

1. List files.
2. Find CSV files.
3. Read each file.
4. Clean data.
5. Remove duplicates.
6. Write to Delta.
7. Validate record count.
8. Archive successfully processed files.
9. Leave failed files in Raw.

Architecture:

Files/Raw/Sales
       │
       ▼

 notebookutils.fs.ls()

       │
       ▼
    CSV File
       │
       ▼
    DataFrame
       │
       ▼
    Cleaning
       │
       ▼
   Delta Table
       │
       ▼
   Validation
       │
    ┌──┴──┐
    ▼     ▼
Success  Failed
    │     │
    ▼     ▼
Archive  Raw

4.7.30 Example Notebook

Cell 1 — Configuration

source_folder = "Files/Raw/Sales"
archive_folder = "Files/Archive/Sales"
target_table = "Sales"

Cell 2 — Create Archive

notebookutils.fs.mkdirs(

archive_folder

)

Cell 3 — List Files

files = notebookutils.fs.ls(

source_folder

)

for file in files:
    print(file.name)

Cell 4 — Process CSV

for file in files:

    if not file.name.endswith(".csv"):

continue

    print(
        f"Processing: {file.name}"
    )

    df = spark.read \
        .option("header", "true") \
        .option("inferSchema", "true") \
        .csv(file.path)

    df = df.dropDuplicates()

Cell 5 — Write Delta

    df.write \
        .format("delta") \
        .mode("append") \
        .saveAsTable(target_table)

Cell 6 — Archive After successful validation:

    notebookutils.fs.mv(
        file.path,

archive_folder + "/" + file.name

    )
For production, add explicit validation and error handling around the write/archive steps.

4.7.31 Interview Questions

1. What are Notebook Utilities?

They are Fabric notebook capabilities used for tasks such as file management, notebook orchestration, parameter handling, and environment interaction.

1. What is notebookutils.fs?

It provides filesystem-related utility operations for supported storage locations.

1. How do you list files?

notebookutils.fs.ls("Files")

1. How do you create a directory?

notebookutils.fs.mkdirs(
    "Files/archive"
)

1. How do you copy a file?

notebookutils.fs.cp(
    source,

destination

)

1. How do you move a file?

notebookutils.fs.mv(
    source,

destination

)

1. Why use notebook parameters?

They allow the same notebook to process different datasets, files, dates, or tables without changing the code.

1. Why use child notebooks?

They allow large workflows to be divided into reusable and independently testable components.

1. What should happen if Delta loading fails?

The notebook should fail clearly and the source file should generally remain available for retry rather than being archived as successfully processed.

1. What is an idempotent notebook?

A notebook is idempotent when repeating the same execution produces the correct final state without creating unintended duplicates or inconsistent results.

              Source Files
                   │
                   ▼

          notebookutils.fs

                   │
                   ▼
             Read Data
                   │
                   ▼
            Spark DataFrame
                   │
                   ▼
          Clean / Transform
                   │
                   ▼
             Delta MERGE
                   │
                   ▼
              Validate
                   │
          ┌────────┴────────┐
          ▼                 ▼
       SUCCESS             FAILURE
          │                 │
          ▼                 ▼
       Archive           Keep Raw

Reading CSV files is one of the most common operations in a Microsoft Fabric Notebook. In a typical Fabric data-engineering pipeline, CSV data is read from the Lakehouse Files area into a Spark DataFrame, cleaned, validated, and then written to a Delta table.

CSV File
   │
   ▼
OneLake / Lakehouse Files
   │
   ▼
Fabric Notebook
   │
   ▼
Spark DataFrame
   │
   ▼
Clean / Transform
   │
   ▼
Delta Table

4.8.1 Basic CSV Reading

Assume the Lakehouse contains:

Files/
└── Sales.csv

Read it using PySpark:

df = spark.read.csv("Files/Sales.csv")

Display the data:

display(df)
However, CSV files normally contain a header, so use:
df = spark.read \
    .option("header", "true") \
    .csv("Files/Sales.csv")

display(df)

4.8.2 Example CSV

Suppose Sales.csv contains: SaleID,CustomerID,CustomerName,Country,Amount 1,101,John,USA,2500 2,102,Anita,India,3200 3,103,David,UK,1800 4,104,Priya,India,4100 Read it:

df = spark.read \
    .option("header", "true") \
    .csv("Files/Sales.csv")

Output:

4.8.3 Schema Inference

By default, CSV values may be interpreted as strings. Use:

df = spark.read \
    .option("header", "true") \
    .option("inferSchema", "true") \
    .csv("Files/Sales.csv")

Check the schema:

df.printSchema()

You may get: root |-- SaleID: integer |-- CustomerID: integer |-- CustomerName: string |-- Country: string |-- Amount: integer Without schema inference, you may instead get:

SaleID       → string
CustomerID   → string
Amount       → string

4.8.4 Explicit Schema

For production pipelines, explicitly defining the schema is often preferable.

from pyspark.sql.types import (
    StructType,
    StructField,
    IntegerType,
    StringType,

DoubleType

)

schema = StructType([
    StructField("SaleID", IntegerType(), True),
    StructField("CustomerID", IntegerType(), True),
    StructField("CustomerName", StringType(), True),
    StructField("Country", StringType(), True),
    StructField("Amount", DoubleType(), True)
])

Read the CSV:

df = spark.read \
    .option("header", "true") \
    .schema(schema) \
    .csv("Files/Sales.csv")

Check:

df.printSchema()

4.8.5 Reading Multiple CSV Files

Suppose:

Files/Sales/
├── Sales_20260820.csv
├── Sales_20260821.csv
└── Sales_20260822.csv

Read all files:

df = spark.read \
    .option("header", "true") \
    .option("inferSchema", "true") \
    .csv("Files/Sales/")

Spark creates one logical DataFrame containing the records from all matching files.

4.8.6 Reading Specific Files

You can also specify multiple files:

df = spark.read \
    .option("header", "true") \
    .csv([
        "Files/Sales/Sales_20260821.csv",
        "Files/Sales/Sales_20260822.csv"
    ])

4.8.7 Different Delimiters

Not every CSV-like file uses a comma. Pipe-delimited SaleID|CustomerID|Country|Amount 1|101|India|2500 Read using:

df = spark.read \
    .option("header", "true") \
    .option("delimiter", "|") \
    .csv("Files/Sales.txt")

Tab-delimited

df = spark.read \
    .option("header", "true") \
    .option("delimiter", "\t") \
    .csv("Files/Sales.tsv")

4.8.8 Handling Quoted Values

Consider: SaleID,CustomerName,Country,Amount 1,"John, Kumar",India,2500 The comma inside "John, Kumar" should not be treated as a delimiter. Use:

df = spark.read \
    .option("header", "true") \
    .option("quote", '"') \
    .csv("Files/Sales.csv")

4.8.9 Handling NULL Values

Suppose the file contains: SaleID,CustomerName,Country,Amount 1,John,India,NULL You can specify:

df = spark.read \
    .option("header", "true") \
    .option("nullValue", "NULL") \
    .csv("Files/Sales.csv")

Then check:

df.filter(
    df.Amount.isNull()
).show()

4.8.10 Reading Nested Folders

Example:

Files/Sales/
├── 2025/
│   ├── Sales01.csv
│   └── Sales02.csv
└── 2026/
    ├── Sales01.csv
    └── Sales02.csv

For recursive file discovery:

df = spark.read \
    .option("header", "true") \
    .option("recursiveFileLookup", "true") \
    .csv("Files/Sales/")

4.8.11 Add Source File Name

For ingestion pipelines, tracking the source file is useful.

from pyspark.sql.functions import input_file_name

df = df.withColumn(
    "SourceFile",
    input_file_name()
)

Now you can identify where each record originated. SaleID | Amount | SourceFile 1 | 2500 | Sales_20260821.csv 2 | 3200 | Sales_20260821.csv 3 | 1800 | Sales_20260822.csv This is useful for auditing, troubleshooting, and data lineage.

4.8.12 Inspect the DataFrame

After reading a CSV, always inspect it. Display records

display(df)

Show first records

df.show(10)

Check schema

df.printSchema()

Get column names

print(df.columns)

Count records

print(df.count())

4.8.13 Data Quality Validation

Check whether required columns exist:

required_columns = [
    "SaleID",
    "CustomerID",
    "Amount"
]

for column in required_columns:
    if column not in df.columns:
        raise Exception(
            f"Missing required column: {column}"
        )

Check NULL Sale IDs:

from pyspark.sql.functions import col

null_count = df.filter(
    col("SaleID").isNull()
).count()

if null_count > 0:
    raise Exception(
        f"Found {null_count} NULL SaleID records"
    )

Check duplicates:

duplicates = df.groupBy(
    "SaleID"
).count().filter(
    col("count") > 1
)

display(duplicates)

4.8.14 Clean CSV Data

Suppose the CSV contains:

CustomerName = "  John  "
Country       = "india"
Amount        = "2500"

Clean it:

from pyspark.sql.functions import (
    trim,
    upper,

col

)

clean_df = df \
    .withColumn(
        "CustomerName",
        trim(col("CustomerName"))
    ) \
    .withColumn(
        "Country",
        upper(col("Country"))
    ) \
    .withColumn(
        "Amount",
        col("Amount").cast("double")
    )

4.8.15 CSV → Delta

This is the most important Fabric pattern for this lesson.

CSV
 │
 ▼
Read CSV
 │
 ▼
DataFrame
 │
 ▼
Validate
 │
 ▼
Clean
 │
 ▼
Write Delta

Example:

df = spark.read \
    .option("header", "true") \
    .option("inferSchema", "true") \
    .csv("Files/Sales.csv")

Clean:

clean_df = df.dropDuplicates(
    ["SaleID"]
)

Write:

clean_df.write \
    .format("delta") \
    .mode("overwrite") \
    .saveAsTable("Sales")

Now the Lakehouse contains a Delta table:

Tables/
└── Sales

4.8.16 Production CSV Ingestion Pattern

A production notebook should generally follow:

                 Raw CSV
                    │
                    ▼
             Check File Exists
                    │
                    ▼
                Read CSV
                    │
                    ▼
             Validate Schema
                    │
                    ▼
             Validate Records
                    │
                    ▼
             Clean / Transform
                    │
                    ▼
              Write Delta
                    │
                    ▼
             Validate Target
                    │
             ┌──────┴──────┐
             ▼             ▼
          SUCCESS        FAILURE
             │             │
             ▼             ▼
          Archive       Keep Raw

4.8.17 Complete Fabric Notebook Example

from pyspark.sql.functions import (
    col,
    trim,
    upper,

input_file_name

)

# 1. Source path
source_path = "Files/Raw/Sales/Sales.csv"

# 2. Read CSV
df = spark.read \
    .option("header", "true") \
    .option("inferSchema", "true") \
    .csv(source_path)

# 3. Add source file
df = df.withColumn(
    "SourceFile",
    input_file_name()
)

# 4. Remove duplicates
df = df.dropDuplicates(
    ["SaleID"]
)

# 5. Clean data
df = df \
    .withColumn(
        "CustomerName",
        trim(col("CustomerName"))
    ) \
    .withColumn(
        "Country",
        upper(col("Country"))
    ) \
    .withColumn(
        "Amount",
        col("Amount").cast("double")
    )

# 6. Validate
if df.filter(
    col("SaleID").isNull()
).count() > 0:
    raise Exception("SaleID contains NULL values")

# 7. Write Delta
df.write \
    .format("delta") \
    .mode("overwrite") \
    .saveAsTable("Sales")

# 8. Display result
display(
    spark.table("Sales")
)

4.8.18 Important CSV Options

4.8.19 Interview Questions

What is the basic syntax to read CSV?

df = spark.read \
    .option("header", "true") \
    .csv("Files/Sales.csv")

How do you enable schema inference?

.option("inferSchema", "true")

How do you specify a pipe delimiter?

.option("delimiter", "|")

How do you read all CSV files in a folder?

df = spark.read \
    .option("header", "true") \
    .csv("Files/Sales/")

Why use an explicit schema? It provides predictable and controlled data types, which is particularly important in production pipelines. Why add SourceFile? It provides useful batch-level lineage and helps troubleshoot bad records. Why convert CSV to Delta? CSV is primarily a simple interchange/ingestion format. Delta provides a much stronger foundation for reliable Lakehouse storage, including transactional operations and incremental processing.

df = spark.read \
    .option("header", "true") \
    .option("inferSchema", "true") \
    .csv("Files/Sales.csv")

CSV
 ↓
Spark DataFrame
 ↓
Validate
 ↓
Clean
 ↓
Transform
 ↓
Delta Table

Learning Objectives

By the end of this lesson, you will be able to:

• Understand data cleaning in PySpark.
• Handle NULL and missing values.
• Remove duplicate records.
• Trim and standardize text.
• Correct data types.
• Clean dates and timestamps.
• Validate numeric values.
• Handle invalid records.
• Standardize column names.
• Build a reusable data-cleaning pipeline in Microsoft Fabric.

4.9.1 What is Data Cleaning?

Data cleaning is the process of identifying and correcting inaccurate, incomplete, inconsistent, duplicated, or invalid data before it is loaded into a production data store. Typical raw data: CustomerID | CustomerName | Country | Amount 101 | " John " | india | "2500" 102 | NULL | INDIA | "3200" 101 | " John " | india | "2500" 103 | "David" | UK | "ABC" After cleaning: CustomerID | CustomerName | Country | Amount 101 | John | INDIA | 2500.0 102 | Unknown | INDIA | 3200.0 103 | David | UK | NULL The cleaned data can then be written to a Delta table.

4.9.2 Data Cleaning in Fabric

A common Fabric pattern is:

Raw CSV
   │
   ▼
Spark DataFrame
   │
   ▼
Data Cleaning
   │
   ├── NULL handling
   ├── Duplicate removal
   ├── Type conversion
   ├── Text standardization
   ├── Date validation
   └── Business validation
   │
   ▼
Clean DataFrame
   │
   ▼
Delta Table

4.9.3 Inspect the Raw Data First

Before cleaning, inspect the DataFrame.

display(df)

Check schema:

df.printSchema()

Check columns:

print(df.columns)

Check row count:

print("Records:", df.count())

A good rule is: Inspect first, clean second.

4.9.4 Standardize Column Names

Raw CSV files may contain: Customer ID Customer Name Sale Amount Order Date These names can be inconvenient for Spark SQL and downstream systems. Rename them:

df = df \
    .withColumnRenamed("Customer ID", "CustomerID") \
    .withColumnRenamed("Customer Name", "CustomerName") \
    .withColumnRenamed("Sale Amount", "SaleAmount") \
    .withColumnRenamed("Order Date", "OrderDate")

A common target convention is: CustomerID CustomerName SaleAmount OrderDate

4.9.5 Trim Whitespace

Raw data often contains unwanted spaces. Example: " John Kumar " Use:

from pyspark.sql.functions import trim, col

df = df.withColumn(
    "CustomerName",
    trim(col("CustomerName"))
)

Result: "John Kumar"

4.9.6 Standardize Case

Suppose Country contains: India india INDIA InDiA Standardize it:

from pyspark.sql.functions import upper

df = df.withColumn(
    "Country",
    upper(trim(col("Country")))
)

Now: INDIA INDIA INDIA INDIA Alternatively, use lowercase:

from pyspark.sql.functions import lower

df = df.withColumn(
    "Country",
    lower(trim(col("Country")))
)

4.9.7 Handling NULL Values

NULL is one of the most common data-quality problems. Check NULL values:

df.filter(
    col("CustomerName").isNull()
).show()

Check non-NULL:

df.filter(
    col("CustomerName").isNotNull()
).show()

4.9.8 Fill NULL Values

Use fillna().

df = df.fillna({
    "CustomerName": "Unknown",
    "Country": "UNKNOWN",

"Amount": 0

})

Example: Before: CustomerName = NULL After: CustomerName = Unknown

4.9.9 Drop NULL Records

If a required field is NULL, you may want to remove the record.

df = df.dropna(
    subset=["CustomerID"]
)

This removes records where CustomerID is NULL.

Fill vs Drop

Do not automatically replace every NULL with zero.

4.9.10 Replace Specific Values

Suppose the source contains: Country ------ India IND IN USA US Standardize:

df = df.replace(
    {
        "IND": "INDIA",
        "IN": "INDIA",
        "US": "USA"
    },
    subset=["Country"]
)

Now the values are standardized.

4.9.11 Remove Duplicates

Complete duplicate rows:

df = df.dropDuplicates()

But often you should use a business key. For example:

df = df.dropDuplicates(
    ["SaleID"]
)

This is safer for sales data.

Why Business Keys Matter

Suppose: SaleID | CustomerID | Amount 1001 | 101 | 2500 1001 | 101 | 2500 If SaleID uniquely identifies a sale:

df.dropDuplicates(["SaleID"])
is appropriate.

4.9.12 Data Type Conversion

Raw CSV data frequently arrives as strings. Check:

df.printSchema()

Suppose: Amount: string Convert:

df = df.withColumn(
    "Amount",
    col("Amount").cast("double")
)

Convert ID:

df = df.withColumn(
    "CustomerID",
    col("CustomerID").cast("long")
)

4.9.13 Numeric Validation

Suppose Amount should never be negative. Find invalid records:

invalid_df = df.filter(
    col("Amount") < 0
)

display(invalid_df)

You can reject them:

df = df.filter(
    col("Amount") >= 0
)
However, in production, it is often better to quarantine invalid records rather than silently delete them.

4.9.14 Creating Valid and Invalid DataFrames

valid_df = df.filter(
    (col("Amount") >= 0) &
    col("CustomerID").isNotNull()
)

invalid_df = df.filter(
    (col("Amount") < 0) |
    col("CustomerID").isNull()
)

Architecture:

                 Raw Data
                    │
                    ▼
               Validation
                /       \
               /         \
              ▼           ▼
         Valid Data   Invalid Data
              │           │
              ▼           ▼
         Delta Table    Quarantine

This is a better production pattern than simply deleting bad records.

4.9.15 Cleaning String Columns

Useful functions include: trim() upper() lower() length() substring() regexp_replace() Example:

from pyspark.sql.functions import (
    trim,
    upper,

regexp_replace

)

Remove extra spaces:

df = df.withColumn(
    "CustomerName",
    trim(col("CustomerName"))
)

Remove unwanted characters:

df = df.withColumn(
    "Phone",
    regexp_replace(
        col("Phone"),
        "[^0-9]",
        ""
    )
)

For: (987) 654-3210 you can obtain: 9876543210

4.9.16 Email Validation

A basic email-quality check:

invalid_email = df.filter(
    ~col("Email").rlike(
        r"^[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Za-z]{2,}$"
    )
)
This is a basic validation, not a guarantee that the mailbox actually exists.

4.9.17 Date Cleaning

Raw date: 22/08/2026 Convert it:

from pyspark.sql.functions import to_date

df = df.withColumn(
    "OrderDate",
    to_date(
        col("OrderDate"),
        "dd/MM/yyyy"
    )
)

Another format: 2026-08-22 Use:

df = df.withColumn(
    "OrderDate",
    to_date(
        col("OrderDate"),
        "yyyy-MM-dd"
    )
)

4.9.18 Timestamp Cleaning

Example: 2026-08-22 10:30:45 Use:

from pyspark.sql.functions import to_timestamp

df = df.withColumn(
    "CreatedTimestamp",
    to_timestamp(
        col("CreatedTimestamp"),
        "yyyy-MM-dd HH:mm:ss"
    )
)

4.9.19 Handling Invalid Dates

Suppose: OrderDate 2026-08-20 2026-08-21 ABC Using to_date() can result in NULL for an invalid value. Check:

invalid_dates = df.filter(
    col("OrderDate").isNull()
)

display(invalid_dates)

This gives you a way to identify bad source records.

4.9.20 Business Rules

Data cleaning is not only technical formatting. Business rules are important. Example: Quantity > 0 Amount >= 0 CustomerID cannot be NULL OrderDate cannot be in the future Implement:

valid_df = df.filter(
    (col("Quantity") > 0) &
    (col("Amount") >= 0) &
    col("CustomerID").isNotNull()
)

4.9.21 Using when() for Data Standardization

Suppose: Country IN IND India Convert all to INDIA:

from pyspark.sql.functions import when

df = df.withColumn(
    "Country",
    when(
        col("Country").isin(
            "IN",
            "IND",
            "India",
            "india"
        ),
        "INDIA"
    )
    .when(
        col("Country").isin(
            "US",
            "USA",
            "United States"
        ),
        "USA"
    )
    .otherwise(
        upper(trim(col("Country")))
    )
)

4.9.22 Cleaning Numeric Strings

Sometimes numbers contain commas: "1,250.50" Direct casting may not work as expected. First remove commas:

from pyspark.sql.functions import regexp_replace

df = df.withColumn(
    "Amount",
    regexp_replace(
        col("Amount"),
        ",",
        ""
    ).cast("double")
)

Result: 1250.50

4.9.23 Remove Currency Symbols

Input: ₹2,500.00 $3,200.00 You can remove non-numeric formatting according to your business rules. For example:

df = df.withColumn(
    "Amount",
    regexp_replace(
        col("Amount"),
        "[₹$,]",
        ""
    ).cast("double")
)
Be careful with negative values, decimal separators, and different currency formats.

4.9.24 Standardize Column Names Automatically

For a large DataFrame:

for old_name in df.columns:
    new_name = (

old_name

        .strip()
        .replace(" ", "_")
        .replace("-", "_")
    )

    df = df.withColumnRenamed(
        old_name,

new_name

    )

For example: Customer ID Customer Name Order-Date becomes: Customer_ID Customer_Name Order_Date

4.9.25 Data Cleaning with a Reusable Function

Instead of repeating code:

def clean_sales_data(df):

    df = df.dropDuplicates(
        ["SaleID"]
    )

    df = df \
        .withColumn(
            "CustomerName",
            trim(col("CustomerName"))
        ) \
        .withColumn(
            "Country",
            upper(trim(col("Country")))
        ) \
        .withColumn(
            "Amount",
            regexp_replace(
                col("Amount"),
                ",",
                ""
            ).cast("double")
        )

    return df

Use:

clean_df = clean_sales_data(df)

This makes notebooks easier to maintain.

4.9.26 Complete Cleaning Pipeline

Suppose the raw data is: SaleID | CustomerName | Country | Amount | Quantity 1001 | " John " | india | "2,500" | 2 1002 | " Anita " | INDIA | "3,200" | 1 1001 | " John " | india | "2,500" | 2 1003 | NULL | UK | "-500" | 1 Step 1 — Read

df = spark.read \
    .option("header", "true") \
    .option("inferSchema", "true") \
    .csv("Files/Sales.csv")

Step 2 — Trim

df = df.withColumn(
    "CustomerName",
    trim(col("CustomerName"))
)

Step 3 — Standardize Country

df = df.withColumn(
    "Country",
    upper(trim(col("Country")))
)

Step 4 — Clean Amount

df = df.withColumn(
    "Amount",
    regexp_replace(
        col("Amount"),
        ",",
        ""
    ).cast("double")
)

Step 5 — Remove Duplicates

df = df.dropDuplicates(
    ["SaleID"]
)

Step 6 — Handle Customer Name

df = df.fillna({
    "CustomerName": "Unknown"
})

Step 7 — Separate Invalid Records

invalid_df = df.filter(
    (col("Amount") < 0) |
    (col("Quantity") <= 0) |
    col("SaleID").isNull()
)

valid_df = df.filter(
    (col("Amount") >= 0) &
    (col("Quantity") > 0) &
    col("SaleID").isNotNull()
)

4.9.27 Quarantine Invalid Data

Instead of deleting invalid records:

invalid_df.write \
    .format("delta") \
    .mode("append") \
    .saveAsTable("Sales_Quarantine")

Valid data:

valid_df.write \
    .format("delta") \
    .mode("append") \
    .saveAsTable("Sales")

Architecture:

                 Raw CSV
                    │
                    ▼
             Spark DataFrame
                    │
                    ▼
               Data Cleaning
                    │
                    ▼
              Data Validation
                 /       \
                /         \
               ▼           ▼
          Valid Data   Invalid Data
               │           │
               ▼           ▼
          Sales Delta   Quarantine

This is a strong production pattern.

4.9.28 Data Quality Metrics

Track metrics such as: Source Records = 100,000 Valid Records = 98,500 Invalid Records = 1,500 Duplicate Records = 800 NULL Customer IDs = 300 Invalid Amounts = 400 You can calculate:

source_count = df.count()
valid_count = valid_df.count()
invalid_count = invalid_df.count()

print("Source:", source_count)
print("Valid:", valid_count)
print("Invalid:", invalid_count)

4.9.29 Data Cleaning vs Data Transformation

These concepts are related but different. Data Cleaning Correct data quality problems:

" india " → "INDIA"
NULL → "Unknown"
"2,500" → 2500
Duplicate → Remove
Data Transformation

Apply business logic: Amount × Tax Customer Segmentation Sales Aggregation Currency Conversion Pipeline:

Raw
 ↓
Clean
 ↓
Transform
 ↓
Load

4.9.30 Clean Data → Delta

Once cleaned:

valid_df.write \
    .format("delta") \
    .mode("overwrite") \
    .saveAsTable("Sales")

Now:

Lakehouse
│
├── Files
│   └── Raw
│
└── Tables
    ├── Sales
    └── Sales_Quarantine

4.9.31 Production Best Practices

1. Never silently discard bad data

Prefer:

Valid → Production
Invalid → Quarantine

1. Preserve the raw layer

Keep the original source data when possible.

Raw → Clean → Curated

1. Use explicit schemas

Especially for critical production pipelines.

1. Validate business keys

Check:

• NULL
• Duplicate
• Invalid format

1. Add audit columns

For example:

from pyspark.sql.functions import (
    current_timestamp,

lit

)

df = df \
    .withColumn(
        "LoadTimestamp",
        current_timestamp()
    ) \
    .withColumn(
        "SourceSystem",
        lit("SalesCSV")
    )

1. Make the pipeline idempotent

Repeated execution should not create duplicates.

1. Track data-quality metrics

Monitor:

• Record count
• Invalid count
• Duplicate count
• NULL count

4.9.32 End-to-End Fabric Pattern

                 OneLake
                    │
                    ▼
              Raw CSV Files
                    │
                    ▼
             Fabric Notebook
                    │
                    ▼
             Spark DataFrame
                    │
                    ▼
             ┌──────────────┐
             │ Data Cleaning│
             └──────┬───────┘
                    │
       ┌────────────┼────────────┐
       ▼            ▼            ▼
     NULLs       Duplicates    Types
       │            │            │
       └────────────┼────────────┘
                    ▼
             Business Rules
                    │
              ┌─────┴─────┐
              ▼           ▼
           Valid        Invalid
              │           │
              ▼           ▼
         Silver Delta  Quarantine
              │
              ▼
           Analytics

Hands-On Lab

Project: Clean Sales Data Input Files/Raw/Sales.csv Step 1 — Read CSV

df = spark.read \
    .option("header", "true") \
    .option("inferSchema", "true") \
    .csv("Files/Raw/Sales.csv")

Step 2 — Clean Text

df = df \
    .withColumn(
        "CustomerName",
        trim(col("CustomerName"))
    ) \
    .withColumn(
        "Country",
        upper(trim(col("Country")))
    )

Step 3 — Clean Amount

df = df.withColumn(
    "Amount",
    regexp_replace(
        col("Amount"),
        ",",
        ""
    ).cast("double")
)

Step 4 — Remove Duplicates

df = df.dropDuplicates(
    ["SaleID"]
)

Step 5 — Separate Valid/Invalid

valid_df = df.filter(
    (col("Amount") >= 0) &
    col("SaleID").isNotNull()
)

invalid_df = df.filter(
    (col("Amount") < 0) |
    col("SaleID").isNull()
)

Step 6 — Write Valid Data

valid_df.write \
    .format("delta") \
    .mode("overwrite") \
    .saveAsTable("Sales")

Step 7 — Write Invalid Data

invalid_df.write \
    .format("delta") \
    .mode("append") \
    .saveAsTable("Sales_Quarantine")

Interview Questions

df.dropDuplicates()
or:
df.dropDuplicates(["SaleID"])

df.fillna(...)
or:
df.dropna(...)

df.withColumn(
    "CustomerName",
    trim(col("CustomerName"))
)

df.withColumn(
    "Amount",
    col("Amount").cast("double")
)

Schema validation
      ↓
Data-quality validation
      ↓
Business-rule validation
      ↓
Write/MERGE Delta

             Raw CSV
                │
                ▼
         Spark DataFrame
                │
                ▼
          Clean Data
                │
      ┌─────────┴─────────┐
      ▼                   ▼
   Valid                 Invalid
      │                   │
      ▼                   ▼
Silver Delta          Quarantine
      │
      ▼
Gold / Analytics

Learning Objectives

By the end of this lesson, you will be able to:

• Understand how to write Spark DataFrames to Delta tables.
• Create Delta tables in a Fabric Lakehouse.
• Use overwrite and append modes.
• Write managed Delta tables.
• Write Delta data to paths.
• Partition Delta tables appropriately.
• Handle schema evolution.
• Validate Delta writes.
• Build production-ready CSV → DataFrame → Delta pipelines.

4.10.1 What is Write Delta?

After reading and cleaning data in a Fabric Notebook, the next step is often to store it as a Delta table. Typical flow:

CSV / SQL / API
      │
      ▼
Spark DataFrame
      │
      ▼
Clean Data
      │
      ▼
Write Delta
      │
      ▼
Lakehouse Delta Table

In Fabric:

OneLake
   │
   ▼
Lakehouse
   │
   ├── Files
   │
   └── Tables
        │
        └── Sales  ← Delta Table

4.10.2 Basic Write to Delta

Suppose you already have: df Write it as a Delta table:

df.write \
    .format("delta") \
    .mode("overwrite") \
    .saveAsTable("Sales")

Now you have a table named: Sales You can query it:

SELECT *
FROM Sales;

4.10.3 Why Delta?

Delta provides capabilities such as:

• ACID transactions
• Reliable writes
• Schema management
• UPDATE
• DELETE
• MERGE
• Version history
• Incremental processing

Conceptually: Data Files + Delta Transaction Log = Delta Table

4.10.4 Write Modes

Spark supports several important write modes.

4.10.5 Overwrite Mode

Use:

df.write \
    .format("delta") \
    .mode("overwrite") \
    .saveAsTable("Sales")

Suppose the table currently contains: 100 records and your DataFrame contains: 50 records After overwrite: Sales = 50 records The previous table state is replaced by the new write.

When to Use Overwrite

Typical scenarios:

• Full refresh
• Rebuilding a table
• Development/testing
• Reprocessing a complete partition or dataset

Example:

Source
  │
  ▼
Full Extract
  │
  ▼
Clean
  │
  ▼
Overwrite Delta

4.10.6 Append Mode

Append adds records:

df.write \
    .format("delta") \
    .mode("append") \
    .saveAsTable("Sales")

Suppose: Existing = 10,000 Incoming = 1,000 After append: Total = 11,000

When to Use Append

Typical scenarios:

• Daily files
• Event data
• New transactions
• Historical snapshots
• Immutable records

Example:

Day 1 → 10,000
Day 2 → +2,000
Day 3 → +1,500

4.10.7 Error Mode

You can use:

df.write \
    .format("delta") \
    .mode("error") \
    .saveAsTable("Sales")
If the destination already exists, the write fails.

This is useful when accidental replacement must be prevented.

4.10.8 Ignore Mode

df.write \
    .format("delta") \
    .mode("ignore") \
    .saveAsTable("Sales")
If the table already exists, Spark does not overwrite it.

4.10.9 Save as Table vs Save as Path

There are two common patterns. Save as Table

df.write \
    .format("delta") \
    .mode("overwrite") \
    .saveAsTable("Sales")

Save to Path

df.write \
    .format("delta") \
    .mode("overwrite") \
    .save("Files/Delta/Sales")

The table approach is generally convenient when working with Lakehouse tables.

4.10.10 Reading the Written Delta Table

After writing:

df.write \
    .format("delta") \
    .mode("overwrite") \
    .saveAsTable("Sales")

Read it:

sales_df = spark.read.table("Sales")

Display:

display(sales_df)

4.10.11 Verify the Write

Always validate production writes. Record count

source_count = df.count()

target_count = spark.table(
    "Sales"
).count()

print("Source:", source_count)
print("Target:", target_count)
For a full overwrite, you may expect:

Source Count = Target Count For an incremental append or merge, validation needs to be batch-aware.

4.10.12 Check Schema

spark.table(
    "Sales"
).printSchema()

Compare:

Source Schema
      │
      ▼
Target Schema

This helps identify unexpected schema changes.

4.10.13 Write with Partitioning

You can partition a Delta table:

df.write \
    .format("delta") \
    .partitionBy("Country") \
    .mode("overwrite") \
    .saveAsTable("Sales")

Conceptually:

Sales
│
├── Country=INDIA
├── Country=USA
└── Country=UK

When Should You Partition?

Partitioning can help when:

• The table is sufficiently large.
• Queries frequently filter by the partition column.
• The partition column has reasonable cardinality.

Example query:

SELECT *
FROM Sales
WHERE Country = 'INDIA';
If Country is an appropriate partition column, partition pruning can reduce the amount of data read.

Avoid High-Cardinality Partitions

Do not blindly partition by: CustomerID TransactionID Email if they contain millions of distinct values. This can create too many small partitions/files. A good partitioning strategy depends on:

• Data volume
• Query patterns
• Cardinality
• File sizes
• Write frequency

4.10.14 Write with Schema Evolution

Suppose the existing table has: CustomerID CustomerName Amount New data contains: CustomerID CustomerName Amount Country You can enable schema merging:

df.write \
    .format("delta") \
    .option("mergeSchema", "true") \
    .mode("append") \
    .saveAsTable("Sales")

The new column can be incorporated into the table schema where supported.

Schema Evolution Warning

Schema evolution should be controlled. Don't automatically allow every source change into production. For example: Amount unexpectedly changing from: DOUBLE to: STRING could create downstream problems. A production pipeline should validate schema changes.

4.10.15 Overwrite Schema

When rebuilding a table with an intentionally changed schema, you may use:

df.write \
    .format("delta") \
    .option("overwriteSchema", "true") \
    .mode("overwrite") \
    .saveAsTable("Sales")

Use this carefully because the table schema is being replaced.

4.10.16 Write Metadata Columns

It is often useful to add audit columns before writing.

from pyspark.sql.functions import (
    current_timestamp,

lit

)

df = df \
    .withColumn(
        "LoadTimestamp",
        current_timestamp()
    ) \
    .withColumn(
        "SourceSystem",
        lit("SalesCSV")
    )

Then:

df.write \
    .format("delta") \
    .mode("append") \
    .saveAsTable("Sales")

Example:

These columns are valuable for:

• Auditing
• Troubleshooting
• Data lineage
• Operational monitoring

4.10.17 Write CSV → Delta

This is one of the most important Fabric notebook patterns. Read CSV

df = spark.read \
    .option("header", "true") \
    .option("inferSchema", "true") \
    .csv("Files/Raw/Sales.csv")

Clean

from pyspark.sql.functions import col, trim, upper

df = df \
    .withColumn(
        "CustomerName",
        trim(col("CustomerName"))
    ) \
    .withColumn(
        "Country",
        upper(trim(col("Country")))
    )

Remove duplicates

df = df.dropDuplicates(
    ["SaleID"]
)

Write Delta

df.write \
    .format("delta") \
    .mode("overwrite") \
    .saveAsTable("Sales")

4.10.18 Append-Based Daily Load

Suppose every day you receive: Sales_20260820.csv Sales_20260821.csv Sales_20260822.csv Process each file:

Day 20
   ↓
Delta + records

Day 21
   ↓
Delta + records

Day 22
   ↓
Delta + records

Code:

df.write \
    .format("delta") \
    .mode("append") \
    .saveAsTable("Sales")

This works well when the incoming records are genuinely new and duplicates are controlled.

4.10.19 Append vs MERGE

This distinction is extremely important. Append

df.write \
    .format("delta") \
    .mode("append") \
    .saveAsTable("Sales")

Use when: New records only

MERGE

Use when: New records + Updated records Conceptually:

Incoming
   │
   ▼

Existing Key?

   │
 ┌─┴─────┐
Yes      No
 │        │
 ▼        ▼

UPDATE   INSERT

4.10.20 Delta MERGE

For upserts:

from delta.tables import DeltaTable

target = DeltaTable.forName(
    spark,
    "Sales"
)

target.alias("target") \
    .merge(
        df.alias("source"),
        "target.SaleID = source.SaleID"
    ) \
    .whenMatchedUpdateAll() \
    .whenNotMatchedInsertAll() \
    .execute()

This is generally more appropriate than append when source records can change.

4.10.21 Full Refresh vs Incremental

Full Refresh

Source
  │
  ▼
Extract Everything
  │
  ▼
Transform
  │
  ▼
Overwrite Delta

Code:

.mode("overwrite")

Incremental

Source
  │
  ▼
New/Changed Records
  │
  ▼
Transform
  │
  ▼

MERGE Delta

Code:

MERGE

4.10.22 Write Delta with SQL

You can also create a Delta table using Spark SQL:

CREATE TABLE Sales
USING DELTA
AS
SELECT *
FROM sales_df;

If sales_df is registered as a temporary view:

df.createOrReplaceTempView("sales_df")

Then:

CREATE TABLE Sales
USING DELTA
AS
SELECT *
FROM sales_df;

4.10.23 INSERT INTO Delta

INSERT INTO Sales
SELECT *
FROM sales_df;

This is useful for SQL-based ingestion.

4.10.24 Updating Delta

Once the data is in Delta, you can use:

UPDATE Sales
SET Amount = Amount * 1.10
WHERE Country = 'INDIA';

4.10.25 Deleting Delta Records

DELETE FROM Sales
WHERE Amount < 0;

These operations are possible because Delta maintains transactional table state.

4.10.26 Write Performance

Writing Delta efficiently requires attention to:

• Number of partitions
• Number of files
• File sizes
• Data skew
• Partitioning
• Frequency of writes

Avoid:

100 records
   ↓
Write
   ↓
100 tiny files

Instead, batch appropriately:

1 million records
   ↓
Reasonable number of files
   ↓
Delta

4.10.27 Small Files Problem

Frequent tiny writes can create many small files:

Sales
├── part-0001.parquet
├── part-0002.parquet
├── part-0003.parquet
├── ...
└── part-50000.parquet

This can negatively affect performance. Use appropriate batching and Fabric/Delta optimization capabilities where applicable.

4.10.28 Repartition Before Write

You can control the number of output partitions:

df = df.repartition(10)

Then:

df.write \
    .format("delta") \
    .mode("append") \
    .saveAsTable("Sales")

Don't choose a number arbitrarily. The correct partition count depends on:

• Data size
• Cluster resources
• File sizes
• Workload

4.10.29 Coalesce Before Write

If you need fewer output partitions:

df = df.coalesce(4)

Then:

df.write \
    .format("delta") \
    .mode("append") \
    .saveAsTable("Sales")
coalesce() is commonly useful when reducing the number of partitions without a full redistribution.

4.10.30 Validate Delta After Writing

A strong pipeline should perform validation.

target_df = spark.table("Sales")

target_count = target_df.count()

print(
    f"Target records: {target_count}"
)

Check for duplicate business keys:

duplicates = target_df.groupBy(
    "SaleID"
).count().filter(
    col("count") > 1
)

display(duplicates)

Check NULL keys:

null_keys = target_df.filter(
    col("SaleID").isNull()
).count()

print(
    f"NULL SaleID count: {null_keys}"
)

4.10.31 Table History

After writing Delta, you can inspect table history:

DESCRIBE HISTORY Sales;

This can help you understand:

• Table versions
• Write operations
• Timestamps
• Operation details

4.10.32 Complete Production Pattern

                 CSV
                  │
                  ▼
             Read CSV
                  │
                  ▼
           Spark DataFrame
                  │
                  ▼
             Clean Data
                  │
                  ▼
          Data Validation
                  │
             ┌────┴────┐
             ▼         ▼
           Valid     Invalid
             │         │
             ▼         ▼
        Delta Table  Quarantine
             │
             ▼
          Validate
             │
             ▼
          Archive

4.10.33 Complete Example

from pyspark.sql.functions import (
    col,
    trim,
    upper,
    current_timestamp,

input_file_name

)

# 1. Read CSV
df = spark.read \
    .option("header", "true") \
    .option("inferSchema", "true") \
    .csv("Files/Raw/Sales.csv")

# 2. Clean
df = df \
    .withColumn(
        "CustomerName",
        trim(col("CustomerName"))
    ) \
    .withColumn(
        "Country",
        upper(trim(col("Country")))
    ) \
    .withColumn(
        "Amount",
        col("Amount").cast("double")
    )

# 3. Remove duplicates
df = df.dropDuplicates(
    ["SaleID"]
)

# 4. Add audit columns
df = df \
    .withColumn(
        "SourceFile",
        input_file_name()
    ) \
    .withColumn(
        "LoadTimestamp",
        current_timestamp()
    )

# 5. Validate
invalid_count = df.filter(
    col("SaleID").isNull()
).count()

if invalid_count > 0:
    raise Exception(
        f"Invalid records: {invalid_count}"
    )

# 6. Write Delta
df.write \
    .format("delta") \
    .mode("overwrite") \
    .saveAsTable("Sales")

# 7. Validate target
target_count = spark.table(
    "Sales"
).count()

print(
    f"Target record count: {target_count}"
)

4.10.34 Interview Questions

1. How do you write a DataFrame to Delta?

df.write \
    .format("delta") \
    .mode("overwrite") \
    .saveAsTable("Sales")

1. What is the difference between append and overwrite?

Append adds new data. Overwrite replaces the existing destination data.

1. When should you use MERGE instead of append?

Use MERGE when incoming data can contain both new and changed records.

1. What is saveAsTable()?

It writes the DataFrame as a table that can be accessed through Spark SQL/table APIs.

1. What is partitionBy()?

It organizes the output data based on partition column values.

.partitionBy("Country")

1. What is schema evolution?

It allows controlled changes to the Delta table schema, such as adding columns.

1. How do you enable schema merging?

.option("mergeSchema", "true")

1. How do you inspect Delta table history?

DESCRIBE HISTORY Sales;

1. Why are too many small files a problem?

They increase file and metadata overhead and can reduce query performance.

1. What should you validate after a Delta write?

At minimum: Record count Schema NULL business keys Duplicate business keys Data-quality rules

Hands-On Lab

Project: CSV → Clean Data → Delta
Input
Files/Raw/Sales.csv
Step 1 — Read

df = spark.read \
    .option("header", "true") \
    .option("inferSchema", "true") \
    .csv("Files/Raw/Sales.csv")

Step 2 — Clean

df = df \
    .withColumn(
        "CustomerName",
        trim(col("CustomerName"))
    ) \
    .withColumn(
        "Country",
        upper(trim(col("Country")))
    ) \
    .withColumn(
        "Amount",
        col("Amount").cast("double")
    )

Step 3 — Deduplicate

df = df.dropDuplicates(
    ["SaleID"]
)

Step 4 — Write Delta

df.write \
    .format("delta") \
    .mode("overwrite") \
    .saveAsTable("Sales")

Step 5 — Verify

display(
    spark.table("Sales")
)

Step 6 — Query

SELECT
    Country,
    COUNT(*) AS RecordCount,
    SUM(Amount) AS TotalSales
FROM Sales
GROUP BY Country
ORDER BY TotalSales DESC;

df.write \
    .format("delta") \
    .mode("overwrite") \
    .saveAsTable("Sales")

df.write \
    .format("delta") \
    .mode("append") \
    .saveAsTable("Sales")

df.write \
    .format("delta") \
    .option("mergeSchema", "true") \
    .mode("append") \
    .saveAsTable("Sales")

Incoming Data
     │
     ▼

   MERGE

UPDATE  INSERT

  \     /
   ▼   ▼
 Delta Table
Fabric Data Engineering Pattern
          OneLake
             │
             ▼
       Lakehouse Files
             │
             ▼
         Read CSV
             │
             ▼
       Spark DataFrame
             │
             ▼
        Clean Data
             │
             ▼
         Validate
             │
             ▼
        Write Delta
             │
             ▼
       Delta Table
             │
       ┌─────┴─────┐
       ▼           ▼
    Spark SQL   Power BI

Key principle: choose the write mode based on the business requirement—overwrite for full refreshes, append for genuinely new records, and MERGE for incremental workloads containing both inserts and updates.

Learning Objectives

By the end of this lesson, you will understand how to:

• Perform INSERT + UPDATE operations using Delta Lake.
• Understand Delta MERGE.
• Implement upsert logic in Microsoft Fabric.
• Update existing records and insert new records.
• Delete records conditionally.
• Handle incremental data loads.
• Avoid duplicate records.
• Build a production-ready Bronze → Silver Delta merge process.

4.11.1 What is Merge?

A MERGE operation compares incoming records with existing records using a matching condition. It can then:

• UPDATE records that already exist.
• INSERT records that don't exist.
• DELETE records when required.

In simple terms:

Incoming Data
      │
      ▼

   MERGE

      │
      ▼

Is key already present?

      │
 ┌────┴────┐
 │         │
YES       NO
 │         │
 ▼         ▼

UPDATE    INSERT

This is also called an upsert:

UPDATE + INSERT = UPSERT

4.11.2 Why Merge is Important

Suppose your existing Delta table contains:

Today's source contains:

We want:

Customer 101 → UPDATE
Customer 104 → INSERT

Final table:

This is exactly what Delta MERGE is designed for.

4.11.3 MERGE Architecture

A typical Fabric incremental pipeline:

Source System
     │
     ▼
Incremental Extract
     │
     ▼
Bronze Delta
     │
     ▼
Clean / Transform
     │
     ▼
Source DataFrame
     │
     ▼
Delta MERGE
     │
     ▼
Silver Delta Table

4.11.4 DeltaTable API

In a Fabric notebook, import:

from delta.tables import DeltaTable
Then get the target table:
target = DeltaTable.forName(
    spark,
    "Customers"
)

Now target represents the existing Delta table.

4.11.5 Basic MERGE Syntax

The general pattern is:

target.alias("target") \
    .merge(
        source.alias("source"),
        "target.CustomerID = source.CustomerID"
    ) \
    .whenMatchedUpdateAll() \
    .whenNotMatchedInsertAll() \
    .execute()

This means:

If CustomerID exists
        ↓

UPDATE

If CustomerID doesn't exist
        ↓

INSERT

4.11.6 Complete Example

Existing Delta Table Customers

Incoming DataFrame

source_df

Merge

from delta.tables import DeltaTable

target = DeltaTable.forName(
    spark,
    "Customers"
)

target.alias("target") \
    .merge(
        source_df.alias("source"),
        "target.CustomerID = source.CustomerID"
    ) \
    .whenMatchedUpdateAll() \
    .whenNotMatchedInsertAll() \
    .execute()

4.11.7 Result

Before: 101 John Hyderabad 5000 102 Anita Guntur 7000 103 David Chennai 6000 Incoming: 101 John Bengaluru 5500 104 Priya Hyderabad 8000 After:

101 John  Bengaluru 5500   ← UPDATED
102 Anita Guntur    7000
103 David Chennai   6000
104 Priya Hyderabad 8000   ← INSERTED

4.11.8 whenMatchedUpdateAll()

This updates all matching columns.

.whenMatchedUpdateAll()

Conceptually:

Target.CustomerID = Source.CustomerID
             │
             ▼
        Record exists
             │
             ▼
      Replace/update
       source values

It is convenient when source and target schemas align.

4.11.9 Explicit Update

Instead of updating every column, specify exactly what should change.

target.alias("target") \
    .merge(
        source_df.alias("source"),
        "target.CustomerID = source.CustomerID"
    ) \
    .whenMatchedUpdate(
        set={
            "CustomerName": "source.CustomerName",
            "City": "source.City",
            "Amount": "source.Amount"
        }
    ) \
    .whenNotMatchedInsert(
        values={
            "CustomerID": "source.CustomerID",
            "CustomerName": "source.CustomerName",
            "City": "source.City",
            "Amount": "source.Amount"
        }
    ) \
    .execute()

This gives you much more control.

4.11.10 Why Explicit Updates Are Useful

Suppose your target has audit columns: CustomerID CustomerName City Amount CreatedDate UpdatedDate SourceSystem You may not want to overwrite: CreatedDate when an existing customer is updated. Instead:

.whenMatchedUpdate(
    set={
        "CustomerName": "source.CustomerName",
        "City": "source.City",
        "Amount": "source.Amount",
        "UpdatedDate": "current_timestamp()"
    }
)

This preserves the original creation timestamp.

4.11.11 whenNotMatchedInsertAll()

For a new record:

.whenNotMatchedInsertAll()

inserts all compatible source columns. Conceptually:

Source CustomerID = 104
        │
        ▼
Target doesn't contain 104
        │
        ▼

INSERT

4.11.12 Explicit Insert

You can specify the columns:

.whenNotMatchedInsert(
    values={
        "CustomerID": "source.CustomerID",
        "CustomerName": "source.CustomerName",
        "City": "source.City",
        "Amount": "source.Amount"
    }
)

This is preferable when target and source structures aren't identical.

4.11.13 Conditional Update

You don't necessarily need to update every matching record. Example:

.whenMatchedUpdate(
    condition="source.Amount <> target.Amount",
    set={
        "Amount": "source.Amount",
        "UpdatedDate": "current_timestamp()"
    }
)

Now the update happens only if the amount has changed.

4.11.14 Conditional Insert

You can also apply conditions to inserts. Conceptually:

.whenNotMatchedInsert(
    condition="source.IsActive = true",
    values={
        "CustomerID": "source.CustomerID",
        "CustomerName": "source.CustomerName"
    }
)

This allows business rules to be incorporated into the merge.

4.11.15 Delete with MERGE

Delta MERGE can also support conditional deletes. Example:

target.alias("target") \
    .merge(
        source_df.alias("source"),
        "target.CustomerID = source.CustomerID"
    ) \
    .whenMatchedDelete(
        condition="source.IsDeleted = true"
    ) \
    .whenMatchedUpdateAll() \
    .whenNotMatchedInsertAll() \
    .execute()

Conceptually:

Source
 │
 ├── IsDeleted = TRUE
 │        ↓
 │      DELETE
 │
 ├── Existing record
 │        ↓
 │      UPDATE
 │
 └── New record
          ↓

        INSERT

4.11.16 Soft Delete

Instead of physically deleting records, many enterprise systems use a soft delete. Example: IsActive = false Instead of:

DELETE

use:

.whenMatchedUpdate(
    condition="source.IsDeleted = true",
    set={
        "IsActive": "false",
        "UpdatedDate": "current_timestamp()"
    }
)

This preserves history.

4.11.17 MERGE with Business Key

The merge condition should normally use a reliable business key. Example: "target.CustomerID = source.CustomerID" For a composite key: "target.CustomerID = source.CustomerID AND " \ "target.ProductID = source.ProductID" Example: CustomerID + ProductID may uniquely identify a customer-product relationship.

4.11.18 Importance of the MERGE Key

A bad merge condition can cause serious data problems. Bad: "target.Country = source.Country" Many records may have: Country = INDIA This does not uniquely identify a record. Better: "target.CustomerID = source.CustomerID" The key should uniquely identify the business entity or transaction being merged.

4.11.19 Duplicate Source Records

This is a very important production issue. Suppose the source contains:

The same target record matches multiple source records. This can cause the MERGE to fail or produce ambiguous behavior depending on the operation and runtime. Therefore: Deduplicate the source before MERGE.

4.11.20 Deduplicate Source Data

Example:

source_df = source_df.dropDuplicates(
    ["CustomerID"]
)
For more sophisticated logic, choose the latest record.

Suppose: CustomerID | Amount | UpdatedDate 101 | 5000 | 2026-08-20 101 | 5500 | 2026-08-22 We want the latest record. Use a window:

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

window_spec = Window \
    .partitionBy("CustomerID") \
    .orderBy(
        col("UpdatedDate").desc()
    )

source_df = source_df \
    .withColumn(
        "rn",
        row_number().over(window_spec)
    ) \
    .filter(
        col("rn") == 1
    ) \
    .drop("rn")

Now only the latest record remains.

4.11.21 Full Incremental Pipeline

A realistic process:

             Source System
                   │
                   ▼
            Incremental Data
                   │
                   ▼
               CSV/API
                   │
                   ▼
             Bronze Delta
                   │
                   ▼
             Clean Data
                   │
                   ▼
            Deduplicate
                   │
                   ▼

              MERGE

/ \ / \

       Existing     New
          │           │
          ▼           ▼

       UPDATE       INSERT

\ /

             \     /
              ▼   ▼
           Silver Delta

4.11.22 CSV → MERGE Example

Suppose: Files/Raw/Customers.csv Read:

source_df = spark.read \
    .option("header", "true") \
    .option("inferSchema", "true") \
    .csv("Files/Raw/Customers.csv")

Clean:

from pyspark.sql.functions import (
    col,
    trim,

upper

)

source_df = source_df \
    .withColumn(
        "CustomerName",
        trim(col("CustomerName"))
    ) \
    .withColumn(
        "Country",
        upper(trim(col("Country")))
    )

Deduplicate:

source_df = source_df.dropDuplicates(
    ["CustomerID"]
)

Get target:

target = DeltaTable.forName(
    spark,
    "Customers"
)

Merge:

target.alias("target") \
    .merge(
        source_df.alias("source"),
        "target.CustomerID = source.CustomerID"
    ) \
    .whenMatchedUpdateAll() \
    .whenNotMatchedInsertAll() \
    .execute()

4.11.23 MERGE with Audit Columns

Suppose the target contains: CustomerID CustomerName Country Amount CreatedDate UpdatedDate Use explicit mappings:

target.alias("target") \
    .merge(
        source_df.alias("source"),
        "target.CustomerID = source.CustomerID"
    ) \
    .whenMatchedUpdate(
        set={
            "CustomerName": "source.CustomerName",
            "Country": "source.Country",
            "Amount": "source.Amount",
            "UpdatedDate": "current_timestamp()"
        }
    ) \
    .whenNotMatchedInsert(
        values={
            "CustomerID": "source.CustomerID",
            "CustomerName": "source.CustomerName",
            "Country": "source.Country",
            "Amount": "source.Amount",
            "CreatedDate": "current_timestamp()",
            "UpdatedDate": "current_timestamp()"
        }
    ) \
    .execute()

This is a common enterprise pattern.

4.11.24 SCD Type 1

MERGE is commonly used to implement Slowly Changing Dimension Type 1.

Type 1 means: Keep the latest value and overwrite the old value. Example: Before:

CustomerID = 101 City = Hyderabad New source: CustomerID = 101 City = Bengaluru After MERGE: CustomerID = 101 City = Bengaluru The old value is not retained.

4.11.25 SCD Type 2

SCD Type 2 keeps historical versions. Example: CustomerID | City | StartDate | EndDate | IsCurrent 101 | Hyderabad | 2026-01-01 | 2026-08-21 | false 101 | Bengaluru | 2026-08-22 | NULL | true The implementation is more complex than a basic MERGE because you generally need to:

1. Expire the existing version.
2. Insert the new version.

Conceptually:

Existing Current Record
          │
          ▼
       Expire
          │
          ▼
Insert New Version

SCD Type 2 will be covered in greater depth when dimensional modeling is introduced.

4.11.26 MERGE Performance

MERGE can be expensive on very large Delta tables.

Performance depends on:

• Target table size
• Merge key
• Partitioning
• Data distribution
• Number of source records
• Number of files
• File organization

A good design reduces the amount of target data that needs to be examined.

4.11.27 Avoid Full-Table MERGE When Possible

Suppose: Target = 1 billion records Source = 10,000 records You don't want an inefficient process that unnecessarily scans huge amounts of data. Use appropriate:

• Partitioning
• Filtering
• Incremental extraction
• Data organization

to reduce the work required.

4.11.28 MERGE Validation

After the merge, validate: Record count

result_df = spark.table("Customers")

print(
    "Total records:",
    result_df.count()
)

Duplicate keys

duplicates = result_df.groupBy(
    "CustomerID"
).count().filter(
    col("count") > 1
)

display(duplicates)

Expected: No duplicate CustomerID values

4.11.29 Verify Updated Records

Suppose CustomerID 101 was updated.

display(
    spark.table("Customers")
    .filter(
        col("CustomerID") == 101
    )
)

Check: CustomerID = 101 City = Bengaluru

4.11.30 Verify Inserted Records

For CustomerID 104:

display(
    spark.table("Customers")
    .filter(
        col("CustomerID") == 104
    )
)

Expected: CustomerID = 104

4.11.31 MERGE vs Append

4.11.32 MERGE vs Overwrite

4.11.33 Production Best Practices

1. Use a reliable business key

CustomerID OrderID ProductID

1. Deduplicate the source

Do this before MERGE.

1. Validate source schema

Make sure expected columns exist.

1. Validate NULL keys

Don't allow: CustomerID = NULL into the merge source unless explicitly designed for it.

1. Use explicit mappings when necessary

Avoid blindly updating audit columns.

1. Track audit metadata

Useful columns: CreatedDate UpdatedDate SourceSystem BatchID

1. Make the process idempotent

Running the same batch twice should not create duplicate records.

1. Validate after MERGE

Check:

• Row counts
• Duplicate keys
• NULL keys
• Updated records
• Inserted records

4.11.34 Complete Production Example

from delta.tables import DeltaTable
from pyspark.sql.functions import (
    col,
    trim,
    upper,

current_timestamp

)

# 1. Read incremental source
source_df = spark.read \
    .option("header", "true") \
    .option("inferSchema", "true") \
    .csv("Files/Raw/Customers.csv")

# 2. Clean
source_df = source_df \
    .withColumn(
        "CustomerName",
        trim(col("CustomerName"))
    ) \
    .withColumn(
        "Country",
        upper(trim(col("Country")))
    )

# 3. Validate key
source_df = source_df.filter(
    col("CustomerID").isNotNull()
)

# 4. Deduplicate source
source_df = source_df.dropDuplicates(
    ["CustomerID"]
)

# 5. Get target Delta table
target = DeltaTable.forName(
    spark,
    "Customers"
)

# 6. MERGE
target.alias("target") \
    .merge(
        source_df.alias("source"),
        "target.CustomerID = source.CustomerID"
    ) \
    .whenMatchedUpdate(
        set={
            "CustomerName": "source.CustomerName",
            "Country": "source.Country",
            "Amount": "source.Amount",
            "UpdatedDate": "current_timestamp()"
        }
    ) \
    .whenNotMatchedInsert(
        values={
            "CustomerID": "source.CustomerID",
            "CustomerName": "source.CustomerName",
            "Country": "source.Country",
            "Amount": "source.Amount",
            "CreatedDate": "current_timestamp()",
            "UpdatedDate": "current_timestamp()"
        }
    ) \
    .execute()

# 7. Validate target
result_df = spark.table("Customers")

print(
    "Target records:",
    result_df.count()
)

4.11.35 Hands-On Lab

Scenario You have an existing Delta table: Customers Existing data: CustomerID,CustomerName,Country,Amount 101,John,India,5000 102,Anita,India,7000 103,David,UK,6000 New file: CustomerID,CustomerName,Country,Amount 101,John,India,5500 104,Priya,India,8000 Step 1 — Read the file

source_df = spark.read \
    .option("header", "true") \
    .option("inferSchema", "true") \
    .csv("Files/Raw/Customers.csv")

Step 2 — Deduplicate

source_df = source_df.dropDuplicates(
    ["CustomerID"]
)

Step 3 — Get Delta table

from delta.tables import DeltaTable

target = DeltaTable.forName(
    spark,
    "Customers"
)

Step 4 — MERGE

target.alias("target") \
    .merge(
        source_df.alias("source"),
        "target.CustomerID = source.CustomerID"
    ) \
    .whenMatchedUpdateAll() \
    .whenNotMatchedInsertAll() \
    .execute()

Step 5 — Verify

display(
    spark.table("Customers")
)

Expected:

101 | John  | India | 5500  ← UPDATE

102 | Anita | India | 7000 103 | David | UK | 6000

104 | Priya | India | 8000  ← INSERT

4.11.36 Interview Questions

1. What is Delta MERGE?

A transactional operation that can perform updates, inserts, and conditional deletes by matching source records with target records.

1. What is an upsert?

An operation that performs:

UPDATE + INSERT

1. What is the basic MERGE pattern?

target.alias("target") \
    .merge(
        source.alias("source"),
        "target.ID = source.ID"
    ) \
    .whenMatchedUpdateAll() \
    .whenNotMatchedInsertAll() \
    .execute()

1. Why should the source be deduplicated?

Multiple source records matching the same target key can cause ambiguous MERGE behavior.

1. What is whenMatchedUpdateAll()?

It updates all compatible target columns using the matching source record.

1. What is whenNotMatchedInsertAll()?

It inserts a source record when no matching target record exists.

1. When should you use MERGE instead of APPEND?

When incoming data contains both new records and changes to existing records.

1. What is SCD Type 1?

It overwrites the existing dimension value with the latest value.

1. What is SCD Type 2?

It preserves historical versions of dimension records.

1. What is the most important part of a MERGE?

The matching condition/business key. For example: "target.CustomerID = source.CustomerID"

If the key is incorrect, the MERGE can update or insert the wrong records.

                Source
                  │
                  ▼
             DataFrame
                  │
                  ▼
              Validate
                  │
                  ▼
             Deduplicate
                  │
                  ▼

                MERGE

       Match           No Match
         │                │
         ▼                ▼

      UPDATE            INSERT

         │                │
         └───────┬────────┘
                 ▼
            Delta Table

from delta.tables import DeltaTable

target = DeltaTable.forName(
    spark,
    "Customers"
)

target.alias("target") \
    .merge(
        source_df.alias("source"),
        "target.CustomerID = source.CustomerID"
    ) \
    .whenMatchedUpdateAll() \
    .whenNotMatchedInsertAll() \
    .execute()

Learning Objectives

By the end of this lesson, you will be able to:

• Understand data partitioning in Apache Spark.
• Understand why partitioning improves distributed processing.
• Differentiate Spark partitions from Delta table partitions.
• Use repartition() and coalesce().
• Partition Delta tables using partitionBy().
• Choose suitable partition columns.
• Avoid over-partitioning and small-file problems.
• Understand partition pruning.
• Build an efficient Fabric Lakehouse partitioning strategy.

4.12.1 What is Data Partitioning?

Partitioning means dividing a large dataset into smaller pieces so that Spark can process those pieces in parallel. Instead of:

10 Million Records
       │
       ▼
 One Large Dataset

Spark can divide the data:

10 Million Records
       │
       ▼
 ┌─────┼─────┬─────┐
 ▼     ▼     ▼     ▼
P1    P2    P3    P4

Different Spark tasks can process these partitions simultaneously.

4.12.2 Why Partition Data?

Partitioning helps with:

• Parallel processing
• Distributed computation
• Query performance
• Data organization
• Large-scale ingestion
• Filtering large datasets

Example:

1 Billion Records
       │
       ▼
Distributed across
multiple partitions
       │
       ▼
Multiple Spark tasks
       │
       ▼
Faster processing

4.12.3 Two Types of Partitioning

This is one of the most important concepts. In Fabric/Spark, you will encounter:

1. Spark partitions

These control how data is distributed for Spark processing. Examples: repartition() coalesce()

1. Table partitions

These control how data is physically organized in a Delta table. Example: partitionBy("Year") They are related but not the same thing.

4.12.4 Spark Partitions

Suppose: df contains: 10 million records Spark may divide it into: Partition 1 Partition 2 Partition 3 ... Partition N Each partition can be processed by a Spark task.

Driver
  │
  ▼
DataFrame
  │
  ├── Partition 1 → Task 1
  ├── Partition 2 → Task 2
  ├── Partition 3 → Task 3
  └── Partition 4 → Task 4

4.12.5 Check Number of Partitions

You can check:

df.rdd.getNumPartitions()

For example: 8 means the DataFrame currently has eight Spark partitions.

4.12.6 repartition()

Use repartition() to redistribute data. Example:

df = df.repartition(8)

Now the DataFrame is distributed into approximately eight partitions.

4.12.7 Repartition by Column

You can partition based on a column:

df = df.repartition(
    "Country"
)
Or specify both number and column:
df = df.repartition(
    8,
    "Country"
)

Conceptually:

Country
   │
   ├── INDIA
   ├── USA
   ├── UK
   └── Canada

Spark redistributes records according to the partitioning expression.

4.12.8 Why Repartition Can Be Expensive

repartition() generally causes a shuffle. Example:

Partition 1 ──┐
Partition 2 ──┼── Shuffle ──► New Partitions
Partition 3 ──┤
Partition 4 ──┘

Data may need to move between executors. Therefore: Don't use repartition() unnecessarily.

4.12.9 coalesce()

coalesce() is primarily used to reduce the number of partitions. Example:

df = df.coalesce(4)

If you have: 20 partitions you can reduce them to: 4 partitions It generally avoids a full shuffle when reducing partitions.

4.12.10 Repartition vs Coalesce

Example:

df.repartition(20)

versus:

df.coalesce(5)

4.12.11 Delta Table Partitioning

Now consider physical table partitioning. You can write:

df.write \
    .format("delta") \
    .partitionBy("Country") \
    .mode("overwrite") \
    .saveAsTable("Sales")

Conceptually:

Sales/
│
├── Country=INDIA/
│
├── Country=USA/
│
├── Country=UK/
│
└── Country=CANADA/

The exact physical layout is managed by the Lakehouse/Delta engine, but the partitioning concept is based on the partition column values.

4.12.12 Why Partition a Delta Table?

Suppose you have: 1 Billion Sales Records and most queries are:

SELECT *
FROM Sales
WHERE Country = 'INDIA';
If Country is an appropriate partition column, Spark can potentially avoid reading unrelated partitions.

This is called: Partition pruning

4.12.13 Partition Pruning

Without useful partitioning:

Sales
│
├── Read INDIA
├── Read USA
├── Read UK
├── Read Canada
└── Read Australia

With appropriate partitioning:

WHERE Country = 'INDIA'
              │
              ▼
        Read INDIA only

Conceptually:

                 Query
                   │
                   ▼
            Country = INDIA
                   │
                   ▼
             Partition Pruning
                   │
                   ▼
             INDIA partition

This can reduce the amount of data scanned.

4.12.14 Good Partition Columns

Good candidates are usually:

• Frequently filtered columns
• Relatively low-to-moderate cardinality columns
• Columns that divide data into reasonably large chunks

Examples: Year Month Region Country BusinessUnit But suitability depends heavily on the actual data and query patterns.

4.12.15 Bad Partition Columns

Avoid partitioning on very high-cardinality columns such as: CustomerID TransactionID Email PhoneNumber Suppose: CustomerID = 10 million unique values Partitioning by CustomerID could create an enormous number of partitions. This leads to: Small-file / excessive-partition problems

4.12.16 Cardinality

Cardinality means the number of distinct values in a column. Example: Country -------- India USA UK Canada Cardinality: 4 CustomerID: 101 102 103 104 ... Could have: 10 million distinct values. Therefore:

Country       → Low cardinality
CustomerID    → High cardinality

4.12.17 Partition by Date

Date-based partitioning is common for large fact tables. For example: Year or: Year + Month Example:

from pyspark.sql.functions import year, month

df = df \
    .withColumn(
        "Year",
        year("OrderDate")
    ) \
    .withColumn(
        "Month",
        month("OrderDate")
    )

Then:

df.write \
    .format("delta") \
    .partitionBy(
        "Year",
        "Month"
    ) \
    .mode("overwrite") \
    .saveAsTable("Sales")

Conceptually:

Sales
│
├── Year=2025
│   ├── Month=1
│   ├── Month=2
│   └── ...
│
└── Year=2026
    ├── Month=1
    ├── Month=2
    └── ...

4.12.18 Be Careful with Date Partitioning

Don't automatically create: Year Month Day Hour Minute Second as partitions. This can create too many small partitions. A better strategy might be: Year + Month for a large historical fact table, depending on workload.

4.12.19 Example: Sales Table

Suppose: Sales 1 Billion records Queries are usually: WHERE OrderDate >= '2026-08-01' AND OrderDate < '2026-09-01' A date-oriented layout may be beneficial. For example:

Year=2026
   │
   ├── Month=07
   └── Month=08

The query can potentially focus on August data.

4.12.20 Partitioning vs Filtering

These are different. Filtering:

df.filter(
    col("Country") == "INDIA"
)

selects records from a DataFrame. Partitioning:

df.write \
    .partitionBy("Country")

organizes the physical table layout.

4.12.21 Partitioning vs Sorting

Partitioning:

Country
├── INDIA
├── USA
└── UK

Sorting: Records within data files organized according to a sort/order strategy They solve different problems.

4.12.22 Spark Partitioning Example

Create a DataFrame:

df = spark.range(0, 1000000)

Check partitions:

print(
    df.rdd.getNumPartitions()
)

Repartition:

df2 = df.repartition(8)

Check again:

print(
    df2.rdd.getNumPartitions()
)

4.12.23 Repartition Before Writing

Sometimes you may want to control Spark's output parallelism:

df = df.repartition(8)

df.write \
    .format("delta") \
    .mode("append") \
    .saveAsTable("Sales")

But remember: Spark partition count and Delta table partitioning are separate concepts.

4.12.24 Partitioning and Small Files

Suppose you partition by: CustomerID and have: 1,000,000 customers You could end up with an undesirable number of tiny data files. Conceptually: CustomerID=1 CustomerID=2 CustomerID=3 ... CustomerID=1000000 This is usually a bad partitioning strategy.

4.12.25 The Small Files Problem

Too many small files can cause:

• Metadata overhead
• Slower queries
• Slower table operations
• More file management overhead
• Inefficient scans

Bad:

10 million records
       │
       ▼
1 million tiny files

Better:

10 million records
       │
       ▼
Reasonable number of well-sized files

4.12.26 Partition Skew

Partitioning can also create data skew. Suppose: Country INDIA = 900 million records USA = 50 million UK = 30 million Other = 20 million Partition sizes become:

INDIA → Huge
USA   → Medium
UK    → Small
Other → Small

One task may have much more work than others. Conceptually:

Task 1 → 900M records  ← Slow
Task 2 → 50M
Task 3 → 30M
Task 4 → 20M

This is called data skew.

4.12.27 How to Handle Skew

Possible approaches include:

• Choose a better partition strategy.
• Avoid extremely skewed partition columns.
• Repartition for Spark processing.
• Use appropriate data distribution techniques.
• Optimize the query/workload.

Don't assume that a column with low cardinality is automatically a good partition column.

4.12.28 Partitioning a Delta Table

Example:

df.write \
    .format("delta") \
    .partitionBy("Year") \
    .mode("overwrite") \
    .saveAsTable("Sales")

For multiple columns:

df.write \
    .format("delta") \
    .partitionBy(
        "Year",
        "Month"
    ) \
    .mode("overwrite") \
    .saveAsTable("Sales")

4.12.29 Querying a Partitioned Table

Suppose: Sales partitioned by Year Query:

SELECT
    COUNT(*)
FROM Sales
WHERE Year = 2026;

The engine can potentially prune partitions that don't contain 2026.

4.12.30 Don't Partition Every Table

Partitioning is not automatically beneficial. For small tables: 100,000 records partitioning may add unnecessary complexity. For example: Customer Dimension may not need physical partitioning at all. A large fact table may benefit more.

4.12.31 Typical Fabric Lakehouse Architecture

A common architecture:

                  OneLake
                     │
             ┌───────┴───────┐
             ▼               ▼
          Bronze           Silver
             │               │
             ▼               ▼
         Raw Delta       Clean Delta
                             │
                             ▼
                          Gold
                             │
                             ▼
                       BI / Analytics

Partitioning should be designed primarily where it provides measurable benefit.

4.12.32 Practical Example

Suppose we have: Sales Columns: SaleID CustomerID ProductID OrderDate Country Amount Queries frequently filter by: OrderDate Country A possible strategy: Partition: Year Month and use: Country as a regular filtering column. You don't necessarily want: CustomerID ProductID as physical partitions.

4.12.33 Complete Example

Step 1 — Read data

df = spark.read \
    .format("delta") \
    .table("RawSales")

Step 2 — Create date attributes

from pyspark.sql.functions import (
    year,

month

)

df = df \
    .withColumn(
        "Year",
        year("OrderDate")
    ) \
    .withColumn(
        "Month",
        month("OrderDate")
    )

Step 3 — Write partitioned Delta

df.write \
    .format("delta") \
    .partitionBy(
        "Year",
        "Month"
    ) \
    .mode("overwrite") \
    .saveAsTable("Sales")

Step 4 — Query

SELECT
    Country,
    SUM(Amount) AS TotalSales
FROM Sales
WHERE Year = 2026
  AND Month = 8
GROUP BY Country;

The partition predicates can help reduce the amount of data scanned.

4.12.34 repartition() vs partitionBy()

This is a very common interview question. repartition() Controls Spark DataFrame partitions:

df = df.repartition(

10

)

Purpose: Spark processing partitionBy() Controls physical partitioning of the output table:

df.write \
    .partitionBy("Year") \
    .saveAsTable("Sales")

Purpose: Storage organization

4.12.35 coalesce() vs repartition()

repartition()

df.repartition(10)

• Can increase partitions.
• Can decrease partitions.
• Usually causes a shuffle.

coalesce()

df.coalesce(5)

• Primarily decreases partitions.
• Usually avoids a full shuffle.
• Useful before writing when you have too many partitions.

4.12.36 Production Checklist

Before partitioning a Delta table, ask:

1. Is the table large enough?

If not, don't add unnecessary partitions.

1. What columns are commonly filtered?

Use workload patterns.

1. What is the cardinality?

Avoid millions of partition values.

1. Is the data evenly distributed?

Check for skew.

1. Will partitions create tiny files?

If yes, reconsider the design.

1. Is partition pruning actually useful?

Test query performance.

4.12.37 Interview Questions

1. What is data partitioning?

Dividing data into smaller units so it can be processed or stored efficiently.

1. What is a Spark partition?

A logical chunk of data processed by a Spark task.

1. What does repartition() do?

Redistributes data into a specified number of Spark partitions, generally involving a shuffle.

1. What does coalesce() do?

Reduces the number of Spark partitions, generally avoiding a full shuffle.

1. What does partitionBy() do?

Defines physical partitioning when writing data to a table/storage format.

1. What is partition pruning?

The ability of the query engine to skip irrelevant physical partitions based on filter predicates.

1. What is cardinality?

The number of distinct values in a column.

1. Why is CustomerID often a bad partition column?

Because it may have very high cardinality, potentially creating many small partitions/files.

1. What is data skew?

Uneven distribution of data across partitions, causing some Spark tasks to process substantially more data than others.

1. Are Spark partitions and Delta table partitions the same?

No.

Spark partitions → Processing
Delta partitions → Storage organization

Hands-On Lab

Project: Partition a Sales Delta Table Input RawSales Columns: SaleID CustomerID OrderDate Country Amount Step 1 — Read

df = spark.read.table(
    "RawSales"
)

Step 2 — Add Year and Month

from pyspark.sql.functions import (
    year,

month

)

df = df \
    .withColumn(
        "Year",
        year("OrderDate")
    ) \
    .withColumn(
        "Month",
        month("OrderDate")
    )

Step 3 — Write Partitioned Delta

df.write \
    .format("delta") \
    .partitionBy(
        "Year",
        "Month"
    ) \
    .mode("overwrite") \
    .saveAsTable("Sales")

Step 4 — Query

SELECT
    Country,
    SUM(Amount) AS TotalSales
FROM Sales
WHERE Year = 2026
  AND Month = 8
GROUP BY Country;

Final Summary

Think about partitioning at two different levels:

                PARTITIONING
                     │
          ┌──────────┴──────────┐
          ▼                     ▼
   Spark Partition         Table Partition
          │                     │
          ▼                     ▼
     Processing              Storage
          │                     │
   repartition()            partitionBy()
   coalesce()
Most important commands

# Check Spark partitions
df.rdd.getNumPartitions()

# Increase/reorganize Spark partitions
df.repartition(10)

# Reduce Spark partitions
df.coalesce(4)

# Partition Delta table
df.write \
    .format("delta") \
    .partitionBy("Year", "Month") \
    .saveAsTable("Sales")

Golden Rule Do not partition a Delta table just because you can. Choose partition columns based on table size, query patterns, cardinality, data distribution, and file size. For modern Fabric Lakehouse workloads, physical partitioning is only one optimization technique; don't assume that adding more partitions automatically makes queries faster.