Spark Performance Optimization

Data Engineering & Modern Data Platforms · Lesson 31 · Big Data Processing and Apache Spark

Lesson Overview

After the engagement join in Spark Joins and Data Shuffling stopped broadcasting the wrong side, the TechLearningPro morning job still missed 06:00 UTC. The formatted explain showed a sort-merge with two hundred partitions. Most of those partitions held a few megabytes. Three of them spilled. The write then produced about eight thousand Parquet files for one event_date, because the two hundred shuffle partitions were split again by a partitionBy("course_id") on a high-cardinality column the fact table did not need. Someone had also called cache() on the raw events, used that cache once, and left it in memory while the sort asked for execution memory. Adaptive query execution was off. The runbook that disabled it was written for Spark 2.4, copied forward, and never re-read. The job was no longer wrong. It was shaped like a cluster from 2019.

Performance work on Spark is a sequence of measurements against a plan, not a bag of flags. Adaptive query execution rewrites three things at runtime from shuffle statistics: it coalesces small shuffle partitions, it can switch a sort-merge to a broadcast hash join when a side turns out small, and it splits skewed sort-merge partitions. Caching stores a DataFrame so a second action does not recompute it, and the storage level decides whether that copy can spill from memory to disk. Spill during a sort or a hash aggregate means execution memory was not enough for the operator, so Spark wrote sorted runs to local disk and merged them. File size at the sink is a layout decision: partitions, bytes per file, and whether partitionBy multiplies the task count. An explain plan is how you see the choices before you add executors.

This lesson uses Apache Spark 3.5 configuration names as documented in the SQL performance tuning guide. Defaults below are the 3.5 defaults. Read them back with spark.conf.get before you change them. Adding machines is a real lever when the plan is already even and CPU-bound. It is an expensive way to hide two hundred tiny tasks and a cache that serves nothing.

Learning Objectives

By the end of this lesson you will be able to:

1. Describe the three adaptive query execution rewrites that use runtime shuffle statistics: coalescing, join strategy switch, and skew split.
2. Distinguish spark.sql.shuffle.partitions from spark.sql.adaptive.advisoryPartitionSizeInBytes, and say when coalescing shrinks the task count.
3. Choose a DataFrame storage level, and explain why cache() on a DataFrame is MEMORY_AND_DISK while cache() on an RDD is MEMORY_ONLY.
4. Decide when a reused intermediate should be cached and when a one-shot DataFrame should not.
5. Read spill (memory) and spill (disk) on a sort or aggregate as a memory-layout signal.
6. Use explain(mode="formatted") and name the modes simple, extended, codegen, cost, and formatted.
7. Size a Parquet write so one event_date is not thousands of small files, without using partitionBy on a high-cardinality attribute.
8. State why Kryo does not change a DataFrame shuffle, and why a Python UDF blocks predicate pushdown.
9. Separate "add executors" from "fix the plan" using one before-and-after metric.
10. Leave a job whose expensive stages match the join design from lesson 30.

Prerequisites

• Spark Joins and Data Shuffling: broadcast versus sort-merge, the hot course_id, and explain.
• Apache Spark Architecture: executors, cores, and the driver. Memory fractions in this lesson are fractions of executor heap.
• Spark RDDs and Execution Model: persistence levels began on RDDs. DataFrames reuse the names and change the default.
• Data Formats and Serialization: Parquet row groups and the small-file problem. This lesson is the writer that creates or avoids those files.
• A Spark UI you can open on a local run or a cluster. Screenshots of task metrics are part of the exercise. No proprietary profiler is required.

Detailed Core Concepts

What you are allowed to optimize

A faster plan that changes the grain is a different product. Caching does not make a duplicate enrollment row correct. Coalescing files does not repair an inner join that dropped unknown lessons. The optimizations in this lesson preserve the row set of the join you already designed. If a change alters counts, stop and treat it as a logic bug. Runtime is the second acceptance test.

The cost model that matters on this job has five lines: bytes read from Parquet after column pruning, bytes written and read in shuffles, spill bytes, task time lost to skew or to tiny tasks, and bytes written at the sink including the file count a later reader will open. CPU shows up inside task time. Network shows up inside shuffle read time. Guessing which line dominates from the SQL text is how the 2.4 runbook survived.

Adaptive query execution

Adaptive query execution (spark.sql.adaptive.enabled, default true since Spark 3.2) replans a query between stages using sizes from completed shuffles. A static explain, captured before the job runs, shows the plan the optimizer starts with. The Spark UI's SQL tab shows the final plan, including adaptive changes. Compare those two before you declare that a setting did nothing.

