04.11 PySpark Performance Engineering

PySpark performance is fundamentally about data reduction, parallelism, data movement, computation and storage. The goal is to identify expensive work, remove it or make it smaller, and validate the result.

1. Performance mental model

SOURCE
  ↓
Read less data
  ↓
Prune columns / partitions
  ↓
Transform efficiently
  ↓
Avoid unnecessary shuffle
  ↓
Choose appropriate join strategy
  ↓
Balance partitions
  ↓
Use efficient compute
  ↓
Measure in Spark UI

Every optimization should answer: What expensive work did I remove or make smaller?

2. Reduce the data first

High-value optimizations usually reduce rows or columns before expensive joins, aggregations and windows.

filtered = orders.filter(F.col("order_date") >= "2026-01-01")
narrow = filtered.select("customer_id", "order_date", "amount")

3. File and scan performance

File-based sources are split into input partitions. Current Spark exposes controls including spark.sql.files.maxPartitionBytes, spark.sql.files.openCostInBytes, and file partition-count guidance. citeturn939614search0

Small-file problem

10,000 tiny files
      ↓
many file opens
      ↓
many small tasks
      ↓
scheduler overhead
      ↓
poor throughput

The target is not one universal file size. The target is a layout with healthy parallelism and efficient reads.

4. Partitions: the unit of parallel work

One Spark task normally processes one partition. Too few partitions can limit parallelism; too many tiny partitions can create scheduling overhead.

df.rdd.getNumPartitions()

Measure task duration and partition sizes instead of choosing a universal partition count.

5. repartition vs coalesce

df.repartition(200)
df.repartition(200, "customer_id")
df.coalesce(20)

Operation	Use when
repartition	You need redistribution, a new key distribution or controlled parallelism
coalesce	You mainly need fewer partitions without a full redistribution

Every unnecessary repartition is a potential shuffle and therefore a performance/cost decision.

6. Shuffle engineering

Shuffle redistributes records between executors. Common causes include joins, groupBy, distinct, orderBy, windows and repartition.

Executors → network → new partitions

Shuffle consumes network, CPU, memory, local disk and scheduling time. Do not try to eliminate necessary shuffles; make them smaller and healthier.

7. Broadcast joins

result = fact.join(
    F.broadcast(customer_dim),
    "customer_id",
    "left"
 )

Broadcast can avoid shuffling the large side when the build side is safely small. Spark's current automatic broadcast threshold is 10 MiB by default, while runtime statistics and AQE can affect the final decision. citeturn939614search0

8. Sort-merge joins

Left → shuffle by key → sort ─┐
                              ├→ Sort-Merge Join
Right → shuffle by key → sort ─┘

Large distributed equi-joins commonly use sort-merge execution. This is more expensive than a safe broadcast join but scales for large relations. Do not force broadcast simply because it sounds faster.

9. AQE — Adaptive Query Execution

AQE uses runtime statistics to adapt execution. Modern Spark enables it by default and can coalesce post-shuffle partitions, convert certain join strategies and handle skewed joins. citeturn939614search0

Initial plan
    ↓
Runtime statistics
    ↓
Adaptive re-optimization
    ↓
Better physical execution

Current Spark 4.3 also includes additional adaptive join conversion improvements. citeturn939614search1

10. Shuffle partition sizing

spark.sql.shuffle.partitions controls shuffle partitioning for joins and aggregations; Spark's current default is 200. citeturn939614search0

200 is not a target. A 2 TB shuffle and a 2 GB shuffle should not be expected to behave identically with the same static number. AQE can coalesce post-shuffle partitions using runtime statistics. citeturn939614search0

11. Data skew

Skew occurs when a small number of keys own a disproportionate amount of data.

A → 2K rows
B → 2K rows
C → 3K rows
BIG_CUSTOMER → 900M rows

Classic symptom: most tasks finish quickly while one or a few tasks run dramatically longer.

Inspect task duration, shuffle read/write, input size and spill in Spark UI. AQE has adaptive skew-join handling that can split skewed shuffle partitions. citeturn939614search0

12. Skew mitigation

Prefer, in order:

1. Reduce data before the skewed operation.
2. Use AQE skew handling.
3. Pre-aggregate when semantics allow.
4. Use salting for severe, known hot keys.

Salting adds application complexity and should be a deliberate last-resort technique.

13. Predicate and dynamic partition pruning

A filter can reduce data read when the physical storage layout supports pruning. Dynamic partition pruning can generate predicates from join relationships and is enabled by default in current Spark. citeturn939614search2

Do not assume a filter is physically cheap. Inspect the scan and execution plan to verify pruning.

14. Column pruning

Wide rows increase I/O, memory, serialization and network cost.

sales = spark.table("sales").select(
    "customer_id",
    "product_id",
    "order_date",
    "amount"
 )

Current Spark optimizer documentation also describes column pruning and expression optimizations. citeturn939614search2

15. Aggregation performance

Aggregation generally follows:

Input partitions
      ↓
Partial aggregation
      ↓
Shuffle by grouping key
      ↓
Final aggregation

Reduce rows before aggregation when semantics allow. For example, aggregate a large fact table by customer before joining a customer dimension if the downstream result only needs customer-level measures.

16. Window performance

Windows can require partitioning and sorting.

w = Window.partitionBy("customer_id").orderBy("order_date")
df.withColumn("previous_amount", F.lag("amount").over(w))

