Spark EXPLAIN Plans

What EXPLAIN does and why it matters

An EXPLAIN plan is a textual representation of how Spark will execute your query. Every DataFrame transformation passes through Catalyst, Spark’s query optimizer, which rewrites your logical operations into an efficient physical execution strategy. The raw logical plan is what Catalyst sees directly from your code; the optimized logical plan shows what rewriting rules have been applied; the physical plan is the actual sequence of executable operations that will run on the cluster.

The importance cannot be overstated: a query that looks reasonable in code can explode in cost because of a filter placed after an expensive join, an unexpected Cartesian product, or a skewed partition that brings a single executor to a crawl. The EXPLAIN plan reveals all of this instantly. Three explain modes exist:

• simple (default): prints only the physical plan.
• extended: shows both logical and optimized logical plans, plus the physical plan.
• codegen: additionally includes the generated Java bytecode that Spark compiles and executes.

Most of the time, explain("extended") is all you need. Codegen is useful only when you suspect code generation itself is a bottleneck.

Reading the physical plan

The physical plan is a tree of stages and operations, read from bottom to top (data flows up). Each line represents one operation, indented to show nesting. The structure matters: operations nested deeper run first, with results flowing upward to their parents.

Aggregate (count(1) [count#123L])
  +- Exchange hashpartition(col_a#10, 200)
       +- Aggregate (partial_count(1))
            +- Filter (col_a#10 > 100)
                 +- Scan parquet [col_a#10, col_b#20]

Reading bottom-up: Spark scans the Parquet file, applies a filter, does a partial count aggregation, then shuffles (Exchange) the partial results to 200 partitions, and finally performs the global aggregation. The number after Exchange is the target partition count; it is often a tuning knob.

Watch for these indicators in the physical plan:

• Exchange = shuffle. Every shuffle is expensive; multiple shuffles in a plan are a red flag.
• SortMergeJoin = sorted join, usually necessary for large tables. Can be slow if sort cost is high.
• BroadcastHashJoin = one table broadcast to all executors. Fast if the broadcast fits in memory; fails silently if broadcast config is too small.
• Scan with high partition count = fine-grained work; with 1 partition = a bottleneck (all data on one executor).

Logical vs. optimized logical plans

The logical plan is your query as written, before any rewriting. A simple select-filter-join might look like:

Aggregate [col_a#10]
  Filter (col_b#20 > 100)
    Join Inner, (tab1.id = tab2.id)
      Scan table1 [col_a, col_b]
      Scan table2 [id]

The optimized logical plan is what Catalyst produces after applying rewriting rules. It might reorder the filter to run before the join (predicate pushdown), or eliminate redundant aggregations. These rewrites are automatic and correct; they exist to lower cost. If the optimized plan looks substantially different from the logical one, it is a sign the query was inefficient to begin with.

The golden rule: push filtering as close to the scan as possible. A filter after a join processes millions of rows; a filter before a join processes fewer rows before the expensive operation. Catalyst does this automatically through predicate pushdown, but if you have a complex query with subqueries and CTEs, inspect the optimized plan to confirm filters are not stranded downstream.

Join strategies: broadcast vs. shuffle

The type of join in the physical plan often dominates performance. Spark chooses among three:

BroadcastHashJoin: One table (the ‘broadcast side’) is sent to every executor as a HashSet; the other table scans and probes. Zero shuffle cost, extremely fast, but only works if the broadcast table fits in the executor’s memory. Spark has a heuristic: tables smaller than spark.sql.autoBroadcastJoinThreshold (default 10 MB) are broadcast automatically. If you want to force a broadcast, use broadcast() in your code; if Spark refuses to broadcast because it thinks the table is too big (but you know it fits), increase the threshold.

SortMergeJoin: Both tables are sorted on the join key, then scanned in parallel. Expensive because of the sorts, but works for tables of any size and is Spark’s default for large joins. The physical plan will show Sort operations before the join.

ShuffleHashJoin: Rarely seen; Spark usually prefers SortMerge. Both tables are shuffled by join key, then co-partitioned for a local hash join.

A common mistake is to join two large tables when one or both could be broadcast. If you can reshape the problem so one side is small and static, do it; a broadcast join is orders of magnitude faster than a shuffle join. Conversely, if you broadcast a 5 GB table to 1000 executors, you will run out of memory; read the memory errors in executor logs and either filter the broadcast table or fall back to shuffle join.

Shuffle: partitions, skew, and parallelism

Every Exchange in the plan is a shuffle: Spark repartitions data by sending all rows with the same partition key to the same executor. Shuffles are the most expensive operation in Spark because data must be written to disk, sent over the network, and then read back. A good execution plan minimizes shuffles.

The partition count matters hugely. If you shuffle to 200 partitions and each partition is 1 MB, you have 200 tiny tasks, each doing almost nothing — scheduling overhead dominates. If you shuffle to 8 partitions and each is 500 MB, you have severe load imbalance and head-of-line blocking. The rule of thumb is 128 MB per partition: divide your data size by 128 MB to get a starting partition count, then tune from there.

Skew is the silent killer. If the shuffle key is heavily skewed (a few keys appear millions of times, most appear rarely), some partitions will be enormous and others tiny. A single executor stuck with the skewed partition will take 100x longer than its peers, starving the entire job. Spark has no built-in skew handling; you must detect it (a small number of tasks taking forever in the DAG visualization) and fix it in code: repartition the small dimension, salt the skewed key, or use adaptive partitioning.

The EXPLAIN plan shows the partition count but not the data distribution, so you cannot spot skew from the plan alone. Use df.rdd.getNumPartitions() to see the number and histogram task times in the Spark UI to spot skew.

Broadcast operations and memory pressure

Every BroadcastHashJoin in the plan means one table is being broadcast. This is good for speed but dangerous for memory. When Spark broadcasts a table, it sends the entire table to every executor, not a subset. A 500 MB table broadcast to 100 executors consumes 50 GB of memory across the cluster — and if each executor has only 4 GB, it will fail.

Three related settings control this:

• spark.sql.autoBroadcastJoinThreshold: tables smaller than this are broadcast automatically. Default 10 MB. Set to -1 to disable auto-broadcast.
• spark.sql.broadcastTimeout: timeout for the broadcast operation itself. If a broadcast takes longer than this, it fails.
• spark.broadcast.blockSize: size of blocks during broadcast. Usually not tuned.

If an EXPLAIN plan shows a BroadcastHashJoin but the job fails with an out-of-memory error on executors, the broadcast table is too big for the executor memory. Options: (1) reduce executor count so the same data spread across fewer nodes means larger heap per node, (2) increase executor memory, or (3) disable auto-broadcast and force a SortMergeJoin by lowering the threshold or using df.join(other, ...) instead of broadcast(df).join(...).

Aggregations and partial aggregates

Aggregation in Spark happens in two phases: partial and final. The physical plan shows this clearly:

Aggregate (final_count, sum_col, etc.)
  +- Exchange hashpartition(grouping_col)
       +- Aggregate (partial_count, partial_sum, etc.)

Spark first computes partial aggregates on each partition (count, sum, min/max on local data), then shuffles the partial results by grouping key, and finally merges them globally. This two-phase strategy is how Spark avoids sending all data to a single reducer.

If you see an aggregation with no Exchange before the final aggregate, that means the data was already partitioned by the grouping key, so Spark skips the shuffle — a huge win. If every aggregation in your query has an Exchange, consider whether you can repartition the source data once and reuse it for multiple aggregations.

Window functions and their cost

Window functions (ROW_NUMBER, RANK, SUM OVER, etc.) are implemented as shuffles plus sorts:

WindowFunction (row_number() OVER (PARTITION BY col_a ORDER BY col_b))
  +- Sort [col_b ASC]
       +- Exchange hashpartition(col_a)

Spark shuffles data by the PARTITION BY clause, then sorts by the ORDER BY clause within each partition. Window functions are not cheap; if you have nested window functions or multiple window functions, the plan will show multiple Exchange-Sort pairs. Try to consolidate them: compute all window functions you need in one pass rather than adding multiple window operations in sequence, because each adds overhead.

Practical debugging: when to optimize

Not every inefficiency in an EXPLAIN plan matters. A small query running in 100 milliseconds does not need optimization, even if the plan looks ugly. Focus on jobs that are:

• Running repeatedly (ETL pipelines, dashboards).
• Processing large data (billions of rows).
• On the critical path of user workflows.

For those, inspect the physical plan and ask:

1. How many shuffles? If more than one, can you eliminate any?
2. Which joins are SortMerge? Can you broadcast one side instead?
3. What is the partition count? Is it in the range of 100-500 for this data size?
4. Are there unnecessary sorts? Sorts appear in window functions and some joins; they are expensive on large data.
5. Is data pre-filtered? Or does the scan read all partitions and filter later?

A good EXPLAIN plan for a complex job typically has 1-2 shuffles maximum, at least one broadcast join if applicable, and filters applied close to the scans. If your plan has 10 exchanges and no broadcasts, something is wrong.

Comparing plans: before and after

The best use of EXPLAIN is comparative: write two versions of a query, explain both, and see which is more efficient. A few common transformations:

Rearranging joins: if you must join table A (10 GB) to table B (100 GB) to table C (5 GB), the order matters. Joining A to C first (10 GB join, smaller result) then to B is cheaper than joining A to B first (much larger intermediate result). Check the EXPLAIN plans and measure the difference.

Filtering earlier: if your query computes a join and then filters heavily on one side, move the filter before the join. Catalyst should do this automatically via predicate pushdown, but in complex queries with CTEs or subqueries, it sometimes does not. Check the optimized logical plan and add the filter explicitly if needed.

Caching and reuse: if your query uses the same subquery twice, consider caching it: df.cache().count() triggers materialization, so the plan for downstream operations will show a InMemoryTableScan instead of the full subquery logic. This only helps if the subquery is expensive and used more than once, so compare the two EXPLAIN plans side by side.

Common pitfalls and red flags

Cartesian products: if the EXPLAIN plan shows a Join with no condition (or condition omitted), you have a Cartesian product: every row of one table joins to every row of the other. On billion-row tables this is instant death. Check your join condition; missing ON clause is usually a typo.

Unexpected data type conversions: if the EXPLAIN plan shows Cast operations in the critical path, type mismatches might be forcing conversions. Ensure join keys, filter columns, and grouping columns are the same type (string to string, int to int) to avoid implicit casts.

Unbounded partition scans: if a table scan shows partition pruning is not being used (no PushedFilters in the Scan line), filters on partitioned columns are not being pushed down. Check whether the filter is a literal or a computed value; computed filters cannot always be pushed.

OOM during broadcast: BroadcastHashJoin in the plan but out-of-memory errors at runtime means the broadcast table is larger than executor memory. Disable the broadcast or increase executor memory.

Key takeaway

EXPLAIN plans are the most direct path to understanding and optimizing Spark workloads. Start with df.explain("extended") to see the logical and physical plans. Look for unnecessary shuffles, inefficient join strategies, and skewed partitions. Compare multiple versions of a query to see which Catalyst rewrites matter. Once you can read an EXPLAIN plan, every slow Spark job becomes a puzzle with a visible solution. The skill pays for itself in weeks on any production workload.