Coalescing shuffle partitions. spark.sql.adaptive.coalescePartitions.enabled (default true) merges adjacent small shuffle partitions toward spark.sql.adaptive.advisoryPartitionSizeInBytes. That advisory value follows spark.sql.adaptive.shuffle.targetPostShuffleInputSize, whose usual default is 64 MB. The initial count is still spark.sql.shuffle.partitions (default 200) unless spark.sql.adaptive.coalescePartitions.initialPartitionNum is set. Coalescing reduces tasks when the shuffle is smaller than 200 times the target. It does not split a partition that is already large. spark.sql.adaptive.coalescePartitions.minPartitionSize (default 1 MB) stops the target from collapsing to tiny files of work. spark.sql.adaptive.coalescePartitions.parallelismFirst (default true) is the setting people miss: when true, Spark may choose a target smaller than 64 MB so that a small cluster does not lose parallelism and look slower than Spark 2.x. The tuning guide recommends setting parallelismFirst to false when you want the advisory size respected. On a production engagement job that already has enough cores, false is the setting that makes "64 MB partitions" a true sentence. On a laptop, true avoids a single-thread surprise. Label which one you ran.

Join strategy switch. If a sort-merge's build side is at or under spark.sql.adaptive.autoBroadcastJoinThreshold (default equal to spark.sql.autoBroadcastJoinThreshold, 10 MB) after the shuffle is measured, Spark can replace the second shuffle with a broadcast hash join and a local shuffle read (spark.sql.adaptive.localShuffleReader.enabled, default true). The lesson manifest should not depend on this rescue if you already hinted a broadcast. The rescue matters for a filtered side whose file-size estimate was pessimistic. It will not broadcast a side that is large in memory and small on disk if the measured shuffle bytes are over the threshold. Trust the measured bytes.

Skew join. Covered in lesson 30. Both spark.sql.adaptive.skewJoin.enabled and the two thresholds (factor 5 and 256 MB) apply only when adaptive execution is on. Turning adaptive execution off to "stabilize the plan" also turns skew split off. That was the Tuesday runbook.

spark.sql.adaptive.optimizeSkewsInRebalancePartitions.enabled (default true) splits skewed partitions produced by a rebalance, which includes a repartition written to distribute a sink. It is not a substitute for a skewed aggregation key.

Adaptive execution has limits you should say out loud. It reacts after a stage produces statistics, so the first shuffle still uses the initial partition count. It does not rewrite a Python UDF into native expressions. It does not change a bad partitionBy at the sink. It does not cache anything. A job with adaptive execution on can still be slow for ordinary reasons: it read too many columns, it shuffled a wide row, it spilled, or it wrote a million files.

Memory: execution, storage, and spill

Executor heap is split so that Spark's unified memory region is spark.memory.fraction of (heap minus a reserved portion, 300 MB by default). The default fraction is 0.6. Inside that region, spark.memory.storageFraction (default 0.5) is the share that storage can hold without eviction. Execution can borrow from storage and evict cached blocks when it needs memory for a sort, an aggregate, or a join. Storage cannot always borrow back. The practical reading: a large cache and a large sort compete, and the sort wins by evicting the cache. If the cache was the only copy you needed for a second action, the second action recomputes. You paid for the cache and then lost it at the moment of pressure.

When an operator cannot keep its working set in execution memory, Spark spills. The UI reports spill (memory) and spill (disk) on the task. Sort-merge and hash aggregates spill sorted runs and merge them. Spill is correctness-preserving and slow. A task with spill disk in the tens of gigabytes and a key you already know is hot is the skew problem again. A stage where every task spills is a memory-per-task problem: too few partitions, rows that are wide because you carried unused strings through the shuffle, or an aggregate that holds a huge map per key. Raising executor memory helps the second case after the row is narrow and the partitions are sized. It does not fix a single key.

spark.memory.offHeap.enabled and spark.memory.offHeap.size move Tungsten pages off the JVM heap. Use them when garbage collection time in the executor logs is the dominant task time and the on-heap picture is already understood. They are not the first response to spill caused by skew.

Caching

DataFrame.cache() is persist at StorageLevel.MEMORY_AND_DISK. RDD.cache() is persist at StorageLevel.MEMORY_ONLY. The names match. The defaults do not. A cached DataFrame that does not fit is written to executor local disk and read back. A cached RDD that does not fit is recomputed when the block is evicted, unless you chose a disk level yourself.

Storage levels you will actually choose:

DataFrame cache uses Spark's columnar cached form, which is cheaper to scan for later SQL than a row of Python objects. It is still executor memory. Call unpersist() when the second action is done, especially in a long-lived SparkSession such as a notebook or a streaming driver that also runs batch lookups. A session that caches every intermediate will evict the one you needed.

Cache when two or more actions share a lineage that includes a shuffle or a large scan. Do not cache the raw scan if the next line filters it to one event_date and one project list: cache the filtered projection if it is reused. Do not cache "because the UI has a storage tab." The storage tab should list one or two names you can justify.