Ask whether the partition key is appropriate, whether ordering is required, whether input can be reduced first, and whether multiple expressions can share the same window specification.

17. Cache and persist

Cache is useful when an expensive result is reused.

prepared = expensive_transformation(source).cache()
prepared.count()
a = prepared.groupBy("region").count()
b = prepared.groupBy("product").count()
prepared.unpersist()

Measure recomputation avoided, cache size, executor memory pressure, GC and total runtime. Spark's Storage UI exposes persisted relation sizes and partitions. citeturn939614search0

Do not cache every intermediate DataFrame. Cache should answer: What expensive computation am I avoiding?

18. Photon and compute-aware optimization

In Databricks, observed performance depends on data layout, query plan, Spark/AQE, execution engine and compute sizing.

Data layout + query plan + Spark/AQE + execution engine + compute sizing
                         ↓
                  observed performance

Do not use a larger cluster to hide excessive shuffle or poor data layout. Conversely, do not rewrite a good plan endlessly when the actual bottleneck is insufficient compute.

19. Spark UI performance workflow

Slow job
  ↓
Slow stage
  ↓
Slow task pattern
  ↓
Physical operator
  ↓
Data movement / computation cause
  ↓
Targeted change
  ↓
Benchmark

Inspect Jobs/SQL, Stages, Tasks, SQL/DAG, Storage and Executors. Look for Exchange, Sort, Join, Aggregate, scan behavior, skew, spill, memory pressure and uneven task duration.

20. Symptoms → likely causes

Symptom	First hypothesis
One task much slower	Skew
Many tiny tasks	Small files / excessive partitioning
Huge shuffle	Join / groupBy / repartition
Executor OOM	Broadcast / cache / oversized partition
High spill	Memory pressure / large shuffle
Low CPU but slow	I/O / shuffle / scheduling / skew
Driver OOM	collect / toPandas / large driver result
Many output files	Excessive output partitions

These are hypotheses, not conclusions. Verify them in the UI.

21. End-to-end performance example

Suppose a 4 TB sales pipeline takes 55 minutes:

Scan 4 TB → select * → large join → repartition → groupBy → write

Optimization sequence:

1. Filter the required date range.
2. Project only required columns.
3. Validate join cardinality.
4. Check whether the dimension is safely broadcastable after reduction.
5. Remove unnecessary repartitioning.
6. Inspect shuffle distribution.
7. Investigate skew.
8. Benchmark.

The objective is reduce the expensive work required to produce the same answer.

22. Enterprise performance checklist

• Correct grain established
• Input filters validated
• Required columns only
• Small-file condition assessed
• Partition count measured
• Join cardinality validated
• Broadcast suitability checked
• Shuffle volume inspected
• Skew checked
• AQE behavior verified
• Cache justified
• Driver collection controlled
• Spark UI reviewed
• Runtime benchmark captured
• Cost impact considered

23. Hands-on Lab — Find the bottleneck

Create a representative fact/dimension workload with a deliberately skewed key and unused columns. Build a slow baseline and independently test earlier filtering, narrower projection, safe broadcast, removal of unnecessary repartition, shuffle partition tuning, AQE behavior, skew mitigation, and cache versus recomputation.

Experiment	Runtime	Shuffle	Spill	Max task time	Cost observation
Baseline
Filter
Projection
Broadcast
AQE
Skew fix

24. Architect decision framework

Problem	First investigation	Typical direction
Too much input	Scan metrics	Filtering / pruning / better layout
Too many columns	Projection	Column pruning
Huge join shuffle	Join plan	Broadcast / reduce input / join strategy
One huge task	Task distribution	Skew handling
Many tiny tasks	File/task layout	Compact files / healthier partitions
Excessive spill	Memory + shuffle	Reduce data / partition better / compute sizing
Driver OOM	Driver operations	Bound results / keep distributed
Repeated computation	DAG	Cache if reused
Long lineage	Plan	Checkpoint when justified

25. Interview questions

What is the first thing you do when a PySpark job is slow? Inspect the physical plan and Spark UI before increasing cluster size.

How do you identify skew? Look for highly uneven task durations and shuffle/input sizes, then identify the responsible key or operation.

Why can more executors fail to solve a slow job? Skew, excessive shuffle, I/O, poor data layout or driver-side work can remain the bottleneck.

When would you broadcast a table? When the build side is safely small enough for executor memory and the join supports broadcast execution.

What does AQE solve? It uses runtime statistics for adaptive partition coalescing, certain join conversions and skew handling. citeturn939614search0

26. Production cheat sheet

PYSPARK PERFORMANCE
│
├── Reduce rows
├── Reduce columns
├── Read efficiently
├── Control file sizes
├── Balance partitions
├── Avoid unnecessary shuffle
├── Validate join cardinality
├── Broadcast only when safe
├── Understand skew
├── Use AQE
├── Cache only for measured reuse
├── Keep large results off the driver
├── Inspect Spark UI
├── Benchmark every material change
└── Optimize cost + runtime

Official References

• Apache Spark SQL Performance Tuning
• Apache Spark Configuration
• Apache Spark Tuning Guide
• Apache Spark 4.3.0 Release Notes

Key Takeaway

Performance engineering is the discipline of turning slow workload → evidence → physical plan + Spark UI → root cause → targeted change → benchmark → production validation.

A strong Databricks engineer does not memorize one partition count or cluster size. They know how to measure, explain and remove expensive work.

Next: 04.12 PySpark Project — an end-to-end production-grade implementation.