Tuning PySpark Shuffle, Memory and Skew: A Measured Approach

Start with evidence, not a larger cluster

A slow join is a question to investigate. Save the input snapshot, executed plan, cluster settings and stage metrics before making a change. Compare typical task duration with the longest tasks; record shuffle volume, garbage collection, failures and output counts. Decide which outcome matters: elapsed time, cost per successful run or a freshness deadline. This guide uses Apache Spark 3.5.6 documentation as an explicit reference baseline. Check the documentation for your deployed runtime before using any configuration.

Separate heap pressure from container pressure

For YARN and Kubernetes, budget executor resources across JVM heap, memory overhead, configured off-heap memory and explicitly configured PySpark memory. When PySpark memory is unset, Python shares the overhead allowance. A larger heap does not automatically resolve a Python-process or container-limit failure. Inspect the actual failure and cluster-manager logs. Configure deployment memory before starting the application; do not assume a runtime setting resizes an existing executor.

# Resource-budget model for YARN / Kubernetes
# executor memory request =
#   spark.executor.memory
# + spark.executor.memoryOverhead
# + spark.memory.offHeap.size (when configured)
# + spark.executor.pyspark.memory (when configured)
# This is not a formula for the memory available to one task.

Use a portable AQE starting point

Adaptive Query Execution can adjust eligible query plans using runtime statistics. The following settings make the intended behavior explicit. The partition count is an illustrative starting value, not a recommended production default; choose it using your measured shuffle size and available parallelism. Stock Spark expects an integer for spark.sql.shuffle.partitions. Do not copy a vendor-specific auto setting into a different runtime. AQE does not eliminate the need to inspect the final executed plan.

# Illustrative Spark SQL session configuration
spark.conf.set("spark.sql.adaptive.enabled", "true")
spark.conf.set("spark.sql.adaptive.coalescePartitions.enabled", "true")
spark.conf.set("spark.sql.adaptive.skewJoin.enabled", "true")
spark.conf.set("spark.sql.shuffle.partitions", "400")
# Execute the query and inspect its final plan and task metrics.

Interpret spill without declaring failure

Spark's UI distinguishes the deserialized memory size of spilled shuffle data from its serialized disk size. Neither is simply the peak executor footprint. Treat spill as a diagnostic alongside task duration, disk pressure and retries; an operation can spill and still finish successfully. Compare the same query and input before and after tuning. Avoid claiming that every spill proves an executor ran out of memory or that zero spill is always the best cost-performance outcome.

A small, correctness-first salting example

The example below assigns each fact row one deterministic bucket and replicates dimension rows across the buckets. It keeps the original key and salt as separate join columns. This is an inner equijoin example: null keys do not match, and duplicate dimension keys retain their multiplicity. It salts every key for clarity; in a production experiment, isolate demonstrated hot keys and assess the cost of replicating the matching dimension subset. It is a teaching example, not a performance benchmark. Run it in a compatible PySpark environment and verify the assertions before adapting it.

from pyspark.sql import SparkSession, functions as F

spark = SparkSession.builder.appName("salting-correctness").getOrCreate()
facts = spark.createDataFrame(
    [(1, "hot", 10), (2, "hot", 20), (3, "cold", 30),
     (4, "missing", 40), (5, None, 50)],
    "event_id long, user_id string, amount long"
)
dimension = spark.createDataFrame(
    [("hot", "A"), ("hot", "B"), ("cold", "C")],
    "user_id string, segment string"
)

# Choose a salt factor experimentally; 4 is for this tiny example only.
SALT_FACTOR = 4
salted_facts = facts.withColumn(
    "_salt", F.pmod(F.xxhash64("event_id"), F.lit(SALT_FACTOR))
)
buckets = spark.range(SALT_FACTOR).select(F.col("id").alias("_salt"))
salted_dimension = dimension.crossJoin(buckets)

columns = ["event_id", "user_id", "amount", "segment"]
actual = salted_facts.join(
    salted_dimension, ["user_id", "_salt"], "inner"
).select(*columns)
expected = facts.join(dimension, ["user_id"], "inner").select(*columns)

# Check both directions, preserving duplicate-row multiplicity.
assert actual.exceptAll(expected).limit(1).count() == 0
assert expected.exceptAll(actual).limit(1).count() == 0
actual.orderBy("event_id", "segment").show()
# Five matched rows: the two hot facts each match A and B;
# the cold fact matches C. Null and missing keys do not match.
spark.stop()

Make the experiment falsifiable

Write the hypothesis before the run: for example, “a hot key creates the long tail in this join.” Then test one intervention against the baseline. Keep the data snapshot, runtime, worker shape and query output equivalent; document cache state and repeat runs. A correctness regression rejects a faster result. Record exceptions and retries rather than discarding inconvenient runs. Calculate infrastructure cost from actual usage and billing assumptions, including failed attempts and extra dimension replication. No production measurements are claimed here.

Decide what to keep

Retain a change only when it improves your chosen outcome without breaking correctness or operational constraints. If salting adds overhead without shortening the critical stage, remove it. If a memory change reduces retries but raises total cost, record that tradeoff explicitly. A useful engineering article ends with evidence someone else can reproduce, not an impressive percentage without a dataset, configuration and method.

What to record

Official documentation

• Apache Spark 3.5.6 — Configuration ↗
• Apache Spark 3.5.6 — SQL performance tuning ↗
• Apache Spark 3.5.6 — Web UI metrics ↗