spark.catalog.cacheTable is the same idea for a registered table. The invalidation rule is the same: unpersist when the job is finished, or the next job on a shared session inherits the memory.

Explain plans

DataFrame.explain accepts a mode:

Whole-stage codegen (spark.sql.codegen.wholeStage, default true) fuses operators into generated Java so rows do not pay a virtual call per operator. A Python UDF is a barrier: rows are converted, shipped to a Python process, and codegen stops around that call. spark.sql.execution.arrow.pyspark.enabled speeds the conversion for Pandas UDFs and some Arrow paths. It does not make the UDF a Catalyst expression. Prefer pyspark.sql.functions for filters, joins, and aggregates. Keep a Python UDF for a rule that has no SQL equivalent, and place it after the scan has already dropped columns and rows.

Predicate pushdown and column pruning are why the Parquet scan in a good plan lists a PartitionFilters or a pushed filter and a narrow ReadSchema. A filter written as a Python UDF on the raw DataFrame reads every column, then throws rows away in the interpreter. Push the date partition filter and the column list first, in Spark expressions, and the scan's bytes drop before any shuffle.

cost mode is only as good as the statistics. File size is a statistic. Row count after a filter often is not, until adaptive execution measures a shuffle. Do not quote a cost number in a review as if it were elapsed time.

Partitions at the source and at the sink

On read, spark.sql.files.maxPartitionBytes (default 128 MB) splits large files into scan partitions, and spark.sql.files.openCostInBytes (default 4 MB) inflates the cost of opening a file so that a directory of tiny files is combined into fewer tasks. Those settings fix a small-file read. They do not merge files on disk. A directory of eight thousand files still makes the driver list eight thousand files. The repair is a rewrite that produces fewer files.

On write, each task writes at least one file per partitionBy value it holds. Two hundred tasks and a partitionBy("course_id") with hundreds of courses produce thousands of files, many of them small, and the hot course produces two hundred files of its own that a later reader must open. Partition the sink by event_date, a low-cardinality column the queries filter on. Inside that directory, control the file count with repartition or coalesce to a target near 128 MB per file, or set spark.sql.files.maxRecordsPerFile when you need a cap rather than an exact count. coalesce avoids a full shuffle and can pile data into fewer tasks unevenly. repartition shuffles and spreads. For a skewed frame, a round-robin repartition(n) before the write evens the files and does not need to preserve the join key, because the join is finished. Adaptive skew optimization on rebalance can split a remaining hot output partition.

The target this course uses for curated Parquet is about 128 MB to 256 MB per file after compression, one partition column event_date, and no partition column for learner_key or lesson_id. That range is a default you confirm with a file listing after the write, not a property Spark guarantees.

Serialization, and the Kryo myth

spark.serializer defaults to the Java serializer. Setting org.apache.spark.serializer.KryoSerializer helps RDD shuffles of Scala or Java objects and shuffles of cached RDDs when you also register classes. DataFrame and SQL shuffles already use Spark's unsafe binary row format, compressed with the codec in spark.io.compression.codec (lz4 by default). Changing the serializer does not shrink that shuffle. Measure shuffle write bytes before and after if someone insists. If the bytes do not move, revert the flag. Registering every class "just in case" adds a runbook line nobody can explain.

Architecture and Design Diagrams

Adaptive execution sits between stage results and the next physical plan. It does not sit in front of the source scan.

mermaidCopy
flowchart TD
    scan[Parquet scan with partition filter and column prune]
    exch1[Shuffle exchange initial partition count]
    stats[Runtime partition sizes]
    aqe{Adaptive rewrites}
    coal[Coalesce small partitions]
    bhj[Switch to broadcast hash if a side is under the threshold]
    skew[Split skewed sort-merge partitions]
    sink[Write event_date directory with sized files]

    scan --> exch1 --> stats --> aqe
    aqe --> coal
    aqe --> bhj
    aqe --> skew
    coal --> sink
    bhj --> sink
    skew --> sink

Memory inside one executor, for the same job, is a competition:

mermaidCopy
flowchart LR
    heap[Executor heap]
    reserved[Reserved 300 MB]
    user[User and overhead outside the fraction]
    unified[Unified region spark.memory.fraction]
    storage[Storage cache]
    execution[Execution sort hash join]
    spill[Local spill files]

    heap --> reserved
    heap --> user
    heap --> unified
    unified --> storage
    unified --> execution
    execution -->|working set does not fit| spill
    execution -->|pressure| storage

When execution needs memory, it evicts storage down to the protected fraction. Spill files are local to the executor. They are not a second HDFS copy of the dataset, and they disappear when the task ends. They are a reason the task was slow.

Technical Explanation

Reading a formatted plan for this job

