Performance Tuning

Part V — Running It Well · Module 17 of 18 Prerequisites: all of Parts II–III, plus Caching & Persistence and Joins & Broadcast You will learn: the working engineer's tuning playbook — the highest-leverage settings and habits that turn a slow, resource-starved job into a fast one: shuffle partitions, Adaptive Query Execution, dynamic allocation, coalesce vs. repartition, avoiding driver OOM, the small-file problem, native nulls, and how to read the Spark UI to find the trouble.

---

How to think about tuning

This module gathers the levers scattered across the guide into one checklist. But before the knobs, the mindset: most Spark performance problems come from too much data moving (shuffles), work being repeated (no caching), or parallelism being wrong (partition count). Nearly every rule below is an attack on one of those three.

And the meta-rule that outranks all the others: stay in the structured, optimized world and let Catalyst and Tungsten work (Module 08). The fanciest config tuning can't rescue code that fights the optimizer. So we start there.

Analogy — debugging Spark is detective work. Tuning a distributed job isn't like stepping through code line-by-line in a debugger. It's sleuthing: you follow trails of clues — a stage that ran far longer than its siblings, one task processing ten times the data of the others, a spike in shuffle-write bytes — across the Spark UI to locate the real culprit. Keep this investigative posture; the UI is your crime scene.

---

Rule 0: Prefer DataFrames, built-ins, and native nulls

The foundation, restated as actionable rules because it matters more than any config:

• Use DataFrames/Spark SQL over RDDs (Module 04). RDDs bypass Catalyst and Tungsten; you lose every automatic optimization.
• Prefer built-in functions over UDFs (Module 12). Built-ins stay in Tungsten's binary format; standard Python UDFs pay the row-by-row SerDe tax and block optimization.
• Represent missing values with Spark's native null, never dummy strings like " " or "EMPTY". Spark's engine is optimized to skip nulls early in the plan; custom placeholders force it to carry and process meaningless data through shuffles. Native nulls keep you on the optimizer's fast paths.

If your code violates these, fix them before touching any tuning knob — no amount of config compensates for leaving the optimized world.

---

Rule 1: Tune shuffle partitions

The single most common beginner performance bug (Modules 03, 07): spark.sql.shuffle.partitions defaults to 200, which is wildly wrong for many workloads.

# Small or local dataset: 200 tiny tasks = pure overhead. Match to your cores.
spark.conf.set("spark.sql.shuffle.partitions", 8)

# Huge dataset: raise it so each shuffle partition isn't so big it spills to disk.
spark.conf.set("spark.sql.shuffle.partitions", 400)

• Small/local/streaming workloads: reduce toward the number of executor cores (e.g., 5–16). Otherwise every groupBy/join spawns 200 near-empty tasks drowning in scheduling overhead.
• Very large workloads: increase it so no single shuffle partition grows large enough to spill to disk.
• Aim for a healthy partition-to-core ratio of ~2:1 to 3:1 (Module 03) so every core stays saturated without excessive task overhead.

---

Rule 2: Let Adaptive Query Execution do it for you

Manually guessing the perfect shuffle-partition count is hard because the right number depends on data you haven't seen yet. Adaptive Query Execution (AQE), introduced in Spark 3.0, lets Spark re-optimize the plan at runtime using statistics from completed stages — so it adjusts on the fly instead of committing to your static guess.

spark.conf.set("spark.sql.adaptive.enabled", True)                       # master switch
spark.conf.set("spark.sql.adaptive.coalescePartitions.enabled", True)    # merge tiny shuffle partitions
spark.conf.set("spark.sql.adaptive.skewJoin.enabled", True)              # handle skewed joins

With AQE on, Spark can:

• Coalesce empty/tiny shuffle partitions after a shuffle, fixing an over-high shuffle.partitions automatically.
• Handle data skew in joins by splitting oversized partitions — a lifesaver when one key dwarfs the others.
• Switch join strategies at runtime (e.g., flip to a broadcast join once it knows a side is small).

Enable AQE. It's one of the highest-value flags in modern Spark and removes much of the guesswork from Rule 1.

---

Rule 3: coalesce to shrink, repartition to grow

From Module 03, made precise as a tuning rule:

• coalesce(n) — reduces partition count by merging adjacent partitions on the same node. It's a narrow transformation — no shuffle — so it's cheap. Use it to consolidate after a heavy filter left you with many sparse partitions, or before writing to avoid tiny files.
• repartition(n) — does a full shuffle to create n evenly-balanced partitions. Expensive, but the only way to increase partitions or to rebalance skewed data.

big_df.coalesce(10).write.parquet("/out")   # cheap: fewer output files, no shuffle
skewed_df.repartition(200)                  # expensive but rebalances/increases partitions

Rule of thumb: coalesce to reduce, repartition only to increase or to fix imbalance. Reaching for repartition when coalesce would do buys you an unnecessary shuffle.

---

Rule 4: Don't blow up the driver

The driver has finite memory (Module 02), and certain actions pull data back to it. The classic crash:

# DANGER: collect() pulls the ENTIRE dataset into the driver's memory.
all_rows = huge_df.collect()      # OutOfMemoryError if it doesn't fit → app dies

collect() sends every row to the driver. On a large DataFrame this exceeds the driver's heap and kills the application. Safer patterns:

• Write to a sink instead of collecting: df.write.parquet(...).
• Peek at a few rows with df.show() or df.take(20) — bounded, safe.
• Stream results serially with df.toLocalIterator() when you truly must process all rows in the driver, fetching partition-by-partition instead of all at once.

