Deep dive · Engine skill

PySpark deep dive

The exam shows you PySpark and asks what it does — or what is wrong with it. This page is the code-first companion to Domain 2: Spark's execution model, the DataFrame API, Delta Lake operations, structured streaming and the performance knobs the exam loves to test.

How this page fits the exam

DP-700 rarely asks you to write Spark from a blank cell, but it constantly asks you to read a snippet and pick the correct or corrected version — the right write mode, the deterministic dedup, the join that avoids a shuffle, the missing watermark. Everything here is grounded in the official Fabric Spark and Apache Spark documentation. Pair it with Domain 2 (loading patterns) and Domain 3 (Spark optimization).

On this page:

The Spark execution model — driver, executors, partitions, lazy evaluation
Reading & writing data — Delta read/write, save modes, Spark SQL vs the DataFrame API
Core DataFrame transformations — select, filter, withColumn, groupBy, agg
Joins — join types and broadcast joins
Window functions — ranking, lag/lead, running aggregates
Delta Lake operations — MERGE, time travel, OPTIMIZE, VACUUM
Structured streaming — readStream, watermarks, output modes, checkpoints
Performance & optimization — shuffle, skew, AQE, caching, V-Order
Scenario quick reference

The Spark execution model

A Fabric notebook running the Spark kernel is a distributed program. One driver plans the work and coordinates; many executors do the work in parallel across a pool of worker nodes. Data is split into partitions, and each partition is processed by one task on one core.