A useful formatted plan for the morning job contains, in order: a scan of events with event_date as a partition filter and a ReadSchema that lists the fact columns; a BroadcastExchange and BroadcastHashJoin for the manifest; a shuffle whose keys are the enrollment join keys; a SortMergeJoin or an adaptive replacement; a project that does not include _salt; a write partitioned by event_date. Adaptive changes appear in the UI plan as coalesced partition counts or as a join node that differs from the pre-run explain. Attach both to the change if they differ, and write one sentence about which rewrite fired.

If the scan's ReadSchema lists every raw column, including a payload string you never reference, pruning did not happen. A select of the needed columns before the join fixes it. If the scan has no partition filter, the job read every ingest_date. Filter event_date or ingest_date with a column expression before any UDF.

WholeStageCodegen nodes wrapping the scan and the project mean the fused path is on. A BatchEvalPython or similar Python node in the middle means a UDF barrier. Move that UDF or replace it.

A concrete memory budget

Suppose an executor has 16 GB of heap and four cores, and spark.memory.fraction is 0.6. Roughly 0.6 times (16 GB minus 300 MB) is available to unified memory, and execution and storage share it. Four concurrent tasks divide the execution side. A sort-merge partition of 64 MB compressed can expand several times in memory while it sorts. Two hundred such partitions running four at a time is a normal shape. One partition of several gigabytes on one core is the hot key, and giving the executor 32 GB only delays the spill. This arithmetic is why the lesson asks for partition bytes and spill before it asks for a larger instance type.

spark.executor.memoryOverhead covers off-heap native memory, Python worker memory, and similar costs outside the JVM heap. PySpark jobs that use Python UDFs need headroom here. A container killed by the node manager at the memory limit, while the JVM heap looks calm, is often overhead rather than spark.executor.memory. Raise overhead when the kill reason is the container, and remove the UDF when the Python worker is the reason the overhead grew.

Shuffle bytes and wide rows

Shuffle write bytes scale with rows times width. Carrying lesson_title, a description, and a JSON payload through the enrollment shuffle multiplies the exchange by columns the merge does not need. Project to keys plus measures before the shuffle, and join attributes after the aggregate if the output needs them. Broadcast attributes are cheap only while the broadcast stays small. A title on a forty-thousand-row manifest is fine. A title repeated onto sixty million shuffled rows is not.

The small-file write, step by step

1. Finish the join and the aggregate so the grain is the fact grain, one row per (learner_key, lesson_id, event_date) or whatever you declared.
2. Drop implementation columns.
3. repartition(n) with n chosen so that output bytes divided by n land near 128–256 MB. For a 20 GB day, about 80 to 160 files.
4. write.mode("overwrite").partitionBy("event_date") only if the overwrite scope is the partition you computed. Overwrite of a single date must not delete other dates. Use a path that includes the date, or the dynamic partition overwrite mode you have tested (spark.sql.sources.partitionOverwriteMode set to dynamic). The default static overwrite can replace the whole table path. Know which mode the session uses before the first production write.
5. List the directory. Record file count, min size, and max size. A max many times the median means the round-robin repartition did not happen and a hot key still owns a task.

dynamic partition overwrite is a real Spark SQL setting. It overwrites only the partitions present in the DataFrame being written. It is the tool that makes a rerun of Tuesday leave Wednesday in place. Test it. A static overwrite of the table root is how a backfill deletes the month.

What not to tune in the same change

Change one of the following at a time: adaptive coalescing, a cache, executor memory, shuffle partitions, the write partitioner. A job that gets all five in one deploy teaches you nothing when it improves, and too many things when it regresses. Record stage duration, shuffle bytes, spill, and output file count for the date you ran.

Practical Implementation or Design Walkthrough

Problem

Take the engagement DataFrame from lesson 30, aggregate it to a daily course summary for one event_date, and write Parquet that a later reader can open without listing thousands of files. Show the plan. Persist only the intermediate that two actions share.

Target schema

course_day, grain (course_id, event_date):

Unknown lessons stay inside event_count and outside matched_lesson_count. That is the audit. The summary does not store learner keys, so this particular sink is not the place the salt could leak through a learner grain. The salt must still be dropped before the groupBy, or it splits the aggregate.

Code

pythonCopy
"""One-date course summary: project, aggregate, size the files, explain.

PySpark 3.5.x. Paths are local stand-ins for the warehouse.
"""

from pyspark.sql import SparkSession, DataFrame
from pyspark.sql import functions as F
from pyspark import StorageLevel

EVENT_DATE = "2026-03-02"
# 20 GB compressed would want ~100 files. The fixture is small; the argument is explicit.
OUTPUT_FILES = 8