Same caution for broadcasts (Module 16): only broadcast genuinely small tables, since a broadcast also stages through the driver.

---

Rule 5: Avoid the small-file problem

From Module 03: thousands of tiny files bloat file-system metadata and force Spark to launch a task per file. Control output file size on write:

# Cap records per file so you write fewer, healthy-sized files (tens of MB+).
df.write.option("maxRecordsPerFile", 50000).parquet("/out")

# Or coalesce down before writing to reduce the number of output files.
df.coalesce(16).write.parquet("/out")

Target files of at least tens of megabytes; prefer fewer, fatter files over a swarm of tiny ones — on both the read and write sides.

---

Rule 6: Cache deliberately (not reflexively)

From Module 15, as a tuning discipline: cache a DataFrame only when it's reused across multiple actions/iterations, is expensive to build, and fits in memory — and remember to materialize it with an action, and unpersist() when done. Over-caching steals memory from computation and can slow you down. For large cached data, prefer MEMORY_ONLY_SER to shrink the footprint and reduce GC pressure.

---

Rule 7: Right-size joins

From Module 16, as tuning habits:

• Broadcast the small side of a huge-⋈-small join (broadcast() hint; tune spark.sql.autoBroadcastJoinThreshold, default 10 MB) to skip the big shuffle.
• Bucket two large tables joined repeatedly on a stable key to eliminate the join shuffle permanently.
• Filter before joining so less data enters the shuffle.

---

Rule 8: Advanced — shuffle I/O, dynamic allocation, and GC

These are lower-frequency knobs, but worth knowing when the basics aren't enough.

Shuffle I/O buffers — shuffles are I/O-intensive; larger buffers mean less disk spilling during the map/write phase:

spark.shuffle.file.buffer            1m      # default 32k — buffer more map output in memory
spark.io.compression.lz4.blockSize   512k    # larger compression blocks → smaller shuffle files

Dynamic allocation — instead of a fixed cluster size, let Spark scale executors up and down with the workload (ideal for shared/multitenant clusters):

spark.dynamicAllocation.enabled                  true
spark.dynamicAllocation.minExecutors             2
spark.dynamicAllocation.maxExecutors             20
spark.dynamicAllocation.executorIdleTimeout      2min    # release idle executors

Garbage collection — RDDs and Datasets create many short-lived JVM objects, pressuring the garbage collector (DataFrames largely sidestep this via Tungsten — another reason to prefer them). On large heaps, the G1GC collector (-XX:+UseG1GC) generally handles Spark's allocation patterns better, and you can collect GC metrics with -verbose:gc -XX:+PrintGCDetails to diagnose pauses.

---

Reading the Spark UI: where to look

Tuning without measurement is guessing. The Spark UI (typically at port 4040 during a running app) is where you find the actual bottleneck. A quick tour of what to check:

• Jobs / Stages tabs — find the stage that took disproportionately long. Look for an Exchange (shuffle) with large shuffle-read/write bytes — that's usually where the time goes.
• Task-level view within a stage — compare task durations. If one task took far longer than its peers, you have data skew (one partition got most of the data) — a case for AQE skew handling (Rule 2) or a better key.
• Storage tab — confirm your cached DataFrames are actually cached and how much memory they occupy (did you materialize them? Module 15).
• SQL tab — inspect the query plan visually, count the Exchange nodes (shuffles), and confirm broadcast joins and pushed filters happened as intended.

The diagnostic loop: measure in the UI → find the longest stage / most skewed task / biggest shuffle → apply the matching rule above → measure again. Tune what the data tells you is slow, not what you guess is slow.

---

Key takeaways

• Most Spark slowness is excess shuffling, repeated work, or wrong parallelism — and the meta-rule is stay structured; let Catalyst/Tungsten work.
• Rule 0: prefer DataFrames/SQL over RDDs, built-ins over UDFs, and native nulls over dummy values.
• Rule 1: tune spark.sql.shuffle.partitions (default 200 is often wrong) toward your core count for small data, higher for huge data; aim for a 2:1–3:1 partition-to-core ratio.
• Rule 2: enable AQE (spark.sql.adaptive.enabled) to coalesce small partitions, handle skew, and switch join strategies at runtime.
• Rule 3: coalesce to shrink (no shuffle), repartition to grow/rebalance (full shuffle).
• Rule 4: avoid collect() on big data (driver OOM) — write to a sink, show()/take(), or toLocalIterator().
• Rule 5: dodge the small-file problem (maxRecordsPerFile, coalesce before write).
• Rule 6–7: cache deliberately (reused + expensive + fits, materialized), and right-size joins (broadcast small sides, bucket repeated large joins, filter early).
• Rule 8: advanced knobs — shuffle I/O buffers, dynamic allocation, and G1GC.
• Measure in the Spark UI — longest stage, skewed tasks, biggest Exchange — and tune what's actually slow.

---

Check your understanding

1. A small local job with a few groupBys is oddly slow, spawning hundreds of tiny tasks. Which single setting is the likely cause, and what would AQE do about it?
2. Why is collect() dangerous on a large DataFrame, and name two safer alternatives.
3. You open the Spark UI and see one task in a stage took 10× longer than all the others. What is this called, and which Rule-2 feature helps?

---

Up next

Glossary & Cheatsheet — Glossary & Cheat Sheet. The final module: a quick-reference glossary of every term we've defined and a one-page PySpark cheat sheet, so you can look things up fast long after your first read-through.