How to read a Spark physical plan, line by line

Read a Spark physical plan from the bottom up. The bottom lines read the data, the ones above filter and trim it, and every Exchange is a shuffle that ends one stage and starts the next. Joins and aggregates sit above the shuffles they need, and the top line is the last thing Spark does.

That is the whole method. The rest of this article reads a real plan from Spark 4.2 line by line, so that each word in it means something the next time you open one.

Print the plan

On a DataFrame, explain() prints the physical plan. A mode gives more:

df.explain()                   # the physical plan
df.explain(mode="formatted")   # an outline, then each step's details
df.explain(mode="extended")    # the 3 logical plans and the physical plan
df.explain(mode="cost")        # the logical plan with size estimates

In SQL, the same choices are EXPLAIN, EXPLAIN FORMATTED, EXPLAIN EXTENDED and EXPLAIN COST before the query. formatted is the easiest to read: a short tree with a number on each step, then a block per number with its columns, filters and arguments.

The query

We ran this on Spark 4.2.0, with 10 million orders and 1,000 customers in Parquet: orders over 90, joined to their customers, total spent per name, largest first.

top = (orders.where(F.col("amount") > 90)
       .join(customers, "customer_id")
       .groupBy("name").agg(F.sum("amount").alias("total"))
       .orderBy(F.desc("total")))
top.explain()