def build_spark() -> SparkSession:
    return (
        SparkSession.builder.appName("tlp-course-day")
        .master("local[*]")
        .config("spark.sql.adaptive.enabled", "true")
        .config("spark.sql.adaptive.coalescePartitions.enabled", "true")
        # Respect the advisory size instead of maximizing tiny-cluster parallelism.
        .config("spark.sql.adaptive.coalescePartitions.parallelismFirst", "false")
        .config("spark.sql.adaptive.advisoryPartitionSizeInBytes", str(64 * 1024 * 1024))
        .config("spark.sql.adaptive.skewJoin.enabled", "true")
        .config("spark.sql.shuffle.partitions", "16")
        .config("spark.sql.sources.partitionOverwriteMode", "dynamic")
        .getOrCreate()
    )


def course_day(events_joined: DataFrame) -> DataFrame:
    # Implementation columns and wide attributes stay out of the aggregate shuffle.
    projected = events_joined.select(
        "course_id",
        "event_date",
        "manifest_id",
        "learner_key",
        F.col("enrolled_on").isNotNull().alias("is_enrolled"),
    )
    return projected.groupBy("course_id", "event_date").agg(
        F.count("*").alias("event_count"),
        F.count("manifest_id").alias("matched_lesson_count"),
        F.count(F.when(F.col("is_enrolled"), F.lit(1))).alias("enrolled_event_count"),
    )


def write_course_day(summary: DataFrame, path: str) -> None:
    (
        summary.repartition(OUTPUT_FILES)
        .write.mode("overwrite")
        .partitionBy("event_date")
        .parquet(path)
    )


def main() -> None:
    spark = build_spark()
    # Stand-in for engagement_join(...).select already narrowed.
    joined = spark.read.parquet("/tmp/tlp/joined").where(F.col("event_date") == EVENT_DATE)

    # Two actions share this aggregate: an audit count and the write.
    summary = course_day(joined).persist(StorageLevel.MEMORY_AND_DISK)
    audit = summary.agg(
        F.sum("event_count").alias("events"),
        F.sum("matched_lesson_count").alias("matched"),
    ).collect()[0]
    print(f"events={audit['events']} matched={audit['matched']}")

    summary.explain(mode="formatted")
    write_course_day(summary, "/tmp/tlp/course_day")
    summary.unpersist()
    spark.stop()


if __name__ == "__main__":
    main()

count("manifest_id") ignores nulls, which is what makes matched_lesson_count the matched subset. count("*") does not. That pair is the audit. If they are equal every day, including days when you planted a missing lesson, the left join was lost upstream.

What to record from a run

From the SQL tab: the final adaptive plan and whether partition counts shrank. From the stage that aggregates: median task time, max task time, spill (disk) sum, shuffle write sum. From the output directory: file count under event_date=2026-03-02, and the size of the largest file. A local fixture will show small files. The metric to practice is that you looked. Production fails the release if the date directory has more than a few hundred files or any file far above 1 GB, thresholds you write into the capstone checklist rather than discovering during an incident.

Dependencies

pyspark==3.5.3 or a later 3.5 patch is enough for this script. No extra package is imported. Hadoop's S3A connector is how the same write.parquet talks to object storage on a cluster; the path becomes s3a://... when the cluster documents that scheme. This course does not invent a wrapper client around the write.

Enterprise Real-World Scenarios

Scenario 1: The runbook that disabled adaptive execution

A platform team set spark.sql.adaptive.enabled to false in 2020 because an early 3.0 build changed a plan the regression suite had pinned. The flag stayed in the Spark submit for every engagement job. In 2026 the join is a healthy sort-merge of a broadcast plus a large enrollment, and the shuffle is 40 GB. Two hundred partitions are the right order of magnitude on some days and five times too many on weekend traffic. With adaptive execution off, weekend jobs schedule two hundred tasks of a few megabytes and spend as long in the scheduler as in the scan. Skew split is also off, so Monday's launch course is a single long task again.

The change is to enable adaptive execution, set coalescePartitions.parallelismFirst to false on the production pool, and keep the skew-join factor at 5 and the byte threshold at 256 MB until a measured partition says otherwise. The regression suite should pin row counts and the output schema, not the operator tree of a 2020 build. Capture one formatted explain from the UI after the change. Expect coalesced partitions on the small day and a skew split, or an accepted salt, on launch day. Do not combine this change with an executor-memory increase. If the job gets faster, you want to know which lever did it.

Scenario 2: A cache that caused the spill

An analyst notebook, later pasted into the job, caches the raw event frame "so the join is instant," then joins, then caches the join, then caches the aggregate. Each cache() is MEMORY_AND_DISK. The aggregate's sort spills even though the cluster is large, because storage is holding three copies and execution is evicting and recomputing them. The SQL is the same SQL that runs cleanly in a fresh session.

