Spark Performance: DAG & Query Profile

Understanding Spark execution

The four performance villains

1. Data skew

Problem: One partition has far more data than others. One task takes 10 minutes while 199 tasks finish in 10 seconds.

Diagnosis (Spark UI):

• Stages tab → look for one task with much higher duration/data processed
• Task metrics show uneven distribution

Fixes:

# Salting: add a random prefix to the skewed key
from pyspark.sql.functions import concat, lit, floor, rand, col

df_salted = df.withColumn("salted_key",
    concat(col("customer_id"), lit("_"), floor(rand() * 10)))

# Or use Adaptive Query Execution (AQE) — enabled by default
spark.conf.set("spark.sql.adaptive.enabled", "true")
spark.conf.set("spark.sql.adaptive.skewJoin.enabled", "true")

2. Spilling

Problem: Data doesn’t fit in memory and overflows to disk. Disk I/O is 10-100x slower than memory.

Diagnosis (Spark UI):

• Stages tab → “Spill (Memory)” and “Spill (Disk)” columns
• Any non-zero spill indicates memory pressure

Fixes:

• Increase executor memory
• Increase number of partitions (smaller partitions per executor)
• Reduce data being processed (push filters earlier)

3. Shuffle

Problem: Data must be redistributed across executors for operations like JOIN, GROUP BY, and DISTINCT. Shuffle = network transfer + disk I/O.

Diagnosis (Spark UI):

• Stages tab → “Shuffle Read” and “Shuffle Write” columns
• Large shuffle = potential bottleneck

Fixes:

• Broadcast join: for small tables, broadcast to all executors (avoids shuffle)

  from pyspark.sql.functions import broadcast
  result = large_df.join(broadcast(small_df), "key")

• Partition pruning: filter data BEFORE the join
• Reduce partition count: spark.sql.shuffle.partitions (default 200)

4. Caching

Problem: Recomputing the same DataFrame multiple times wastes resources.

Diagnosis: Same stages appearing repeatedly in the DAG.

When to cache:

# Cache a DataFrame that's reused multiple times
expensive_df = spark.table("big_table").filter("year = 2026").cache()

# Use it multiple times
summary = expensive_df.groupBy("region").count()
details = expensive_df.filter("amount > 1000")

# Unpersist when done
expensive_df.unpersist()

When NOT to cache: DataFrames used only once, or when memory is limited.

Diagnostic tools

Tool	What It Shows	Best For
DAG Visualisation	Visual flow of stages and their relationships	Understanding execution plan structure
Spark UI — Stages	Per-stage metrics: duration, shuffle, spill, task count	Finding slow stages and skew
Spark UI — SQL	Logical and physical query plans	Understanding join strategies and filter pushdown
Query Profile	SQL warehouse execution plan with operator stats	SQL-specific query optimization

💡 Exam tip: AQE solves many issues automatically

Adaptive Query Execution (AQE) is enabled by default in recent Databricks runtimes. It automatically:

• Coalesces small partitions after shuffle (reduces partition overhead)
• Converts sort-merge joins to broadcast joins when one side is small
• Handles skew joins by splitting skewed partitions

If the exam asks about the “easiest” or “simplest” fix for skew or shuffle issues, AQE is often the answer — it’s automatic and requires no code changes.

Question

What is data skew and how do you detect it?

Click or press Enter to reveal answer

Answer

Data skew = one partition has far more data than others, causing one task to take much longer. Detect in Spark UI → Stages → look for one task with disproportionately high duration or data processed.

Click to flip back

Question

What is spilling and what causes it?

Click or press Enter to reveal answer

Answer

Spilling occurs when data doesn't fit in executor memory and overflows to disk. Disk I/O is 10-100x slower. Fix: increase executor memory, increase partitions, or push filters earlier to reduce data volume.

Click to flip back

Question

When should you use a broadcast join?

Click or press Enter to reveal answer

Answer

When one table is small enough to fit in executor memory (typically under 10 MB (Databricks default is 30 MB with AQE)). Broadcasting avoids shuffle — the small table is sent to all executors. Use broadcast() hint in PySpark.

Click to flip back

Knowledge check

Knowledge Check

Tomás notices that a Spark join in NovaPay's pipeline takes 45 minutes. In the Spark UI, he sees one task processed 90% of the data while 199 tasks each processed tiny amounts. What is the likely issue?

AInsufficient cluster memoryBCaching not enabledCToo many shuffle partitionsDData skew on the join key

Check Answer

---

Next up: Optimizing Delta Tables & Azure Monitor — OPTIMIZE, VACUUM, log streaming, and Azure Monitor integration.