Spark printed this. We took out the column ids (name#4) and the file paths; nothing else is changed.

AdaptiveSparkPlan isFinalPlan=false
+- Sort [total DESC NULLS LAST], true, 0
   +- Exchange rangepartitioning(total DESC NULLS LAST, 200)
      +- HashAggregate(keys=[name], functions=[sum(amount)])
         +- Exchange hashpartitioning(name, 200)
            +- HashAggregate(keys=[name], functions=[partial_sum(amount)])
               +- Project [amount, name]
                  +- BroadcastHashJoin [customer_id], [customer_id], Inner, BuildRight
                     :- Filter ((isnotnull(amount) AND (amount > 90.0)) AND isnotnull(customer_id))
                     :  +- FileScan parquet [customer_id,amount] PushedFilters: [IsNotNull(amount), GreaterThan(amount,90.0), IsNotNull(customer_id)]
                     +- BroadcastExchange HashedRelationBroadcastMode
                        +- Filter isnotnull(customer_id)
                           +- FileScan parquet [customer_id,name] PushedFilters: [IsNotNull(customer_id)]

Each +- is a child of the line above it. Where the tree splits, :- starts the first branch and the +- at the same depth below starts the second.

Read it from the bottom

The scans. Each FileScan parquet reads 1 table, and the brackets list the only columns it reads: 2 of the 3 in orders. PushedFilters are handed to the Parquet reader so it can skip data that cannot match. A filter that shows up in Filter but not in PushedFilters is checked only after every row is read, which is worth knowing on a large table. PartitionFilters, in the full output, lists filters on the folders a table is partitioned by; it is empty here because these tables are not partitioned.

The filters. amount > 90 is ours. Spark added the isnotnull checks: an inner join never matches a null key, so those rows can go before the join instead of after.

The join. BroadcastHashJoin ... BuildRight means Spark builds the right side, customers, into a hash table and sends a copy to every executor, which is what BroadcastExchange above the customers scan does. It chose this because customers is far under the 10 MB broadcast threshold. The 10 million orders never move. If you expected a broadcast and see a SortMergeJoin with an Exchange and a Sort on each side, Spark thought both tables were too big; that is the line to check first when a join is slow. Sort-merge joinSort-Merge JoinSpark's default join for 2 large tables: it shuffles both by the join key, sorts each partition, then walks the 2 sorted sides together to find matching rows. explains that plan.

The 2 aggregates. A group by is always 2 HashAggregate lines. The lower one, partial_sum, adds up amounts per name inside each task. Then Exchange hashpartitioning(name, 200) shuffles those partial sums so every name lands in 1 of 200 partitions, and the upper HashAggregate, sum, adds them. Only 1 row per name per task crosses the network, not every order.

The sort. orderBy needs a second shuffle, rangepartitioning: rows go to partitions by ranges of total, the largest in the first. Sort [...], true then sorts each partition, and true means a global sort, so reading the partitions in order gives the whole result in order.

The top. AdaptiveSparkPlan isFinalPlan=false says adaptive query executionAdaptive Query ExecutionSpark's way of re-planning a query while it runs: after each shuffle it looks at the real sizes, then merges small partitions, splits skewed ones and can switch a join to a broadcast. is on and this is the plan before any data was read.

Step through it yourself, and compare it with the plan after the run:

Try itRead a Spark physical plan from the bottom up

Plan

Before the runAfter the run

1. AdaptiveSparkPlan isFinalPlan=false
2. +- Sort [total DESC NULLS LAST], true, 0
3. +- Exchange rangepartitioning(total DESC NULLS LAST, 200)
4. +- HashAggregate(keys=[name], functions=[sum(amount)])
5. +- Exchange hashpartitioning(name, 200)
6. +- HashAggregate(keys=[name], functions=[partial_sum(amount)])
7. +- Project [amount, name]
8. +- BroadcastHashJoin [customer_id], [customer_id], Inner, BuildRight
9. :- Filter ((isnotnull(amount) AND (amount > 90.0)) AND isnotnull(customer_id))
10. : +- FileScan parquet [customer_id,amount] PushedFilters: [IsNotNull(amount), GreaterThan(amount,90.0), IsNotNull(customer_id)]
11. +- BroadcastExchange HashedRelationBroadcastMode
12. +- Filter isnotnull(customer_id)
13. +- FileScan parquet [customer_id,name] PushedFilters: [IsNotNull(customer_id)]

Down a lineNext step up

Count the Exchanges

Every Exchange is a shuffleSpark ShuffleThe step that moves rows between executors so rows with the same key end up together, for a join, group by or repartition; it splits a job into stages.: data written to local disk, then read over the network by the next stage. This query has 2, plus a broadcast. The count is the quickest measure of what a query costs, and the place to look for one you did not mean to write: a groupBy that could have come after a filter, a repartition left in from testing, a join that could have been a broadcast.

The partition count in brackets, 200 here, is spark.sql.shuffle.partitions before any data is read. Our guide to choosing shuffle partitions explains how to size it.

The plan that ran is a different plan

explain() before the run printed isFinalPlan=false for a reason. With adaptive query execution, Spark runs the query in stages, measures each shuffle, then plans what is above it again. After top.collect(), the same DataFrame's executed plan read, in part:

AdaptiveSparkPlan isFinalPlan=true
+- == Final Plan ==
   ResultQueryStage 3
   +- *(4) Sort [total DESC NULLS LAST], true, 0
      +- AQEShuffleRead coalesced
         +- ShuffleQueryStage 2
            +- Exchange rangepartitioning(total DESC NULLS LAST, 200)
               +- *(3) HashAggregate(keys=[name], functions=[sum(amount)])
                  +- AQEShuffleRead coalesced
                     +- ShuffleQueryStage 1
                        +- Exchange hashpartitioning(name, 200)
                           +- *(2) HashAggregate(keys=[name], functions=[partial_sum(amount)])
                           ...
+- == Initial Plan ==
   ...

3 things are new:

• Query stages. BroadcastQueryStage 0, ShuffleQueryStage 1 and 2, ResultQueryStage 3: each ran on its own and was measured before the next was planned.
• AQEShuffleRead coalesced. The shuffles held partial sums for 201 names, far too little for 200 tasks, so Spark merged neighbouring partitions into fewer tasks. The 200 in the Exchange is what was written, not what ran.
• Stars and numbers. *(2) marks whole-stage code generation: every operator with the same number is compiled into 1 function and runs in the same task. Here the scan, filter, join, project and partial aggregate are all *(2). An Exchange never has a star, because it moves data between tasks. The Spark UI shows each number as 1 box, WholeStageCodegen (2), which is how you match a box in the SQL tab to a line here.

The final plan can change the join, too. In another query of ours, a filter left 103 rows on one side; explain() showed a SortMergeJoin and the final plan a BroadcastHashJoin over AQEShuffleRead local. If you read only the plan before the run, you read the wrong one.

On Databricks it is the same: DataFrame.explain() shows the initial and the current plan, and SQL EXPLAIN, which does not run the query, shows only the initial plan.

Logical plans, briefly

explain(mode="extended") prints 3 logical plans before the physical one. The parsed plan is the query as written. The analyzed plan resolves every column and table, so a typo in a column name fails here. The optimized plan is after rules such as pushing filters down and dropping unused columns. Read them when the physical plan does something you did not ask for, to see which step changed it. For a slow query, the physical plan is the one that matters: it is what ran.

Where to find it in the Spark UI

The SQL / DataFrame tab lists every query. Open one: the graph is the physical plan drawn bottom to top, each box with its metrics, such as rows out of a Filter, data size of an Exchange, spillSpark SpillWhen a task's sort, join or aggregation no longer fits in execution memory, Spark writes part of it to local disk and reads it back later, so the task finishes, only slower. size of a Sort. The Details link at the bottom gives the text plan, with the Final Plan and the Initial Plan. Our Spark UI guide opens that tab on a real skewed join and a real spill, and says what to read on each box.

A checklist for a slow query

1. Count the Exchange lines. Each one is a shuffle. Is every one needed?
2. Check every join. BroadcastHashJoin with the small table on the build side is cheap. SortMergeJoin shuffles and sorts both sides; SortMergeJoin(skew=true) means Spark found and split a heavy key (data skewShuffle and Data SkewA shuffle moves rows between machines so rows with the same key meet, for a join or group by; skew is when one key has far more rows, so one task does most of the work.).
3. Check the scans. Only the columns you need, your filters in PushedFilters, and PartitionFilters not empty when the table is partitioned on a column you filter.
4. Read the plan that ran. After the run, in the SQL tab or with explain() on the same DataFrame, look for AQEShuffleRead and a join that changed.
5. Then open the stage. The plan says what ran; the stage page says how long each task took. A plan that looks right with 1 slow task is skew, and one where every task spills is too few partitions.