The repair is one persist on the aggregate, which is the object two actions share, and unpersist after the write. The raw scan stays uncached and is narrowed by select and the date filter before anything else. Executor GC time drops because the extra copies are gone. Spill on the aggregate drops if the remaining cause was memory competition rather than a hot key. If spill remains on one task, return to the lesson 30 skew work. The cache was never a skew fix.

Scenario 3: Eight thousand files and a downstream scan

The engagement job wrote partitionBy("event_date", "course_id") from two hundred tasks. The next morning's completion SQL, still a Spark job in this module, listed tens of thousands of files for a week of dates and opened each one. The driver's listing time exceeded the scan. Readers on a warehouse engine pointed at the same path showed the same pattern: planning time dominated, and bytes read were modest.

The rewrite partitions only by event_date, repartitions to a file count derived from output bytes, and turns on dynamic partition overwrite so Tuesday's rerun does not delete Wednesday. A one-time compaction job rewrites the historical dates the same way. The downstream job's input file count for one date falls from thousands to about a hundred. That is the performance win, and it is a layout win. No executor type changed.

Best Practices

• Enable adaptive execution unless a named Spark bug forces you off it, and record the ticket in the same line as the flag.
• On a production pool, set spark.sql.adaptive.coalescePartitions.parallelismFirst to false when you want the 64 MB advisory size. Say so in the job config, not in folklore.
• Size spark.sql.shuffle.partitions as an initial count. Let coalescing shrink it. Do not use 200 as a moral standard.
• Project to keys and measures before a shuffle. Attach wide attributes after the aggregate when the output needs them.
• Persist a DataFrame only when a second action would recompute a shuffle or a large scan. Unpersist when that action finishes.
• Remember DataFrame cache() includes disk. Remember RDD cache() does not.
• Partition curated Parquet by a low-cardinality filter column. Control file count with repartition or coalesce toward 128–256 MB.
• Set spark.sql.sources.partitionOverwriteMode to dynamic only after a test that a one-date overwrite leaves other dates in place.
• Change one lever per comparison: plan, memory, or layout. Record spill, shuffle bytes, stage time, and file count.
• Prefer Spark SQL functions over Python UDFs on the filter and join path so pruning and codegen survive.

Common Mistakes and Anti-Patterns

Disabling adaptive execution as a standard. You lose coalescing, the runtime broadcast switch, and skew split together. Pin tests to results, not to a frozen operator tree.

Treating spark.sql.shuffle.partitions as the only partition knob. The initial count, the advisory size, parallelismFirst, and the sink's repartition answer different questions. Moving one number and hoping is how tiny tasks and huge files coexist.

Caching the raw input by default. The cache holds columns you will drop, competes with the sort, and is evicted under pressure so the second action recomputes anyway. Cache the narrow reused frame.

Assuming RDD cache semantics on a DataFrame. A DataFrame that does not fit spills to local disk and still "succeeds." You may not notice until the disk and the runtime do. Choose the level on purpose.

partitionBy on course_id or learner_key. File count becomes tasks times cardinality. Listing the path dominates the next job. Partition by event_date.

Static overwrite during a backfill. The write replaces every partition under the table path. Tuesday's repair deletes the rest of the month. Use a date path or dynamic overwrite you have tested, and keep a backup of the previous snapshot until the audit counts match.

Enabling Kryo to speed a SQL shuffle. The DataFrame shuffle format does not use that serializer. The diff in shuffle bytes will be flat. Spend the change on column pruning instead.

A Python UDF wrapped around a date filter. Partition pruning never sees the predicate. The scan reads the month, then the UDF throws rows away. Filter with F.col("event_date") == ... first.

Raising executor memory while one task spills and the rest do not. You bought a bigger box for a hot key. Split or salt the key. Add memory when spill is widespread after the rows are narrow and the partitions are near the advisory size.

Optimizing five flags in one deploy. The next regression has no parent. One change, one set of stage metrics.

Troubleshooting and Design Review

Symptom: Weekend jobs have hundreds of tasks that finish in under a second, and scheduler delay is visible in the stage timeline.

Likely cause: spark.sql.shuffle.partitions is 200, the shuffle is small, and adaptive coalescing is off or parallelismFirst is keeping extra tasks.

Investigation: open the final plan and read the partition count on the shuffle. Check spark.sql.adaptive.enabled and coalescePartitions.enabled.

Corrective action: turn coalescing on. On the production pool, set parallelismFirst to false and compare task counts. Do not drop the initial partition count to 10 on the same day you might run a launch-course skew; coalescing shrinks small shuffles, and skew split handles the large partition.

Symptom: Every task in the aggregate spills to disk. Keys are evenly distributed. Rows in the shuffle are wide.

Likely cause: execution memory per task is below the sort's need, or the row carries unused strings.

Investigation: shuffle write bytes divided by row count is the row width. The scan's ReadSchema shows extra columns. Spill is on all tasks, not one.

