Spark Core · Module 4 of 5

How Spark decides how to run your query

Catalyst rewrites your plan before it runs; AQE rewrites it again while it runs. Knowing both lets you read explain() and predict what Spark will do.

Catalyst phases Optimizer rules Pushdown & pruning Reading explain()Join strategies AQE UDF cost Check yourself

Catalyst: four phases

SQL and the DataFrame API compile to the same plan, which is why they perform the same. Catalyst turns that plan into executable code in four phases.

Analysis. Resolves table and column names and types against the catalog. A misspelt column fails here with AnalysisException, before any job runs.
Logical optimisation. Rule-based rewrites that don't depend on the engine: pushdown, pruning, constant folding, combining filters, simplifying expressions.
Physical planning. Chooses concrete operators, e.g. broadcast hash join vs sort-merge join, using statistics and thresholds.
Code generation. Tungsten fuses operators in a stage into one Java function (whole-stage codegen, shown as *(n) in plans), works on compact binary rows in off-heap memory, and avoids JVM object overhead.

On Databricks, Photon replaces much of step 4 with a vectorised C++ engine. The plan logic is the same; operators show up as Photon… in the plan.

Optimizer rules you can see

Pick a rule to see the logical plan before and after.

Before

After

Predicate pushdown, partition pruning and DPP

Three ways Spark avoids reading data. Interviewers often ask for the difference.

Technique	What gets skipped	How	See it in explain()
Predicate pushdown	Row groups inside files	Filter is passed to the Parquet/ORC reader, which uses min/max statistics per row group (and Delta file stats) to skip blocks that can't match	`PushedFilters: [EqualTo(country,AE)]`
Column pruning	Unused columns	Columnar formats read only the columns the query needs	`ReadSchema: struct<…>`
Partition pruning	Whole partition directories	Filter on a partition column (e.g. `dt = '2026-10-01'`) lists only matching folders	`PartitionFilters: [dt = …]`
Dynamic partition pruning	Fact partitions, decided at runtime	Filter on a dimension (`d.year = 2026`) produces the matching keys at runtime, which then prune the fact table's partitions	`dynamicpruningexpression(…)`

What breaks pushdown

Wrapping the column in a function or cast: WHERE to_date(ts) = '2026-10-01' or upper(country) = 'AE'. Filter on the raw column with a range instead.
Python UDFs in the filter: Spark can't see inside them.
Formats without statistics (CSV, JSON).
Badly clustered data: if every file contains every value, min/max stats can't skip anything. That's what ZORDER and liquid clustering fix.

Reading a physical plan

Run df.explain("formatted") (or EXPLAIN FORMATTED in SQL). Read it bottom up: scans at the bottom, result at the top. This is our pipeline from Module 2 with a small customers table.

AdaptiveSparkPlan isFinalPlan=false
+- HashAggregate(keys=[segment], functions=[sum(amount)])
   +- Exchange hashpartitioning(segment, 200)
      +- HashAggregate(keys=[segment], functions=[partial_sum(amount)])
         +- Project [amount, segment]
            +- BroadcastHashJoin [cust_id], [cust_id], Inner, BuildRight
               :- Filter (isnotnull(country) AND (country = AE))
               :  +- FileScan parquet [cust_id,amount,country]
               :       PushedFilters: [IsNotNull(country), EqualTo(country,AE)]
               :       ReadSchema: struct<cust_id:bigint,amount:double,country:string>
               +- BroadcastExchange HashedRelationBroadcastMode(...)
                  +- FileScan parquet [cust_id,segment]

Look for	Meaning
`Exchange`	A shuffle = a stage boundary. Count them; each one is network and disk cost.
`BroadcastHashJoin` / `SortMergeJoin`	Join strategy chosen. SMJ on a small table is a tuning opportunity.
`partial_sum` → `sum`	Map-side pre-aggregation, the DataFrame version of reduceByKey.
`PushedFilters`, `PartitionFilters`	Whether your filters reached the source. Empty lists mean full scans.
`AdaptiveSparkPlan isFinalPlan=false`	AQE is on. The final plan (after running) can differ; check the SQL tab of the Spark UI.
`*(2)` prefixes	Operators fused into whole-stage codegen block 2 (in the non-formatted output).

Join strategies

For an equi-join, Spark picks roughly in this order: join hint → broadcast if a side is small enough → sort-merge join (default for two large tables) → shuffle hash join when enabled and suitable. Non-equi joins fall back to nested-loop strategies.