flowchart TD D["Driver
builds the plan, schedules tasks"] --> E1["Executor 1
cores run tasks"] D --> E2["Executor 2
cores run tasks"] D --> E3["Executor N
cores run tasks"] E1 --> P1["Partitions
unit of parallelism"] E2 --> P2["Partitions"] E3 --> P3["Partitions"]

Partitions are the unit of parallelism: Spark runs as many tasks at once as there are free cores across the executors.

Transformations vs actions

Spark uses lazy evaluation. Transformations only build up a plan; nothing runs until an action forces execution. This is why a cell with ten withColumn calls returns instantly, and the one .count() at the end takes a minute.

Kind	Examples	Behaviour
Transformation (lazy)	`select`, `filter`, `withColumn`, `join`, `groupBy().agg()`, `orderBy`, `dropDuplicates`	Returns a new DataFrame describing what to do; runs nothing yet
Narrow transformation	`select`, `filter`, `withColumn`	Each output partition depends on one input partition — no data movement
Wide transformation	`groupBy`, `join`, `orderBy`, `distinct`	Requires a shuffle — data is repartitioned across the network
Action (eager)	`count`, `collect`, `show`, `write`, `save`, `toPandas`, `first`	Triggers the whole plan to actually execute

Caveat — nothing runs until an action

If a question says "the read cell finished instantly but the write cell took ten minutes", that is lazy evaluation, not a bug — the read was a transformation and the write is the action that finally executed the whole chain. Errors in a transformation (a bad column name) also surface only when the action runs.

Reading & writing data

Delta is the default table format in Fabric. Tables live under a Lakehouse's Tables/ folder; files live under Files/. You can address a table by name (managed) or by path.

PySpark · read and write Delta

# Read a managed Delta table by name…
df = spark.read.table("bronze_sales")

# …or by format + path
df = spark.read.format("delta").load("Tables/bronze_sales")

# Read raw files (schema inference is expensive on big JSON — supply a schema)
raw = spark.read.option("header", "true").csv("Files/landing/*.csv")

# Write as a managed Delta table
df.write.format("delta").mode("overwrite").saveAsTable("silver_sales")

# Write to an explicit path under Tables/
df.write.format("delta").mode("append").save("Tables/silver_sales")

Save mode	Effect if the table exists
`append`	Adds new rows, keeps existing data
`overwrite`	Replaces all data; enforces the existing schema unless overridden
`error` / `errorifexists` (default)	Fails if the table already exists
`ignore`	Does nothing if the table already exists

Caveat — overwrite vs overwriteSchema vs mergeSchema

mode("overwrite") replaces the data but still enforces the current schema — a changed column set fails. Add .option("overwriteSchema","true") to replace the schema too. For additive changes (new columns only), prefer .option("mergeSchema","true"), which appends the new columns without discarding the old definition. This exact distinction appears on the exam.

Spark SQL vs the DataFrame API

The two are interchangeable and compile to the same plan through the Catalyst optimizer. Use whichever reads more clearly; you must be able to read both on the exam.

PySpark · same query, two styles

# DataFrame API
from pyspark.sql.functions import col, sum as _sum
result = (df.filter(col("amount") > 0)
            .groupBy("region")
            .agg(_sum("amount").alias("total")))

# Spark SQL — register or reference the table by name
result = spark.sql("""
    SELECT region, SUM(amount) AS total
    FROM   bronze_sales
    WHERE  amount > 0
    GROUP BY region
""")

display(result)   # Fabric's rich table/chart renderer

Core DataFrame transformations

The everyday verbs. Almost every snippet on the exam is built from these.

Operation	What it does
`select(...)`	Choose / compute columns
`filter(...)` / `where(...)`	Keep rows matching a predicate (identical operators)
`withColumn("c", expr)`	Add or replace one column
`withColumnRenamed(a, b)`	Rename a column
`drop("c")`	Remove a column
`groupBy(...).agg(...)`	Aggregate by keys
`orderBy(...)` / `sort(...)`	Sort (a wide, shuffling operation)
`dropDuplicates([keys])`	Remove duplicate rows (arbitrary survivor — see caveat)
`fillna(...)` / `na.fill(...)`	Replace nulls
`union(...)`	Stack two DataFrames with the same schema

PySpark · a typical clean-and-shape chain

from pyspark.sql.functions import col, year, month, upper, coalesce, lit

clean = (df
    .filter(col("amount") > 0)                        # drop invalid rows
    .withColumn("order_year", year("order_date"))       # derive columns
    .withColumn("order_month", month("order_date"))
    .withColumn("region", coalesce(col("region"), lit("UNKNOWN")))
    .withColumnRenamed("cust_id", "customer_id")
    .select("customer_id", "order_year", "order_month", "region", "amount"))

Column references: col("x"), "x", and df.x

Most transformations accept a plain string column name. You need col("x") (or df["x"]) when you build an expression — comparisons, arithmetic, .alias(), .desc(). Prefer built-in functions from pyspark.sql.functions over Python UDFs: built-ins run inside the JVM and are optimised by Catalyst, while a Python UDF runs in a separate Python process and blocks many optimisations.

Joins

Join type	Keeps
`inner` (default)	Rows with a match on both sides
`left` / `left_outer`	All left rows; nulls where the right has no match
`right` / `right_outer`	All right rows; nulls where the left has no match
`full` / `full_outer`	All rows from both sides
`left_semi`	Left rows that have a match — left columns only (a filter)
`left_anti`	Left rows with no match — left columns only (an anti-filter)

PySpark · join, and broadcast the small side

from pyspark.sql.functions import broadcast

# Standard join on a shared key
enriched = fact.join(dim_customer, on="customer_id", how="left")

# Different column names → explicit condition
enriched = fact.join(dim, fact.cust_id == dim.id, how="inner")

# Broadcast a small dimension to every executor → no shuffle of the big fact
enriched = fact.join(broadcast(dim_customer), on="customer_id", how="left")

Caveat — broadcast the small side to kill the shuffle

A default join shuffles both sides across the network on the join key — expensive when one side is a huge fact table. If one side is small (a dimension, a lookup), broadcast() ships it to every executor so the big side never moves. Adaptive Query Execution can do this automatically at runtime when it sees the small side is under spark.sql.autoBroadcastJoinThreshold. "Join is slow because of a large shuffle, one table is small" → broadcast join.

Window functions

Window functions compute a value per row over a group of related rows, without collapsing them the way groupBy does. They are the exam's favourite tool for deterministic deduplication and running calculations.

PySpark · ranking, dedup, and running totals

from pyspark.sql.window import Window
from pyspark.sql.functions import row_number, rank, dense_rank, lag, sum as _sum, col

# Keep only the newest row per order_id — deterministic dedup
w = Window.partitionBy("order_id").orderBy(col("modified_at").desc())
latest = (df.withColumn("rn", row_number().over(w))
            .filter(col("rn") == 1)
            .drop("rn"))

# Running total of amount per customer, ordered by date
wr = (Window.partitionBy("customer_id").orderBy("order_date")
            .rowsBetween(Window.unboundedPreceding, Window.currentRow))
running = df.withColumn("running_total", _sum("amount").over(wr))

# Previous order amount for each customer
prev = df.withColumn("prev_amount", lag("amount", 1).over(
           Window.partitionBy("customer_id").orderBy("order_date")))

Function	Returns
`row_number()`	Unique 1,2,3… per partition — no ties, ideal for "keep the latest"
`rank()`	Ties share a rank, then it skips (1,1,3)
`dense_rank()`	Ties share a rank, no gap (1,1,2)
`lag()` / `lead()`	Value from a previous / following row
`sum/avg/count().over(w)`	Running or windowed aggregate that keeps every row

Caveat — dropDuplicates is non-deterministic

dropDuplicates(["order_id"]) keeps an arbitrary row from each group. When the question needs "the most recent version", you must use a window with row_number() ordered by the timestamp and filter to rn == 1. If the stem stresses which row survives, the answer is the window, not dropDuplicates.

Delta Lake operations

Delta adds ACID transactions, upserts, and time travel on top of Parquet. The DeltaTable API exposes the operations that plain DataFrame writes cannot.

PySpark · MERGE (upsert), UPDATE, DELETE

from delta.tables import DeltaTable

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

(target.alias("t")
   .merge(updates.alias("s"), "t.customer_id = s.customer_id")
   .whenMatchedUpdateAll()
   .whenNotMatchedInsertAll()
   .execute())

# Point UPDATE / DELETE without rewriting the table by hand
target.update(condition="segment = 'legacy'", set={"segment": "'standard'"})
target.delete("is_test = true")

PySpark · time travel, OPTIMIZE, VACUUM

# Read an older version (audit, recovery, reproducibility)
old = spark.read.format("delta").option("versionAsOf", 3).table("silver_sales")
old = spark.read.format("delta").option("timestampAsOf", "2026-08-01").table("silver_sales")

# Compact small files; optionally cluster co-accessed columns
spark.sql("OPTIMIZE silver_sales")
spark.sql("OPTIMIZE silver_sales ZORDER BY (customer_id, order_date)")

# Remove files older than the retention window (default 7 days)
spark.sql("VACUUM silver_sales RETAIN 168 HOURS")

Caveat — OPTIMIZE, ZORDER and VACUUM do different jobs

OPTIMIZE compacts many small files into fewer large ones (the small-file problem). ZORDER BY co-locates rows by the columns you filter on, so predicate pushdown skips more files. VACUUM deletes old data files no longer referenced — it reclaims storage but breaks time travel beyond the retention window, so never shorten retention below the history you need. These are Spark SQL commands: they run in notebooks / Spark jobs, not in the SQL analytics endpoint.

Structured streaming

Structured streaming treats an unbounded stream as a table that grows. The DataFrame API is nearly identical to batch; the differences are readStream/writeStream, a mandatory checkpoint, a watermark to bound state, and an output mode. See Domain 2 for how it fits the streaming loading pattern.

PySpark · a stateful streaming aggregation

from pyspark.sql.functions import window, avg, count

stream = spark.readStream.format("delta").table("bronze_events")

agg = (stream
    .withWatermark("event_time", "10 minutes")          # tolerate 10 min of lateness
    .groupBy(window("event_time", "5 minutes"), "device_id")
    .agg(avg("temperature").alias("avg_temp"),
         count("*").alias("n")))

(agg.writeStream
    .format("delta")
    .outputMode("append")
    .option("checkpointLocation", "Files/checkpoints/sensor_agg")   # mandatory
    .trigger(processingTime="1 minute")
    .toTable("silver_sensor_agg"))

Output mode	Writes	Notes
`append`	Only new, final rows	Needs a watermark for windowed aggregations; a window emits when it closes
`update`	Only rows that changed this batch	Sink must support updates
`complete`	The whole result table every batch	Aggregations only; state grows unbounded without a watermark

Caveat — checkpoint and watermark are non-negotiable

The checkpoint location stores stream offsets and aggregation state so a restarted job resumes exactly where it stopped — it is mandatory, and two streams must never share one directory. A watermark bounds how long late-arriving state is kept; without it, a stateful aggregation's state grows forever until the job runs out of memory. "Streaming job memory grows until it crashes" → missing withWatermark(). "Restarted stream reprocessed everything" → checkpoint missing or deleted.

Performance & optimization

Partitions are the unit of parallelism, and the shuffle — moving data across the network for wide operations — is the usual bottleneck. Diagnose in the Spark UI, then match the fix to the symptom.

flowchart TD S([Spark job is slow]) --> UI["Diagnose in the Spark UI"] UI -->|"Max task time much greater than median"| SK["Data skew
salt the key · AQE skew join · repartition"] UI -->|"Large shuffle read/write"| SH["Reduce shuffle
broadcast the small side · filter earlier"] UI -->|"Spill to disk"| SP["Memory pressure
more partitions · larger nodes · avoid huge broadcast"] UI -->|"Thousands of tiny files"| SF["OPTIMIZE the table
enable optimize write"] UI -->|"Driver OOM"| DR["Avoid collect() / toPandas()
on large data"]

Read the symptom off the Spark UI, then apply the matching fix. Most "slow Spark" questions map to exactly one of these branches.

Lever	What it does / when to reach for it
Adaptive Query Execution (AQE)	Re-optimises at runtime: coalesces shuffle partitions, converts sort-merge joins to broadcast, and handles skew joins. Keep it enabled — it is the default in Fabric
Broadcast join	Ship a small table to every executor so the large side never shuffles
Filter & prune early	Push `filter` and column `select` before joins; the Catalyst optimizer pushes predicates down for you when you use the DataFrame API
Repartition vs coalesce	`repartition(n)` reshuffles into n balanced partitions (fixes skew, can grow count). `coalesce(n)` only reduces partitions without a full shuffle — use before a write
Salting	Add a random suffix to a hot join/group key to spread it across partitions when one key dominates
Cache / persist	Materialise a DataFrame reused several times; `unpersist()` when done. Caching something used once wastes memory
V-Order + optimize write	Fabric-native: V-Order lays out Parquet for fast reads across all engines; optimize write produces fewer, larger files on write

PySpark · the common performance knobs

# Fabric write optimizations (often set once per session)
spark.conf.set("spark.sql.parquet.vorder.enabled", "true")
spark.conf.set("spark.microsoft.delta.optimizeWrite.enabled", "true")

# AQE is on by default; skew handling is part of it
spark.conf.set("spark.sql.adaptive.enabled", "true")
spark.conf.set("spark.sql.adaptive.skewJoin.enabled", "true")

# Cache a DataFrame reused across several actions, then release it
from pyspark import StorageLevel
hot = clean.persist(StorageLevel.MEMORY_AND_DISK)
hot.count()          # materialise once
# … reuse hot several times …
hot.unpersist()

# Balance skewed partitions before an expensive stage; shrink before a write
balanced = clean.repartition(200, "customer_id")
clean.coalesce(1).write.format("delta").mode("overwrite").saveAsTable("gold_small")

Caveat — collect() and toPandas() pull to the driver

Actions like collect(), toPandas() and a wide countByKey() move all results into driver memory. On a large DataFrame that is a classic driver out-of-memory cause. Keep work distributed; if you must materialise, aggregate or limit first. Executor OOM, by contrast, comes from wide shuffles, skew, or over-caching — different fix (more partitions, AQE, selective persist).

Prefer DataFrames over RDDs, and built-ins over UDFs

The DataFrame / Spark SQL API runs through the Catalyst optimizer and Tungsten engine; raw RDDs bypass both. Built-in functions from pyspark.sql.functions beat Python UDFs, which run in a separate Python process and block many optimisations. If a snippet uses an RDD or a Python UDF where a DataFrame built-in would do, that is usually the thing the question wants you to improve.

Scenario quick reference

Scenario	Deciding constraint	Answer
Read cell instant, write cell slow	Lazy evaluation	Normal — the write is the first action
Overwrite fails after adding a column	Schema change	`overwriteSchema=true` (or `mergeSchema` if additive)
Keep only the newest row per key	Deterministic dedup	`row_number()` over a window, filter `rn==1`
Join slow, one side is a small dimension	Large shuffle	Broadcast join
Upsert changed rows into a Delta table	Insert + update	Delta `MERGE`
Recover yesterday's version of a table	History	Time travel (`versionAsOf` / `timestampAsOf`)
Table has thousands of tiny files	Small-file problem	OPTIMIZE (+ optimize write on ingest)
Filtered queries still scan everything	File skipping	OPTIMIZE … ZORDER BY the filter columns
A few tasks take far longer than the rest	Data skew	Salt the key / AQE skew join
Streaming job's memory grows until it fails	Unbounded state	Add `withWatermark()`
Restarted stream reprocessed everything	Lost offsets	Checkpoint location missing or deleted
Driver runs out of memory	Data pulled local	Avoid `collect()` / `toPandas()` on big data

Verify against the docs

Spark and Delta APIs are stable, but Fabric adds and renames features (V-Order, optimize write, the Native Execution Engine). Confirm details against the Fabric Spark best-practices, Delta optimization & V-Order and the PySpark API reference before relying on them.