Corrective action: project before the aggregate. If spill remains and partitions are already near 64–128 MB, add executor memory. If one task accounts for the spill, go back to skew.

Symptom: The second action is as slow as the first, and the storage tab shows the cached block evicted.

Likely cause: execution reclaimed the cache during the first action's sort, or unpersist ran too early, or the cache was on a parent frame and the child added a wide transformation that did not hit it.

Investigation: storage tab level and size, and whether the second job's DAG includes the scan.

Corrective action: persist the frame the second action uses, at MEMORY_AND_DISK, after the narrow project. Give the sort fewer competing caches. Confirm the second job's DAG starts at the InMemoryTableScan or equivalent cached read.

Symptom: A one-date rerun deleted other dates.

Likely cause: partitionOverwriteMode is static, and the write target is the table root with partitionBy("event_date").

Investigation: the session config and the output path. List partitions before trusting the job.

Corrective action: restore the snapshot, set dynamic overwrite or write directly to the date path, and add a test that writes date A, writes date B, rewrites date A, and still reads date B.

Symptom: Executor containers are killed. The JVM heap is not full. Python workers are large.

Likely cause: spark.executor.memoryOverhead is too small for the Python processes, or a UDF is holding data.

Investigation: the node-manager kill reason, and whether the plan contains a Python eval node.

Corrective action: replace the UDF with SQL functions where you can. Raise overhead only for the Python that remains. Do not raise heap and overhead without looking at which one the kill used.

A design review asks: What is the initial shuffle count, and what did adaptive execution do to it on a small day and a large day? Which DataFrame is persisted, at which level, and which action unpersists it? Where can spill show up, and is it one task or every task? What is the sink partition column and the target file size? What does a rerun of one date do to other dates? Which single change is in this deploy? A review that adds executors without those answers is a cost change wearing a performance hat.

Interview Questions and Answers

Q1. What does adaptive query execution change at runtime that a plan printed before the job cannot already know?

It uses completed shuffle sizes. It coalesces small adjacent partitions toward the advisory size, it can replace a sort-merge with a broadcast hash join when a measured side is under the adaptive broadcast threshold, and it splits skewed sort-merge partitions that exceed both the factor and the byte threshold. The pre-run explain is the starting plan. The SQL tab after the stage is the plan that ran. I compare them. Adaptive execution does not prune Parquet columns, does not fix a Python UDF, and does not choose the sink's file count.

Q2. Why can cache() on a DataFrame and cache() on an RDD behave differently when the data is larger than memory?

DataFrame.cache() persists at MEMORY_AND_DISK, so the extra blocks spill to executor local disk and the next action can still read them. RDD.cache() persists at MEMORY_ONLY, so an evicted partition is recomputed from lineage. The method name is the same and the storage level is not. I call persist with an explicit level when the choice matters, and I unpersist when the reuse is over. A cache that loses to execution memory under the unified region can still recompute, which is why I do not cache three generations of the same lineage.

Q3. When is caching the wrong optimization?

When the DataFrame is used by one action. The cache writes bytes you read at most once, and those bytes compete with the sort and the join inside spark.memory.fraction. I cache a narrow projection that an audit action and a write both need. I leave the raw scan uncached. I also skip the cache when the "second action" is only explain, which does not need the rows, or count on a parent I could have aggregated inside the write path.

Q4. What is the relationship between spark.sql.shuffle.partitions and spark.sql.adaptive.advisoryPartitionSizeInBytes?

The first is the initial reduce count, default 200. The second is the target size, typically 64 MB, that coalescing aims at after the shuffle is measured. Coalescing merges small partitions. It does not split a partition that is already far above the target; skew handling does that for eligible joins. coalescePartitions.parallelismFirst, default true, may keep more partitions than the advisory size alone would, to protect parallelism on small clusters. On a large pool I set it to false when I want the advisory size to be the real target, and I record that next to the config.

Q5. What do spill (memory) and spill (disk) tell you during a sort-merge or a hash aggregate?

The operator could not keep its working set in execution memory, so it wrote runs to local disk and merged them. The result is still correct. The task is slower and the disk is busy. If a single task owns the spill, the partition is skewed or huge, and more executor memory is the wrong first purchase. If every task spills and the rows are narrow and near the advisory size, memory per task is genuinely tight. I check row width in the shuffle before I buy the memory, because unused columns spill too.

Q6. How do you use explain modes in a review?

formatted is the attachment I want: operators and arguments, including join keys and partition counts. extended is how I show that a filter survived into the optimized plan. cost shows statistics, which I treat as estimates. codegen is for a case where an expression unexpectedly failed to fuse. simple is too thin for a join change. I also look at the final adaptive plan in the UI, because explain before execution may not show coalescing that happens after the first stage.

Q7. A write uses partitionBy("event_date", "course_id") from 200 tasks. What goes wrong, and what do you write instead?