Strategy	How it works	Use when	Watch out
Broadcast hash join	Small side collected to driver, sent to every executor, hashed; big side joined in place	One side < `spark.sql.autoBroadcastJoinThreshold` (10 MB default; often raised to 50–200 MB)	Broadcasting something too large → driver/executor OOM. Can't broadcast the preserved side of an outer join (e.g. the left side of a LEFT JOIN).
Sort-merge join	Shuffle both sides by key, sort, merge	Two large tables; scales to any size	Two shuffles plus sorts; very sensitive to skew
Shuffle hash join	Shuffle both sides, build hash table on the smaller side per partition	One side medium-sized, partitions fit in memory	OOM if a partition's build side doesn't fit
Broadcast nested loop / cartesian	Compare every pair	Non-equi joins (`a.ts BETWEEN b.start AND b.end`)	Quadratic; avoid on large inputs

from pyspark.sql.functions import broadcast
orders.join(broadcast(customers), "cust_id")         # force broadcast
spark.sql("SELECT /*+ BROADCAST(c) */ * FROM orders o JOIN customers c USING (cust_id)")
# other hints: MERGE (sort-merge), SHUFFLE_HASH, SHUFFLE_REPLICATE_NL

Adaptive Query Execution

Catalyst plans with estimates. AQE (on by default since Spark 3.2) re-plans between stages using the real sizes measured at each shuffle.

Without AQE

With AQE

The configs worth naming

Config	Default	Purpose
`spark.sql.adaptive.enabled`	true	Turns AQE on (Spark 3.2+)
`…adaptive.coalescePartitions.enabled`	true	Merge small post-shuffle partitions
`…adaptive.advisoryPartitionSizeInBytes`	64 MB	Target size when coalescing or splitting
`…adaptive.skewJoin.enabled`	true	Split skewed partitions in sort-merge joins
`…skewJoin.skewedPartitionFactor`	5	Skewed if > 5 × median size …
`…skewJoin.skewedPartitionThresholdInBytes`	256 MB	… and also larger than this
`spark.sql.autoBroadcastJoinThreshold`	10 MB	Static broadcast limit (AQE has its own runtime setting)

What AQE can't fix

Skew in a groupBy aggregation or window function (skew-join handling covers joins only), skew in the very first scan stage, a bad file layout, or a Python UDF bottleneck. Those still need salting, repartitioning or redesign.

Python UDF vs pandas UDF vs built-ins

	Built-in functions	pandas UDF (vectorised)	Plain Python UDF
Runs in	JVM / Photon, codegen	Python worker, Arrow batches	Python worker, row at a time
Data transfer	None	Columnar Arrow, efficient	Pickle per row, slow
Catalyst can optimise	Yes	No (black box)	No (black box)
Typical speed	Fastest	Often 3–100× faster than Python UDF	Slowest
Use when	Always first	Logic needs pandas/NumPy or ML libraries	Small data or no alternative

Python workers use memory outside the JVM heap, so UDF-heavy jobs often need more memoryOverhead. Spark 3.5+ can also run regular Python UDFs over Arrow (useArrow=True).

Say it in an interview

"Catalyst analyses the plan against the catalog, applies rule-based rewrites like predicate pushdown and column pruning, picks physical operators such as the join strategy by cost, then generates code with Tungsten. AQE then re-optimises at each shuffle using real sizes: it coalesces small partitions, switches sort-merge to broadcast when a side turns out small, and splits skewed join partitions."

Check yourself

Your filter is on date(event_ts) = '2026-10-01' and the table is partitioned by event_date. Why is it slow?

The filter is on a function of a non-partition column, so partition pruning doesn't happen and pushdown is weakened. Filter on event_date = '2026-10-01', or on an event_ts range.

A 30 MB dimension is joined with a sort-merge join. Two ways to fix it?

Raise autoBroadcastJoinThreshold above 30 MB, or add a broadcast() hint. With AQE on, Spark may also switch at runtime once it measures the filtered side.

Why can explain() show SortMergeJoin but the job actually ran a broadcast join?

AQE. explain() shows the initial plan (isFinalPlan=false); AQE changed it at runtime. Check the SQL / DataFrame tab in the Spark UI for the final plan.

Difference between predicate pushdown and partition pruning?

Partition pruning skips whole directories based on partition-column filters. Pushdown sends filters into the file reader to skip row groups inside the files that are still read, using column statistics.

Why are SQL and DataFrame code equally fast?

Both produce the same unresolved logical plan and go through the same Catalyst pipeline.