If you’re learning Apache Spark for data engineering, analytics, or production ML pipelines, follow GeekyCodes and subscribe to email updates so you don’t miss upcoming tutorials. In this article, we’ll learn how to use explain() to see exactly what Spark plans to do with your query, how to read every line of a physical plan, and how to spot expensive operations before they ever run on a cluster.

About the examples in this article: every plan shown below is real output, captured from PySpark 4.2 running locally on a synthetic dataset of orders and customers. Only file paths were shortened. Expression IDs such as amount#11 and plan_id values will differ on your machine.

Introduction: Why Explain Plans Matter

In the previous article, we used Spark UI to investigate a job_after_it ran. We found the slow stage, compared task durations, and saw that a sort-merge join was moving a lot of data across the cluster.

But what if you could see that sort-merge join before running the job at all?

That is exactly what explain() gives you. Every DataFrame and Spark SQL query is compiled into a plan before a single task starts. That plan decides:

Which columns are read from storage
Whether filters are pushed down into the file reader
Which partitions of a table are skipped entirely
Which join algorithm is used
Where data will be shuffled across the network
Which operators are fused into generated Java code

When you call explain(), Spark prints that plan. Once you can read it, you can answer questions like “Will this join shuffle both tables?” or “Is my filter actually reducing the data read from Parquet?” in seconds, from your laptop, without waiting 15 minutes for a cluster job to finish.

By the end of this tutorial, you’ll be able to read logical and physical plans line by line, recognize the operators that matter most for performance, and use plans to make evidence-based optimizations — and we’ll measure the payoff with a real benchmark.

1. What Is an Explain Plan?

An explain plan is Spark’s description of how it intends to execute a query.

Because Spark uses lazy evaluation, your transformations don’t run when you write them. Instead, Spark records them and, when an action is triggered, hands them to the Catalyst optimizer. Catalyst turns your code into a series of increasingly concrete plans.

Here is what happens at each phase:

The key distinction to remember:

Logical plans describe what the result should be.
The physical plan describes how Spark will compute it.

Performance problems almost always live in the physical plan, so that’s where you’ll spend most of your time.

Important: explain() doesn’t run your query. It shows the plan Spark has chosen based on what it knows before execution, such as file sizes and configuration. With Adaptive Query Execution enabled, the final plan can change at runtime — we’ll cover this in Section 9.

2. How to Call `explain()`

You can call explain() on any DataFrame. Let’s set up the sample data used throughout this article.

0")
.join(customers, "customer_id")
.groupBy("country")
.agg(F.sum("amount").alias("total_revenue"))
)
result.explain()” data-toggle-ms=”1200″ aria-label=”Copy code” data-nosnippet=”” class=”code-copy-btn absolute right-3 top-3 size-8 p-1.5 flex items-center justify-center bg-black/50 text-white rounded-md transition-colors duration-200 cursor-pointer hover:bg-black/70″>

from pyspark.sql import SparkSession
from pyspark.sql import functions as F
spark = (
    SparkSession.builder
    .appName("ExplainPlanDemo")
    .getOrCreate()
)
orders = spark.read.parquet("/data/orders")        # order_id, customer_id, amount, order_date
customers = spark.read.parquet("/data/customers")  # customer_id, country, segment
result = (
    orders.filter("amount > 0")
    .join(customers, "customer_id")
    .groupBy("country")
    .agg(F.sum("amount").alias("total_revenue"))
)
result.explain()

from pyspark.sql import SparkSession
from pyspark.sql import functions as F
spark = (
    SparkSession.builder
    .appName("ExplainPlanDemo")
    .getOrCreate()
)
orders = spark.read.parquet("/data/orders")        # order_id, customer_id, amount, order_date
customers = spark.read.parquet("/data/customers")  # customer_id, country, segment
result = (
    orders.filter("amount > 0")
    .join(customers, "customer_id")
    .groupBy("country")
    .agg(F.sum("amount").alias("total_revenue"))
)
result.explain()

The five explain modes

Since Spark 3.0, explain() accepts a mode argument.

Explain plans in Spark SQL

The same plans are available from SQL:

EXPLAIN SELECT country, SUM(amount) FROM orders GROUP BY country;
EXPLAIN EXTENDED SELECT ...;
EXPLAIN FORMATTED SELECT ...;
EXPLAIN COST SELECT ...;
EXPLAIN CODEGEN SELECT ...;

EXPLAIN SELECT country, SUM(amount) FROM orders GROUP BY country;
EXPLAIN EXTENDED SELECT ...;
EXPLAIN FORMATTED SELECT ...;
EXPLAIN COST SELECT ...;
EXPLAIN CODEGEN SELECT ...;

From PySpark, you can run these with spark.sql("EXPLAIN FORMATTED ...").show(truncate=False) after registering your DataFrames as temporary views.

Performance tip: Make df.explain() a habit before running any expensive job, especially after writing a new join. It costs nothing and catches many problems before they reach the cluster.

3. Walking Through an Extended Plan

Let’s start small. Here is a filter followed by a projection:

0").select("order_id", "amount")
simple.explain(True)” data-toggle-ms=”1200″ aria-label=”Copy code” data-nosnippet=”” class=”code-copy-btn absolute right-3 top-1/2 -translate-y-1/2 size-8 p-1.5 flex items-center justify-center bg-black/50 text-white rounded-md transition-colors duration-200 cursor-pointer hover:bg-black/70″>

simple = orders.filter("amount > 0").select("order_id", "amount")
simple.explain(True)

simple = orders.filter("amount > 0").select("order_id", "amount")
simple.explain(True)

Real output:

cast(0 as double))
+- Relation [order_id#9L,customer_id#10L,amount#11,order_date#12] parquet
== Analyzed Logical Plan ==
order_id: bigint, amount: double
Project [order_id#9L, amount#11]
+- Filter (amount#11 > cast(0 as double))
+- Relation [order_id#9L,customer_id#10L,amount#11,order_date#12] parquet
== Optimized Logical Plan ==
Project [order_id#9L, amount#11]
+- Filter (isnotnull(amount#11) AND (amount#11 > 0.0))
+- Relation [order_id#9L,customer_id#10L,amount#11,order_date#12] parquet
== Physical Plan ==
*(1) Filter (isnotnull(amount#11) AND (amount#11 > 0.0))
+- *(1) ColumnarToRow
+- FileScan parquet [order_id#9L,amount#11] Batched: true,
DataFilters: [isnotnull(amount#11), (amount#11 > 0.0)], Format: Parquet,
Location: InMemoryFileIndex(1 paths)[file:/data/orders], PartitionFilters: [],
PushedFilters: [IsNotNull(amount), GreaterThan(amount,0.0)],
ReadSchema: struct” data-toggle-ms=”1200″ aria-label=”Copy code” data-nosnippet=”” class=”code-copy-btn absolute right-3 top-3 size-8 p-1.5 flex items-center justify-center bg-black/50 text-white rounded-md transition-colors duration-200 cursor-pointer hover:bg-black/70″>

== Parsed Logical Plan ==
'Project ['order_id, 'amount]
+- Filter (amount#11 > cast(0 as double))
   +- Relation [order_id#9L,customer_id#10L,amount#11,order_date#12] parquet
== Analyzed Logical Plan ==
order_id: bigint, amount: double
Project [order_id#9L, amount#11]
+- Filter (amount#11 > cast(0 as double))
   +- Relation [order_id#9L,customer_id#10L,amount#11,order_date#12] parquet
== Optimized Logical Plan ==
Project [order_id#9L, amount#11]
+- Filter (isnotnull(amount#11) AND (amount#11 > 0.0))
   +- Relation [order_id#9L,customer_id#10L,amount#11,order_date#12] parquet
== Physical Plan ==
*(1) Filter (isnotnull(amount#11) AND (amount#11 > 0.0))
+- *(1) ColumnarToRow
   +- FileScan parquet [order_id#9L,amount#11] Batched: true,
      DataFilters: [isnotnull(amount#11), (amount#11 > 0.0)], Format: Parquet,
      Location: InMemoryFileIndex(1 paths)[file:/data/orders], PartitionFilters: [],
      PushedFilters: [IsNotNull(amount), GreaterThan(amount,0.0)],
      ReadSchema: struct<order_id:bigint,amount:double>

== Parsed Logical Plan ==
'Project ['order_id, 'amount]
+- Filter (amount#11 > cast(0 as double))
   +- Relation [order_id#9L,customer_id#10L,amount#11,order_date#12] parquet
== Analyzed Logical Plan ==
order_id: bigint, amount: double
Project [order_id#9L, amount#11]
+- Filter (amount#11 > cast(0 as double))
   +- Relation [order_id#9L,customer_id#10L,amount#11,order_date#12] parquet
== Optimized Logical Plan ==
Project [order_id#9L, amount#11]
+- Filter (isnotnull(amount#11) AND (amount#11 > 0.0))
   +- Relation [order_id#9L,customer_id#10L,amount#11,order_date#12] parquet
== Physical Plan ==
*(1) Filter (isnotnull(amount#11) AND (amount#11 > 0.0))
+- *(1) ColumnarToRow
   +- FileScan parquet [order_id#9L,amount#11] Batched: true,
      DataFilters: [isnotnull(amount#11), (amount#11 > 0.0)], Format: Parquet,
      Location: InMemoryFileIndex(1 paths)[file:/data/orders], PartitionFilters: [],
      PushedFilters: [IsNotNull(amount), GreaterThan(amount,0.0)],
      ReadSchema: struct<order_id:bigint,amount:double>

(The long FileScan line is wrapped here for readability; Spark prints it on one line.)

Let’s read what changed between phases.

Parsed → Analyzed

In the parsed plan, 'order_id and 'amount have a leading quote: they are unresolved. Spark doesn’t yet know whether these columns exist or what type they are.

In the analyzed plan, the quotes are gone. Each column has a unique ID (order_id#9L, amount#11) and the output schema is listed: order_id: bigint, amount: double. If you misspell a column name, this is the phase where Spark raises an AnalysisException.

Analyzed → Optimized

Two optimizer rules are visible:

Constant folding: cast(0 as double) became the literal 0.0.
Null-filter inference: Spark added isnotnull(amount#11), because amount > 0.0 can never be true for a null value. That extra predicate can be pushed into the file reader.

Optimized → Physical

The physical plan introduces concrete operators:

FileScan parquet [order_id#9L, amount#11] reads only two columns, even though the table has four. This is column pruning, and the ReadSchema confirms it.
PushedFilters shows the filters handed to the Parquet reader, which can use row-group statistics to skip data it knows can’t match.
ColumnarToRow converts Parquet’s column are batches into rows for the operators above it.
*(1) means these operators are fused into codegen stage 1 — Spark generates a single Java function for all of them.

Important: PushedFilters means the filter was offered to the data source. Parquet uses it to skip row groups using min/max statistics, but Spark still keeps the Filter operator above the scan to guarantee correctness. Seeing both is normal.

4. How to Read a Physical Plan

Physical plans look intimidating at first, but they follow a few simple rules.

Rule 1: Read from the bottom up

The plan is printed as a tree with the final operator at the top and the data sources at the bottom. Data flows upward. Start reading at the most indented FileScan lines.

Rule 2: Indentation and branch markers show the tree

+- introduces a child operator.
:- marks the first of two children, typically the left side of a join.
A : running down the left margin connects lines belonging to that first branch.

Rule 3: `*(n)` marks whole-stage code generation

Operators sharing the same *(n) prefix are compiled together into one generated function. Operators without the star, such as Exchange, BroadcastExchange, or a Python UDF evaluator, run outside generated code.

The codegen ID is not the same as the stage ID you see in Spark UI. Codegen IDs describe compiled operator groups; stage boundaries come from shuffles.

Rule 4: Every `Exchange` is a shuffle and a stage boundary

Exchange is the operator to look for first. It means Spark redistributes rows across partitions, usually over the network. Each one splits the job into separate execution stages — exactly the boundaries you saw in the Stages tab of Spark UI.

Rule 5: The # numbers are expression IDs

amount#11 and customer_id#10L are unique IDs for attributes. The L suffix means long. These IDs let you trace a column through the plan, which is useful when two tables both have a customer_id column — here they are customer_id#10L (orders) and customer_id#13L (customers).

Now apply these rules to the real plan for our join query, with AQE and auto-broadcast disabled so the classic shape is visible:

0.0)) AND isnotnull(customer_id#10L))
: +- *(1) ColumnarToRow
: +- FileScan parquet [customer_id#10L,amount#11] … PushedFilters: [IsNotNull(amount), GreaterThan(amount,0.0), IsNotNull(customer_id)] …
+- *(4) Sort [customer_id#13L ASC NULLS FIRST], false, 0
+- Exchange hashpartitioning(customer_id#13L, 8), ENSURE_REQUIREMENTS, [plan_id=126]
+- *(3) Filter isnotnull(customer_id#13L)
+- *(3) ColumnarToRow
+- FileScan parquet [customer_id#13L,country#14] … PushedFilters: [IsNotNull(customer_id)] …” data-toggle-ms=”1200″ aria-label=”Copy code” data-nosnippet=”” class=”code-copy-btn absolute right-3 top-3 size-8 p-1.5 flex items-center justify-center bg-black/50 text-white rounded-md transition-colors duration-200 cursor-pointer hover:bg-black/70″>

== Physical Plan ==
*(6) HashAggregate(keys=[country#14], functions=[sum(amount#11)])
+- Exchange hashpartitioning(country#14, 8), ENSURE_REQUIREMENTS, [plan_id=135]
   +- *(5) HashAggregate(keys=[country#14], functions=[partial_sum(amount#11)])
      +- *(5) Project [amount#11, country#14]
         +- *(5) SortMergeJoin [customer_id#10L], [customer_id#13L], Inner
            :- *(2) Sort [customer_id#10L ASC NULLS FIRST], false, 0
            :  +- Exchange hashpartitioning(customer_id#10L, 8), ENSURE_REQUIREMENTS, [plan_id=117]
            :     +- *(1) Filter ((isnotnull(amount#11) AND (amount#11 > 0.0)) AND isnotnull(customer_id#10L))
            :        +- *(1) ColumnarToRow
            :           +- FileScan parquet [customer_id#10L,amount#11] ... PushedFilters: [IsNotNull(amount), GreaterThan(amount,0.0), IsNotNull(customer_id)] ...
            +- *(4) Sort [customer_id#13L ASC NULLS FIRST], false, 0
               +- Exchange hashpartitioning(customer_id#13L, 8), ENSURE_REQUIREMENTS, [plan_id=126]
                  +- *(3) Filter isnotnull(customer_id#13L)
                     +- *(3) ColumnarToRow
                        +- FileScan parquet [customer_id#13L,country#14] ... PushedFilters: [IsNotNull(customer_id)] ...

== Physical Plan ==
*(6) HashAggregate(keys=[country#14], functions=[sum(amount#11)])
+- Exchange hashpartitioning(country#14, 8), ENSURE_REQUIREMENTS, [plan_id=135]
   +- *(5) HashAggregate(keys=[country#14], functions=[partial_sum(amount#11)])
      +- *(5) Project [amount#11, country#14]
         +- *(5) SortMergeJoin [customer_id#10L], [customer_id#13L], Inner
            :- *(2) Sort [customer_id#10L ASC NULLS FIRST], false, 0
            :  +- Exchange hashpartitioning(customer_id#10L, 8), ENSURE_REQUIREMENTS, [plan_id=117]
            :     +- *(1) Filter ((isnotnull(amount#11) AND (amount#11 > 0.0)) AND isnotnull(customer_id#10L))
            :        +- *(1) ColumnarToRow
            :           +- FileScan parquet [customer_id#10L,amount#11] ... PushedFilters: [IsNotNull(amount), GreaterThan(amount,0.0), IsNotNull(customer_id)] ...
            +- *(4) Sort [customer_id#13L ASC NULLS FIRST], false, 0
               +- Exchange hashpartitioning(customer_id#13L, 8), ENSURE_REQUIREMENTS, [plan_id=126]
                  +- *(3) Filter isnotnull(customer_id#13L)
                     +- *(3) ColumnarToRow
                        +- FileScan parquet [customer_id#13L,country#14] ... PushedFilters: [IsNotNull(customer_id)] ...

Reading it bottom-up:

Two scans. Orders reads only customer_id and amount; customers reads only customer_id and country. Column pruning is working on both sides.
Filters. Spark added isnotnull(customer_id) to both sides, because an inner join can never match null keys.
Two exchanges on customer_id. Both tables are shuffled so rows with the same key land in the same partition.
Two sorts. Each side is sorted by customer_id within its partition.
SortMergeJoin. The sorted streams are merged to produce matching rows.
HashAggregate ... partial_sum. Each partition pre-aggregates its own rows by country. This map-side combine is why the next shuffle is tiny.
A third exchange on country. Partial sums are redistributed so each country lands in one partition.
Final HashAggregate ... sum. Partial sums are combined into the final result.

The 8 in hashpartitioning(customer_id#10L, 8) is the number of shuffle partitions, set here through spark.sql.shuffle.partitions.

What to do: Count the Exchange operators in every plan you write. Each one is network and disk I/O. Three exchanges for a simple revenue-by-country report is a sign worth investigating.

5. Join Strategies in Explain Plans

Joins are where reading explain plans pays off most. The join operator in the plan tells you how much data will move.

Sort-merge join

We already read a SortMergeJoin plan above. Its signature is:

SortMergeJoin
:- Sort
:  +- Exchange hashpartitioning(key)
+- Sort
   +- Exchange hashpartitioning(key)

SortMergeJoin
:- Sort
:  +- Exchange hashpartitioning(key)
+- Sort
   +- Exchange hashpartitioning(key)

Two exchanges plus two sorts. This is the right choice when both sides are large, but expensive when one side could have been broadcast.

Broadcast hash join

Now let’s use a broadcast hint on the small customers table:

0")
.join(broadcast(customers), "customer_id")
.groupBy("country")
.agg(F.sum("amount").alias("total_revenue"))
)
result_bhj.explain()” data-toggle-ms=”1200″ aria-label=”Copy code” data-nosnippet=”” class=”code-copy-btn absolute right-3 top-3 size-8 p-1.5 flex items-center justify-center bg-black/50 text-white rounded-md transition-colors duration-200 cursor-pointer hover:bg-black/70″>

from pyspark.sql.functions import broadcast
result_bhj = (
    orders.filter("amount > 0")
    .join(broadcast(customers), "customer_id")
    .groupBy("country")
    .agg(F.sum("amount").alias("total_revenue"))
)
result_bhj.explain()

from pyspark.sql.functions import broadcast
result_bhj = (
    orders.filter("amount > 0")
    .join(broadcast(customers), "customer_id")
    .groupBy("country")
    .agg(F.sum("amount").alias("total_revenue"))
)
result_bhj.explain()

Real output:

0.0)) AND isnotnull(customer_id#10L))
: +- *(2) ColumnarToRow
: +- FileScan parquet [customer_id#10L,amount#11] …
+- BroadcastExchange HashedRelationBroadcastMode(List(input[0, bigint, false]),false), [plan_id=232]
+- *(1) Filter isnotnull(customer_id#13L)
+- *(1) ColumnarToRow
+- FileScan parquet [customer_id#13L,country#14] …” data-toggle-ms=”1200″ aria-label=”Copy code” data-nosnippet=”” class=”code-copy-btn absolute right-3 top-3 size-8 p-1.5 flex items-center justify-center bg-black/50 text-white rounded-md transition-colors duration-200 cursor-pointer hover:bg-black/70″>

== Physical Plan ==
*(3) HashAggregate(keys=[country#14], functions=[sum(amount#11)])
+- Exchange hashpartitioning(country#14, 8), ENSURE_REQUIREMENTS, [plan_id=238]
   +- *(2) HashAggregate(keys=[country#14], functions=[partial_sum(amount#11)])
      +- *(2) Project [amount#11, country#14]
         +- *(2) BroadcastHashJoin [customer_id#10L], [customer_id#13L], Inner, BuildRight, false, false
            :- *(2) Filter ((isnotnull(amount#11) AND (amount#11 > 0.0)) AND isnotnull(customer_id#10L))
            :  +- *(2) ColumnarToRow
            :     +- FileScan parquet [customer_id#10L,amount#11] ...
            +- BroadcastExchange HashedRelationBroadcastMode(List(input[0, bigint, false]),false), [plan_id=232]
               +- *(1) Filter isnotnull(customer_id#13L)
                  +- *(1) ColumnarToRow
                     +- FileScan parquet [customer_id#13L,country#14] ...

== Physical Plan ==
*(3) HashAggregate(keys=[country#14], functions=[sum(amount#11)])
+- Exchange hashpartitioning(country#14, 8), ENSURE_REQUIREMENTS, [plan_id=238]
   +- *(2) HashAggregate(keys=[country#14], functions=[partial_sum(amount#11)])
      +- *(2) Project [amount#11, country#14]
         +- *(2) BroadcastHashJoin [customer_id#10L], [customer_id#13L], Inner, BuildRight, false, false
            :- *(2) Filter ((isnotnull(amount#11) AND (amount#11 > 0.0)) AND isnotnull(customer_id#10L))
            :  +- *(2) ColumnarToRow
            :     +- FileScan parquet [customer_id#10L,amount#11] ...
            +- BroadcastExchange HashedRelationBroadcastMode(List(input[0, bigint, false]),false), [plan_id=232]
               +- *(1) Filter isnotnull(customer_id#13L)
                  +- *(1) ColumnarToRow
                     +- FileScan parquet [customer_id#13L,country#14] ...

Compare it with the sort-merge plan:

BuildRight tells you which side becomes the in-memory hash table — here the right side, customers. Notice that the scan, filter, join, and partial aggregate of orders now all share *(2). Spark fuses them into one generated function and streams the large table straight through the join without a shuffle.

How Spark chooses a join strategy

The most important setting here is spark.sql.autoBroadcastJoinThreshold, which defaults to 10 MB. If Spark’s estimate of one side’s size is below this threshold, it broadcasts automatically — no hint needed. In our run with default settings, Spark chose BroadcastHashJoin on its own, because the customers Parquet files were only about 200 KB.

That word “estimate” matters. For a filtered or aggregated DataFrame, Spark’s size estimate can be far from reality, which is why it sometimes misses a broadcast opportunity or, less often, broadcasts something too large. Section 10 shows how to see those estimates.

Performance tip: Look for BroadcastNestedLoopJoin or CartesianProduct in any plan that touches large tables. They usually mean a join condition is missing, uses an inequality, or wraps the key in an expression Spark can’t use as an equi-join key.

7. The Formatted Mode: Easier Reading for Long Plans

Plans for real pipelines can run to hundreds of lines. The formatted mode separates the tree’s shape from the details.

result.explain(mode="formatted")

result.explain(mode="formatted")

Real output (first part):

== Physical Plan ==
* HashAggregate (15)
+- Exchange (14)
   +- * HashAggregate (13)
      +- * Project (12)
         +- * SortMergeJoin Inner (11)
            :- * Sort (5)
            :  +- Exchange (4)
            :     +- * Filter (3)
            :        +- * ColumnarToRow (2)
            :           +- Scan parquet  (1)
            +- * Sort (10)
               +- Exchange (9)
                  +- * Filter (8)
                     +- * ColumnarToRow (7)
                        +- Scan parquet  (6)

== Physical Plan ==
* HashAggregate (15)
+- Exchange (14)
   +- * HashAggregate (13)
      +- * Project (12)
         +- * SortMergeJoin Inner (11)
            :- * Sort (5)
            :  +- Exchange (4)
            :     +- * Filter (3)
            :        +- * ColumnarToRow (2)
            :           +- Scan parquet  (1)
            +- * Sort (10)
               +- Exchange (9)
                  +- * Filter (8)
                     +- * ColumnarToRow (7)
                        +- Scan parquet  (6)

Then each node is described by number:

(1) Scan parquet
Output [2]: [customer_id#10L, amount#11]
Batched: true
Location: InMemoryFileIndex [file:/data/orders]
PushedFilters: [IsNotNull(amount), GreaterThan(amount,0.0), IsNotNull(customer_id)]
ReadSchema: struct<customer_id:bigint,amount:double>
(4) Exchange
Input [2]: [customer_id#10L, amount#11]
Arguments: hashpartitioning(customer_id#10L, 8), ENSURE_REQUIREMENTS, [plan_id=117]
(11) SortMergeJoin [codegen id : 5]
Left keys [1]: [customer_id#10L]
Right keys [1]: [customer_id#13L]
Join type: Inner
Join condition: None
(13) HashAggregate [codegen id : 5]
Input [2]: [amount#11, country#14]
Keys [1]: [country#14]
Functions [1]: [partial_sum(amount#11)]
Aggregate Attributes [1]: [sum#26]
Results [2]: [country#14, sum#27]

(1) Scan parquet
Output [2]: [customer_id#10L, amount#11]
Batched: true
Location: InMemoryFileIndex [file:/data/orders]
PushedFilters: [IsNotNull(amount), GreaterThan(amount,0.0), IsNotNull(customer_id)]
ReadSchema: struct<customer_id:bigint,amount:double>
(4) Exchange
Input [2]: [customer_id#10L, amount#11]
Arguments: hashpartitioning(customer_id#10L, 8), ENSURE_REQUIREMENTS, [plan_id=117]
(11) SortMergeJoin [codegen id : 5]
Left keys [1]: [customer_id#10L]
Right keys [1]: [customer_id#13L]
Join type: Inner
Join condition: None
(13) HashAggregate [codegen id : 5]
Input [2]: [amount#11, country#14]
Keys [1]: [country#14]
Functions [1]: [partial_sum(amount#11)]
Aggregate Attributes [1]: [sum#26]
Results [2]: [country#14, sum#27]

The numbers in the tree, such as Exchange (4), match the detail sections. In practice:

Scan the compact tree for Exchange, join operators, and anything unusual.
Jump to that node’s number for its inputs, keys, and arguments.

Join condition: None here means there’s no extra non-equi condition beyond the keys — not that the join has no condition.

Performance tip: The formatted node numbers also make plans easier to discuss in code reviews: “Exchange (4) disappears if we broadcast customers” is far clearer than pasting a 40-line plan.

8. Spotting the Optimizer at Work: Pushdown and Pruning

Some of the biggest wins don’t change the join at all. They change how much data is read in the first place. The plan tells you whether these optimizations are actually happening.

8.1 Column pruning: check `ReadSchema`

FileScan parquet [order_id#9L,amount#11] ... ReadSchema: struct<order_id:bigint,amount:double>

FileScan parquet [order_id#9L,amount#11] ... ReadSchema: struct<order_id:bigint,amount:double>

Parquet is columnar, so Spark reads only the columns listed in ReadSchema. If you see every column of a wide table here, look for a select("*"), a UDF that takes the whole row, or a cached DataFrame that included all columns.

8.2 Predicate pushdown: check `PushedFilters`

PushedFilters: [IsNotNull(amount), GreaterThan(amount,100.0)]

PushedFilters: [IsNotNull(amount), GreaterThan(amount,100.0)]

Pushed filters let the Parquet reader skip row groups using their min/max statistics. How much that helps depends on how the data is laid out: values sorted or clustered by the filter column skip far more than randomly distributed ones.

8.3 Partition pruning: check `PartitionFilters`

If a table is written with partitionBy("month"), each month is a separate directory. Filtering on the partition column lets Spark skip whole directories.

orders_by_month = spark.read.parquet("/data/orders_by_month")
orders_by_month.filter("month = 3").select("id", "amount").explain()

orders_by_month = spark.read.parquet("/data/orders_by_month")
orders_by_month.filter("month = 3").select("id", "amount").explain()

Real output:

== Physical Plan ==
*(1) Project [id#262L, amount#264]
+- *(1) ColumnarToRow
   +- FileScan parquet [id#262L,amount#264,month#266] Batched: true, DataFilters: [],
      Format: Parquet, Location: InMemoryFileIndex(1 paths)[file:/data/orders_by_month],
      PartitionFilters: [isnotnull(month#266), (month#266 = 3)], PushedFilters: [],
      ReadSchema: struct<id:bigint,amount:double>

== Physical Plan ==
*(1) Project [id#262L, amount#264]
+- *(1) ColumnarToRow
   +- FileScan parquet [id#262L,amount#264,month#266] Batched: true, DataFilters: [],
      Format: Parquet, Location: InMemoryFileIndex(1 paths)[file:/data/orders_by_month],
      PartitionFilters: [isnotnull(month#266), (month#266 = 3)], PushedFilters: [],
      ReadSchema: struct<id:bigint,amount:double>

The filter now appears under PartitionFilters, not PushedFilters, and there’s no separate Filter operator at all — Spark simply never opens the other eleven months.

8.4 How a Python UDF breaks pushdown

Here is a filter written as a Python UDF:

100, BooleanType())
orders.filter(is_big("amount")).select("order_id").explain()” data-toggle-ms=”1200″ aria-label=”Copy code” data-nosnippet=”” class=”code-copy-btn absolute right-3 top-1/2 -translate-y-1/2 size-8 p-1.5 flex items-center justify-center bg-black/50 text-white rounded-md transition-colors duration-200 cursor-pointer hover:bg-black/70″>

from pyspark.sql.types import BooleanType
is_big = F.udf(lambda a: a is not None and a > 100, BooleanType())
orders.filter(is_big("amount")).select("order_id").explain()

from pyspark.sql.types import BooleanType
is_big = F.udf(lambda a: a is not None and a > 100, BooleanType())
orders.filter(is_big("amount")).select("order_id").explain()

Real output:

== Physical Plan ==
*(2) Project [order_id#9L]
+- *(2) Filter pythonUDF0#54: boolean
   +- BatchEvalPython [<lambda>(amount#11)#53], [pythonUDF0#54]
      +- *(1) ColumnarToRow
         +- FileScan parquet [order_id#9L,amount#11] ... DataFilters: [], ...
            PushedFilters: [], ...

== Physical Plan ==
*(2) Project [order_id#9L]
+- *(2) Filter pythonUDF0#54: boolean
   +- BatchEvalPython [<lambda>(amount#11)#53], [pythonUDF0#54]
      +- *(1) ColumnarToRow
         +- FileScan parquet [order_id#9L,amount#11] ... DataFilters: [], ...
            PushedFilters: [], ...

And the same logic written as a native expression:

100).select("order_id").explain()
== Physical Plan ==
*(1) Project [order_id#9L]
+- *(1) Filter (isnotnull(amount#11) AND (amount#11 > 100.0))
+- *(1) ColumnarToRow
+- FileScan parquet [order_id#9L,amount#11] …
PushedFilters: [IsNotNull(amount), GreaterThan(amount,100.0)], …” data-toggle-ms=”1200″ aria-label=”Copy code” data-nosnippet=”” class=”code-copy-btn absolute right-3 top-3 size-8 p-1.5 flex items-center justify-center bg-black/50 text-white rounded-md transition-colors duration-200 cursor-pointer hover:bg-black/70″>

orders.filter(F.col("amount") > 100).select("order_id").explain()
== Physical Plan ==
*(1) Project [order_id#9L]
+- *(1) Filter (isnotnull(amount#11) AND (amount#11 > 100.0))
   +- *(1) ColumnarToRow
      +- FileScan parquet [order_id#9L,amount#11] ...
         PushedFilters: [IsNotNull(amount), GreaterThan(amount,100.0)], ...

orders.filter(F.col("amount") > 100).select("order_id").explain()
== Physical Plan ==
*(1) Project [order_id#9L]
+- *(1) Filter (isnotnull(amount#11) AND (amount#11 > 100.0))
   +- *(1) ColumnarToRow
      +- FileScan parquet [order_id#9L,amount#11] ...
         PushedFilters: [IsNotNull(amount), GreaterThan(amount,100.0)], ...

Three differences are visible:

PushedFilters: [] — Catalyst can’t see inside a Python function, so nothing can be pushed to Parquet.
BatchEvalPython — every row is serialized and sent to a Python worker process and back.
Two codegen stages instead of one — the UDF breaks the generated code pipeline.

We’ll measure what this costs in Section 11.

8.5 Confirming that a cache is used

0").cache()
cached.count()
cached.groupBy("customer_id").count().explain()
== Physical Plan ==
*(2) HashAggregate(keys=[customer_id#10L], functions=[count(1)])
+- Exchange hashpartitioning(customer_id#10L, 8), ENSURE_REQUIREMENTS, [plan_id=563]
+- *(1) HashAggregate(keys=[customer_id#10L], functions=[partial_count(1)])
+- InMemoryTableScan [customer_id#10L]
+- InMemoryRelation [order_id#9L, customer_id#10L, amount#11, order_date#12], StorageLevel(disk, memory, deserialized, 1 replicas)
+- *(1) Filter (isnotnull(amount#11) AND (amount#11 > 0.0))
…” data-toggle-ms=”1200″ aria-label=”Copy code” data-nosnippet=”” class=”code-copy-btn absolute right-3 top-3 size-8 p-1.5 flex items-center justify-center bg-black/50 text-white rounded-md transition-colors duration-200 cursor-pointer hover:bg-black/70″>

cached = orders.filter("amount > 0").cache()
cached.count()
cached.groupBy("customer_id").count().explain()
== Physical Plan ==
*(2) HashAggregate(keys=[customer_id#10L], functions=[count(1)])
+- Exchange hashpartitioning(customer_id#10L, 8), ENSURE_REQUIREMENTS, [plan_id=563]
   +- *(1) HashAggregate(keys=[customer_id#10L], functions=[partial_count(1)])
      +- InMemoryTableScan [customer_id#10L]
            +- InMemoryRelation [order_id#9L, customer_id#10L, amount#11, order_date#12], StorageLevel(disk, memory, deserialized, 1 replicas)
                  +- *(1) Filter (isnotnull(amount#11) AND (amount#11 > 0.0))
                     ...

cached = orders.filter("amount > 0").cache()
cached.count()
cached.groupBy("customer_id").count().explain()
== Physical Plan ==
*(2) HashAggregate(keys=[customer_id#10L], functions=[count(1)])
+- Exchange hashpartitioning(customer_id#10L, 8), ENSURE_REQUIREMENTS, [plan_id=563]
   +- *(1) HashAggregate(keys=[customer_id#10L], functions=[partial_count(1)])
      +- InMemoryTableScan [customer_id#10L]
            +- InMemoryRelation [order_id#9L, customer_id#10L, amount#11, order_date#12], StorageLevel(disk, memory, deserialized, 1 replicas)
                  +- *(1) Filter (isnotnull(amount#11) AND (amount#11 > 0.0))
                     ...

InMemoryTableScan over an InMemoryRelation confirms Spark will read from the cache rather than from Parquet. If you expected caching and don’t see these operators, the DataFrame you’re querying isn’t the same one you cached — a common bug when a transformation is applied before calling .cache() on a different variable.

9. Explain Plans and Adaptive Query Execution (AQE)

AQE is enabled by default since Spark 3.2. When it’s on, the physical plan is wrapped in AdaptiveSparkPlan:

== Physical Plan ==
AdaptiveSparkPlan isFinalPlan=false
+- HashAggregate(keys=[country#14], functions=[sum(amount#11)])
   +- Exchange hashpartitioning(country#14, 8), ENSURE_REQUIREMENTS, [plan_id=304]
      +- HashAggregate(keys=[country#14], functions=[partial_sum(amount#11)])
         +- Project [amount#11, country#14]
            +- BroadcastHashJoin [customer_id#10L], [customer_id#13L], Inner, BuildRight, false, false
               ...

== Physical Plan ==
AdaptiveSparkPlan isFinalPlan=false
+- HashAggregate(keys=[country#14], functions=[sum(amount#11)])
   +- Exchange hashpartitioning(country#14, 8), ENSURE_REQUIREMENTS, [plan_id=304]
      +- HashAggregate(keys=[country#14], functions=[partial_sum(amount#11)])
         +- Project [amount#11, country#14]
            +- BroadcastHashJoin [customer_id#10L], [customer_id#13L], Inner, BuildRight, false, false
               ...

isFinalPlan=false means this is the initial plan. AQE may change it once real shuffle statistics are available — for example, by converting a sort-merge join into a broadcast join, coalescing small shuffle partitions, or splitting skewed partitions.

After running an action on the same DataFrame, calling explain() again shows both versions:

result.collect()
result.explain()
== Physical Plan ==
AdaptiveSparkPlan isFinalPlan=true
+- == Final Plan ==
   ResultQueryStage 2
   +- *(3) HashAggregate(keys=[country#14], functions=[sum(amount#11)])
      +- AQEShuffleRead coalesced
         +- ShuffleQueryStage 1
            +- Exchange hashpartitioning(country#14, 8), ENSURE_REQUIREMENTS, [plan_id=386]
               +- *(2) HashAggregate(keys=[country#14], functions=[partial_sum(amount#11)])
                  ...
                        +- BroadcastQueryStage 0
                           +- BroadcastExchange HashedRelationBroadcastMode(...)
                              ...
+- == Initial Plan ==
   HashAggregate(keys=[country#14], functions=[sum(amount#11)])
   +- Exchange hashpartitioning(country#14, 8), ...
   ...

result.collect()
result.explain()
== Physical Plan ==
AdaptiveSparkPlan isFinalPlan=true
+- == Final Plan ==
   ResultQueryStage 2
   +- *(3) HashAggregate(keys=[country#14], functions=[sum(amount#11)])
      +- AQEShuffleRead coalesced
         +- ShuffleQueryStage 1
            +- Exchange hashpartitioning(country#14, 8), ENSURE_REQUIREMENTS, [plan_id=386]
               +- *(2) HashAggregate(keys=[country#14], functions=[partial_sum(amount#11)])
                  ...
                        +- BroadcastQueryStage 0
                           +- BroadcastExchange HashedRelationBroadcastMode(...)
                              ...
+- == Initial Plan ==
   HashAggregate(keys=[country#14], functions=[sum(amount#11)])
   +- Exchange hashpartitioning(country#14, 8), ...
   ...

What AQE adds to the plan:

Here, only 6 countries exist, so 8 shuffle partitions would mostly be empty. AQEShuffleRead coalesced shows AQE combined them.

Important: Before execution, explain() on an AQE query shows only the initial plan. To see what really ran, either call explain() after an action on the same DataFrame, or open the SQL/DataFrame tab in Spark UI, which shows the final adaptive plan with runtime metrics.

10. The Cost Mode: Seeing Spark’s Size Estimates

Join selection depends on estimated sizes. The cost mode exposes those estimates.

result_bhj.explain(mode="cost")

result_bhj.explain(mode="cost")

Real output (optimized logical plan section):

== Optimized Logical Plan ==
Aggregate [country#14], [country#14, sum(amount#11) AS total_revenue#29], Statistics(sizeInBytes=1837.4 GiB)
+- Project [amount#11, country#14], Statistics(sizeInBytes=1837.4 GiB)
   +- Join Inner, (customer_id#10L = customer_id#13L), rightHint=(strategy=broadcast), Statistics(sizeInBytes=2.6 TiB)
      :- Project [customer_id#10L, amount#11], Statistics(sizeInBytes=20.5 MiB)
      :  +- Filter (...), Statistics(sizeInBytes=30.8 MiB)
      :     +- Relation [order_id#9L,customer_id#10L,amount#11,order_date#12] parquet, Statistics(sizeInBytes=30.8 MiB)
      +- Project [customer_id#13L, country#14], Statistics(sizeInBytes=129.2 KiB)
         +- Filter isnotnull(customer_id#13L), Statistics(sizeInBytes=201.0 KiB)
            +- Relation [customer_id#13L,country#14,segment#15] parquet, Statistics(sizeInBytes=201.0 KiB)

== Optimized Logical Plan ==
Aggregate [country#14], [country#14, sum(amount#11) AS total_revenue#29], Statistics(sizeInBytes=1837.4 GiB)
+- Project [amount#11, country#14], Statistics(sizeInBytes=1837.4 GiB)
   +- Join Inner, (customer_id#10L = customer_id#13L), rightHint=(strategy=broadcast), Statistics(sizeInBytes=2.6 TiB)
      :- Project [customer_id#10L, amount#11], Statistics(sizeInBytes=20.5 MiB)
      :  +- Filter (...), Statistics(sizeInBytes=30.8 MiB)
      :     +- Relation [order_id#9L,customer_id#10L,amount#11,order_date#12] parquet, Statistics(sizeInBytes=30.8 MiB)
      +- Project [customer_id#13L, country#14], Statistics(sizeInBytes=129.2 KiB)
         +- Filter isnotnull(customer_id#13L), Statistics(sizeInBytes=201.0 KiB)
            +- Relation [customer_id#13L,country#14,segment#15] parquet, Statistics(sizeInBytes=201.0 KiB)

Look at the join: Spark estimates 2.6 TiB, while the actual inputs are about 30 MB and 200 KB, and the real result is six rows.

This happens because, without column-level statistics, Spark has no idea how many rows will match, so its join estimate is extremely pessimistic. Notice also that the Filter didn’t reduce the estimate at all (30.8 MiB before and after) — without statistics, Spark can’t know the filter’s selectivity.

Why this matters in practice:

A broadcast decision downstream of this join would be based on a wildly wrong number.
Spark may refuse to broadcast a genuinely small aggregated result because the estimate looks huge.

How to improve estimates for catalog tables:

ANALYZE TABLE orders COMPUTE STATISTICS FOR COLUMNS customer_id, amount;
ANALYZE TABLE customers COMPUTE STATISTICS FOR COLUMNS customer_id, country;

ANALYZE TABLE orders COMPUTE STATISTICS FOR COLUMNS customer_id, amount;
ANALYZE TABLE customers COMPUTE STATISTICS FOR COLUMNS customer_id, country;

Combined with spark.sql.cbo.enabled=true, column statistics let Spark estimate row counts and selectivity. These commands work on tables registered in a catalog, not on DataFrames read directly from a path.

Performance tip: When Spark won’t broadcast a table you know is small, run explain(mode="cost") and check its sizeInBytes. Often the fix is an explicit broadcast() hint — or relying on AQE, which uses actual runtime sizes instead of estimates.

11. Does Reading the Plan Pay Off? A Real Benchmark

Plans make predictions: fewer exchanges, fewer rows read, no Python round-trip. Let’s check those predictions against a stopwatch.

Setup: 12 million orders and 200,000 customers, Parquet, PySpark 4.2, local[4], AQE disabled so each plan runs exactly as written. Each query was written to Spark’s noop sink to force full execution, and each timing is the median of three runs. Shuffle bytes and rows read came from Spark’s monitoring REST API.

Shuffle: the sort-merge join wrote 153.9 MB of shuffle data. The broadcast join wrote effectively none — only the handful of partial country sums.
Rows read: without partitioning, Spark read all 12 million rows to find month 3. With partition pruning, it read 1 million.

Important: Absolute numbers depend on hardware, data size, and cluster layout. On a real cluster, shuffle crosses the network, so the gap for join strategy is often larger than on a single laptop. The ratios — and the fact that the plan predicted every one of them — are the lesson.

12. A Practical Workflow: Use the Plan Before and After Running

Explain plans and Spark UI work best together. Here is a repeatable workflow.

Step 1: Explain before you run

Write the query and call explain() (or explain(mode="formatted") for long plans). Don’t trigger an action yet.

Step 2: Check the scans

For every FileScan:

Does ReadSchema include only the columns you need?
Are your filters in PushedFilters or PartitionFilters?
Is there a filter that should have been pushed but wasn’t? Look for UDFs, casts on the column, or complex expressions.

Step 3: Count and justify the exchanges

For each Exchange, ask: is this shuffle required by the logic (a join or groupBy on a new key), or is it avoidable (a small table that could be broadcast, a repartition() that isn’t needed)?

Step 4: Check the join operators

SortMergeJoin with one small side → test broadcast().
BroadcastNestedLoopJoin or CartesianProduct → check the join condition.
BroadcastHashJoin on a side that might grow → confirm it will still fit in memory as data grows.

Step 5: Run, then compare with Spark UI

Execute the job. In the SQL/DataFrame tab, open the query and compare the final plan with what you expected. Check runtime metrics on each operator: rows output, shuffle sizes, spill, and time.

Step 6: Change one thing and re-explain

Apply one change, confirm the plan changed the way you intended, then measure.

The rule: Explain → predict → change → re-explain → measure.

13. Red Flags to Look For in Any Plan

Example: a window function without `partitionBy`

from pyspark.sql.window import Window
w = Window.partitionBy("customer_id").orderBy(F.desc("amount"))
orders.withColumn("rk", F.row_number().over(w)).filter("rk = 1").explain()
== Physical Plan ==
*(3) Filter (rk#55 = 1)
+- Window [row_number() windowspecdefinition(customer_id#10L, amount#11 DESC NULLS LAST, ...) AS rk#55], [customer_id#10L], [amount#11 DESC NULLS LAST]
   +- WindowGroupLimit [customer_id#10L], [amount#11 DESC NULLS LAST], row_number(), 1, Final
      +- *(2) Sort [customer_id#10L ASC NULLS FIRST, amount#11 DESC NULLS LAST], false, 0
         +- Exchange hashpartitioning(customer_id#10L, 8), ENSURE_REQUIREMENTS, [plan_id=491]
            +- WindowGroupLimit [customer_id#10L], [amount#11 DESC NULLS LAST], row_number(), 1, Partial
               +- *(1) Sort [customer_id#10L ASC NULLS FIRST, amount#11 DESC NULLS LAST], false, 0
                  +- *(1) ColumnarToRow
                     +- FileScan parquet [order_id#9L,customer_id#10L,amount#11,order_date#12] ...

from pyspark.sql.window import Window
w = Window.partitionBy("customer_id").orderBy(F.desc("amount"))
orders.withColumn("rk", F.row_number().over(w)).filter("rk = 1").explain()
== Physical Plan ==
*(3) Filter (rk#55 = 1)
+- Window [row_number() windowspecdefinition(customer_id#10L, amount#11 DESC NULLS LAST, ...) AS rk#55], [customer_id#10L], [amount#11 DESC NULLS LAST]
   +- WindowGroupLimit [customer_id#10L], [amount#11 DESC NULLS LAST], row_number(), 1, Final
      +- *(2) Sort [customer_id#10L ASC NULLS FIRST, amount#11 DESC NULLS LAST], false, 0
         +- Exchange hashpartitioning(customer_id#10L, 8), ENSURE_REQUIREMENTS, [plan_id=491]
            +- WindowGroupLimit [customer_id#10L], [amount#11 DESC NULLS LAST], row_number(), 1, Partial
               +- *(1) Sort [customer_id#10L ASC NULLS FIRST, amount#11 DESC NULLS LAST], false, 0
                  +- *(1) ColumnarToRow
                     +- FileScan parquet [order_id#9L,customer_id#10L,amount#11,order_date#12] ...

Two useful observations:

The exchange uses hashpartitioning(customer_id, 8), so the work is spread across partitions. If you remove partitionBy, it becomes Exchange SinglePartition and one task must process the entire table.
WindowGroupLimit ... Partial is an optimization in recent Spark versions: because we keep only rk = 1, Spark discards non-top rows before the shuffle, reducing shuffle volume.

14. Common Mistakes When Reading Explain Plans

Mistake 1: Reading top-down

The top of the plan is the_last_operation. Start at the scans and follow the data upward, or you’ll misunderstand which filter applies to which input.

Mistake 2: Treating every `Exchange` as a bug

A groupBy on a new key or a join between two large tables genuinely needs a shuffle. The goal is to remove unnecessary exchanges, not all of them.

Mistake 3: Trusting the pre-execution plan with AQE enabled

With isFinalPlan=false, you’re looking at the initial plan. The join strategy and partition counts may change at runtime. Confirm the final plan in Spark UI or by re-running explain() after an action.

Mistake 4: Confusing `PushedFilters` with skipped data

PushedFilters means the filter was offered to the data source. How much data it actually skips depends on file statistics and data layout. Spark UI’s scan metrics tell you what was really read.

Mistake 5: Confusing codegen IDs with stage IDs

*(5) is a whole stage codegen group, not Stage 5 in Spark UI. Use Exchange positions to reason about stages.

Mistake 6: Forcing broadcast hints everywhere

A broadcast hint overrides Spark’s size check. Broadcasting a table that turns out to be large can cause driver or executor out-of-memory errors. Hint only tables you know are small — and will stay small.

Mistake 7: Ignoring the plan after a code change

A small change — adding a UDF, a cast, or a select("*") — can silently remove pushdown or pruning. Re-run explain () whenever a performance-sensitive query changes.

15. A Reusable Explain Plan Checklist

Use this checklist before running any expensive Spark query.

Call df.explain() or df.explain(mode="formatted") before triggering an action.
Read the plan from the bottom up, starting with each FileScan.
Confirm ReadSchema includes only the columns you need.
Confirm filters appear in PushedFilters or PartitionFilters.
Count the Exchange operators and justify each one.
Check the join operator: broadcast where one side is small, no unexpected nested-loop joins.
Look for BatchEvalPython or ArrowEvalPython and consider native replacements.
If caching, confirm InMemoryTableScan appears.
If AQE is enabled, compare the initial and final plans after execution.
If a join choice looks wrong, inspect estimates with explain(mode="cost").
Make one change, re-explain, and measure the result in Spark UI.

Keep this checklist handy when writing and reviewing PySpark code.

16. Explain Plan Interview Questions

Q1. What does explain () do in Spark?

It prints the plan Spark will use to execute a DataFrame or SQL query without running it. Depending on the mode, it shows logical plans, the physical plan, statistics, or generated code.

Q2. What are the explain modes available in Spark 3 and later?

simple (physical plan only, the default), extended (all logical plans plus the physical plan), formatted (tree overview plus node details), cost (logical plan with statistics), and codegen (generated Java code).

Q3. What is the difference between a logical and a physical plan?

A logical plan describes what to compute — filters, projections, joins — independent of execution. A physical plan describes how — specific algorithms such as SortMergeJoin or BroadcastHashJoin, and data movement such as Exchange.

Q4. In which direction should you read a physical plan?

Bottom up. The leaves are data sources such as FileScan; data flows upward to the final operator at the top.

Q5. What does `Exchange` mean in a physical plan?

It means data is redistributed across partitions, usually through a shuffle. Each Exchange creates a stage boundary and typically involves disk and network I/O.

Q6. What does the `*(n)` prefix mean?

It marks operators fused by whole-stage code generation into a single generated Java function. Operators with the same number run in the same generated code.

Q7. How can you tell whether predicate pushdown is working?

Check the FileScan node. Filters pushed to the data source appear under PushedFilters; filters on partition columns appear under PartitionFilters.

Q8. How do you identify a broadcast join in a plan?

Look for BroadcastHashJoin with a BroadcastExchange beneath one side. The BuildLeft or BuildRight argument shows which side is broadcast.

Q9. Why might Spark choose a sort-merge join for a small table?

Because its size estimate exceeds spark.sql.autoBroadcastJoinThreshold — often due to missing statistics, or because the table is the result of transformations Spark can’t estimate well. A broadcast () hint, better statistics, or AQE can fix it.

Q10. What does `AdaptiveSparkPlan isFinalPlan=false` mean?

The plan is the initial, pre-execution plan, and AQE may re-optimize it at runtime using actual shuffle statistics. After execution, isFinalPlan=true shows the plan that actually ran.

17. Best Practices for Working with Explain Plans

Explain early: Check plans while developing, not after a production job is slow.
Use formatted mode for long plans: The numbered nodes make complex plans manageable.
Pair plans with Spark UI: The plan shows intent; Spark UI shows what actually happened.
Prefer native functions over Python UDFs: They keep pushdown, pruning, and code generation intact.
Design for pruning: Partition large tables by columns you frequently filter on, with sensible cardinality.
Keep statistics current: For catalog tables used in joins, compute column statistics after significant data changes.
Save plans for critical queries: Store plan output alongside code reviews, so plan regressions are easy to spot.

For more information, consult the official documentation:

Spark SQL Performance Tuning
[DataFrame.explain](https://spark.apache.org/docs/latest/api/python/reference/pyspark.sql/api/pyspark.sql.DataFrame.explain.html) API reference
Spark SQL EXPLAIN statement
Apache Spark Web UI
Spark Configuration

Conclusion

Explain plans turn Spark from a black box into something you can reason about before you spend cluster time.

With one call to explain(), you can answer the questions that matter most: Which columns are read? Are filters pushed into the scan? Are partitions skipped? How many shuffles will happen? Which join algorithm did Spark choose — and why?

Our benchmark showed that these aren’t academic details. Removing two exchanges made a join 3.1× faster. Replacing a Python UDF with a native expression made a filter 23.6× faster. Partition pruning cut rows read from 12 million to 1 million. In every case, the plan predicted the result before a single task ran.

The most effective workflow is simple:

Explain → Predict → Change → Re-explain → Measure.

Read the plan bottom up, count the exchanges, check the scans and joins, and confirm the outcome in Spark UI. That habit will catch more performance problems than any amount of configuration tuning.

Continue Learning Apache Spark

Next article: #33 — Memory Management in Apache Spark: How Executors Use Memory and Why Jobs Fail

In the next tutorial, we’ll explore how Spark divides executor memory between execution and storage, what really happens during spill, why out-of-memory errors occur, and which settings actually help.

Explore the complete Apache Spark Tutorials learning path on GeekyCodes for more tutorials covering Spark fundamentals, PySpark, execution, and performance tuning.

Related Spark Concepts

If this tutorial helped you understand Spark query plans, follow GeekyCodes and subscribe to email updates for the next article in the series.

Apache Spark Learning Path — Progress Tracker

Introduction: Why Explain Plans Matter

1. What Is an Explain Plan?

2. How to Call explain()

The five explain modes

Explain plans in Spark SQL

3. Walking Through an Extended Plan

Parsed → Analyzed

Analyzed → Optimized

Optimized → Physical

4. How to Read a Physical Plan

Rule 2: Indentation and branch markers show the tree

Rule 3: *(n) marks whole-stage code generation

Rule 4: Every Exchange is a shuffle and a stage boundary

Rule 5: The # numbers are expression IDs

5. Join Strategies in Explain Plans

Sort-merge join

Broadcast hash join

How Spark chooses a join strategy

7. The Formatted Mode: Easier Reading for Long Plans

8. Spotting the Optimizer at Work: Pushdown and Pruning

8.1 Column pruning: check ReadSchema

8.2 Predicate pushdown: check PushedFilters

8.3 Partition pruning: check PartitionFilters

8.4 How a Python UDF breaks pushdown

8.5 Confirming that a cache is used

9. Explain Plans and Adaptive Query Execution (AQE)

10. The Cost Mode: Seeing Spark’s Size Estimates

11. Does Reading the Plan Pay Off? A Real Benchmark

12. A Practical Workflow: Use the Plan Before and After Running

Step 1: Explain before you run

Step 2: Check the scans

Step 3: Count and justify the exchanges

Step 4: Check the join operators

Step 5: Run, then compare with Spark UI

Step 6: Change one thing and re-explain

13. Red Flags to Look For in Any Plan

Example: a window function without partitionBy

14. Common Mistakes When Reading Explain Plans

Mistake 1: Reading top-down

Mistake 2: Treating every Exchange as a bug

Mistake 3: Trusting the pre-execution plan with AQE enabled

Mistake 4: Confusing PushedFilters with skipped data

Mistake 5: Confusing codegen IDs with stage IDs

Mistake 6: Forcing broadcast hints everywhere

Mistake 7: Ignoring the plan after a code change

15. A Reusable Explain Plan Checklist

16. Explain Plan Interview Questions

Q1. What does explain () do in Spark?

Q2. What are the explain modes available in Spark 3 and later?

Q3. What is the difference between a logical and a physical plan?

Q4. In which direction should you read a physical plan?

Q5. What does Exchange mean in a physical plan?

Q6. What does the *(n) prefix mean?

Q7. How can you tell whether predicate pushdown is working?

Q8. How do you identify a broadcast join in a plan?

Q9. Why might Spark choose a sort-merge join for a small table?

Q10. What does AdaptiveSparkPlan isFinalPlan=false mean?

17. Best Practices for Working with Explain Plans

Conclusion

Continue Learning Apache Spark

Related Spark Concepts

Related Guides & Recommended Reads

AI SRE: what is it, how could it change on-call – and does it work?

My Arse, My Rules. Bum Fun, Backdoor Business and My Very Personal Confession About Anal Sex.

Testing Push Notifications in iOS: Simulator and Real Devices

2. How to Call `explain()`

Rule 3: `*(n)` marks whole-stage code generation

Rule 4: Every `Exchange` is a shuffle and a stage boundary

8.1 Column pruning: check `ReadSchema`

8.2 Predicate pushdown: check `PushedFilters`

8.3 Partition pruning: check `PartitionFilters`

Example: a window function without `partitionBy`

Mistake 2: Treating every `Exchange` as a bug

Mistake 4: Confusing `PushedFilters` with skipped data

Q5. What does `Exchange` mean in a physical plan?

Q6. What does the `*(n)` prefix mean?

Q10. What does `AdaptiveSparkPlan isFinalPlan=false` mean?