Each task writes a file per course it holds, so the date directory fills with thousands of small files. The next job spends its time listing and opening them. I partition only by event_date, then repartition(n) so files land around 128–256 MB, and I list the directory to confirm. I set partition overwrite mode to dynamic, or I write to a date-specific path, so a rerun of one date does not remove the others. I do not partition by learner_key.

Q8. Why does switching spark.serializer to Kryo leave a DataFrame join's shuffle bytes unchanged?

SQL shuffle uses the unsafe row format and the I/O compression codec, lz4 by default. Kryo serializes RDD objects. The join never enters that path, so the bytes stay flat. I would turn Kryo on for an RDD-heavy job whose shuffle is Java serialization bound, and I would register classes deliberately. On this engagement job I prune columns instead.

Q9. A filter is a Python UDF on event_date. The scan still reads every partition. Why?

Catalyst cannot see inside the UDF, so partition pruning and Parquet predicate pushdown do not get the date predicate. The scan reads the files, moves the rows to a Python worker, and then drops them. Whole-stage codegen stops at that boundary. I filter with a column expression first, project the columns I need, and only then call Python for a rule SQL cannot express. Arrow execution can speed the transfer. It does not restore pruning.

Q10. Walk through the memory decision before you ask for larger executors.

I confirm the plan: broadcast where lesson 30 said broadcast, adaptive coalescing on, one cache or none. I look at spill. One spilling task sends me to skew, not to a bigger machine. Spill on every task, after the row is narrow and partitions are near the advisory size, is the case for more memory per task or fewer concurrent tasks. If the container is killed while the JVM heap is calm, I look at spark.executor.memoryOverhead and Python workers. I change one of those, rerun the same date, and compare stage time, spill, and shuffle bytes. Larger executors are the proposal I make after that note, not before.

Q11. What does spark.sql.sources.partitionOverwriteMode=dynamic promise, and what must you test?

It overwrites only the partitions present in the write, and it leaves other partitions of the table in place. Static mode overwrites the directory set the write is aimed at, which can be the whole table. I test by writing two dates, rewriting one, and reading both. I also confirm the session that submits the job actually sets the mode. A cluster default of static silently defeats the line in a notebook.

Hands-On Exercises

Exercise: One change, one measurement

Problem statement. You have a joined events fixture for a single event_date and a course-day aggregate. The baseline job disables adaptive execution, caches the raw read, and writes partitionBy("event_date", "course_id").

Tasks.

1. Run the baseline. Record stage time, the max task time, spill totals if any, the scan schema, and the output file count.
2. Enable adaptive execution and set coalescePartitions.parallelismFirst to false. Change nothing else. Record the same metrics and the final shuffle partition count from the plan.
3. Remove the raw cache. Persist only if you have two actions that share the aggregate. Unpersist after the write.
4. Rewrite the sink partitioned only by event_date, with an explicit output file count. Turn on dynamic partition overwrite and demonstrate that a second date survives a rewrite of the first.
5. Replace any Python UDF filter with a column predicate and show that the formatted plan's scan schema narrowed.
6. Write five sentences: which single change moved which metric, and which change you would not ship because the counts moved.

Expected deliverables. The script, the two plans, the metric table, and the file listing.

Evaluation rubric.

You pass when the file listing and the plan agree with the note, and the event counts before and after the rewrite match.

Lesson Summary

Spark performance on this job is a plan you can read. Adaptive execution coalesces small shuffle partitions, can switch a measured small side to broadcast hash, and splits eligible skewed sort-merge partitions. The initial partition count and the advisory size are different settings, and parallelismFirst decides whether the advisory size is honored. DataFrame cache() is memory and disk, it competes with execution memory, and it pays off only when a second action reuses the rows. Spill on one task is skew. Spill on every task, after the row is narrow, is memory. The sink should partition by event_date and land files near a hundred-plus megabytes, with a tested overwrite mode so one date's rerun leaves the others. Kryo and a larger cluster are real tools for other problems. They are not the first response to this plan.

The next lesson, Structured Streaming with Spark, takes the same events as a stream: triggers, watermarks, and checkpoints. The batch file layout you just wrote is what a streaming file sink has to live beside without producing a second, conflicting grain.

Further Reading

• Spark 3.5 SQL performance tuning — adaptive execution, skew join, and broadcast thresholds. This is the page behind the defaults in this lesson.
• Spark 3.5 configuration — spark.memory.fraction, spark.memory.storageFraction, shuffle partitions, and serializer settings.
• Spark 3.5 SQL programming guide — explain, persist, and StorageLevel.
• Spark 3.5 PySpark DataFrame.explain — the mode strings simple, extended, codegen, cost, and formatted.
• Data Formats and Serialization — why the file sizes you write become the scan costs of the next job.