Understanding Spark’s Execution Plan: How Spark Executes Your Code

The part we care about deeply is everything that happens on the driver before tasks are launched: the Catalyst optimizer and the physical planning phase.

How can we see these various stages in Spark? (Various versions of ‘Explain’ command)

Spark gives you five official EXPLAIN modes (Spark 3.0+). We will go through each stage in order.

Stage-by-Stage Deep Dive

1. Unresolved Logical Plan (also called "Parsed Logical Plan")

• This is the very first tree Catalyst builds from your DataFrame operations.

• It is called "unresolved" because attribute references (column names) are still strings and have no type or scope information.

• Catalyst uses the Analyzer later to resolve them.

Unresolved Logical Plan (you rarely see this directly, but extended explain shows it at the top):

This tree is built purely by syntactic transformation of DataFrame operations.

2. Analyzed (Resolved) Logical Plan

The Analyzer runs a set of resolution rules using the Catalog and Analysis rules:

• Bind column names to attributes in input relations

• Resolve functions (e.g., count, sum)

• Type checking and coercion

• Subquery resolution

• View/CTE handling

Key resolution rules:

• ResolveRelations

• ResolveReferences

• ResolveFunctions

• ResolveAliases

• ResolveAggFunctions, etc.

After this phase, every attribute has an AttributeReference with a unique exprId, data type, nullability, and metadata.

Output example:

Notice user_id#40L – the L suffix means Long, and #40 is a unique expression ID.

3. Optimized Logical Plan (Catalyst Optimization)

This is where Spark's magic happens. Catalyst applies a series of rule batches repeatedly until fix-point.

Major rule batches (in order of execution):

Since Spark 2.2+ there is also Cost-Based Optimization (CBO) – optional, enabled by:

SET spark.sql.cbo.enabled=true
SET spark.sql.cbo.joinReorder.enabled=true

CBO adds:

• Table/column statistics collection (ANALYZE TABLE ... COMPUTE STATISTICS)

• Cardinality estimation

• Cost model for different join strategies (Broadcast vs Shuffle Hash vs Sort-Merge)

With CBO enabled, you will see cost numbers in explain(cost=true):

4. Physical Plans

Spark now selects a physical operator for every logical operator. This is done by a sequence of planning strategies (in SparkStrategies.scala):

Key physical planning decisions:

1. Join Selection

   • If one side ≤ spark.sql.autoBroadcastJoinThreshold (default 10MB) → BroadcastHashJoinExec

   • Else if both sides have equi-keys and sorting is cheap → SortMergeJoinExec

   • Else → ShuffledHashJoinExec

2. Aggregation Strategy

   • If grouping keys have high cardinality → SortAggregate (partial sort + final sort)

   • If low cardinality or pre-shuffled → HashAggregate

   • If using object-based aggregation (maps) → ObjectHashAggregateExec

3. WholeStage Code Generation (Spark 2.0+)

   • Multiple physical operators are collapsed into a single Java class using Janino compiler.

   • Node groups are marked with * in explain() output:

     *(1) Filter
     *(1) ColumnarToRow
     *(1) FileScan parquet ...

   • Each * group becomes one WholeStageCodegenExec → one Java method with fused loops → huge performance win.

4. Adaptive Query Execution (AQE) – Spark 3.0+ Enabled by default in Spark 3.2+:

   spark.sql.adaptive.enabled=true

   • Dynamically coalesces shuffle partitions

   • Switches shuffle hash join → broadcast if size misestimated

   • Optimizes skew joins (skewed partitions split)

   You will see AQE plans change at runtime:

How Everything Connects – From Code to Tasks

Let’s walk through a concrete example:

Execution timeline on driver:

1. df.write() is an action → triggers planning

2. Catalyst builds Unresolved Logical Plan from the lineage

3. Analyzer resolves columns, functions → Analyzed Logical Plan

4. Optimizer applies 30+ rule batches → Optimized Logical Plan (filter pushed into Parquet scan, broadcast hint respected)

5. Physical planner:

   • Chooses BroadcastHashJoinExec because of hint

   • Chooses HashAggregateExec for groupBy

   • Collapses filter + scan + projection into WholeStageCodegen

6. DAGScheduler:

   • Breaks the plan into stages at shuffle boundaries

   • Stage 1: read users → broadcast exchange

   • Stage 2: read events → filter → hash join → aggregate → shuffle write

   • Stage 3: final aggregation (if partial agg was used)

7. Each stage → tasks (one per partition)

8. Tasks run on executors, produce shuffle files or final output.

Reading a Real Physical Plan (Formatted Explain)

Example output snippet:

Interpretation:

• * groups = WholeStage CodeGeneration units

• Numbers in parentheses = stage + codegen ID

• Exchange = shuffle boundary (new stage)

• BroadcastExchange = no shuffle, data broadcasted

• ColumnarToRow = transition from Arrow/columnar to row-based processing

Key Takeaways:

1. Spark is lazy → everything is just a logical plan until an action.

2. Catalyst has four major phases:

   • Unresolved → Analyzed → Optimized Logical → Physical

3. Optimization is rule-based + optional cost-based (CBO).

4. Physical planning decides join types, aggregation strategy, codegen.

5. WholeStage CodeGen is the #1 performance feature of Spark 2.0+.

6. AQE (Spark 3+) makes many decisions at runtime.

7. Understanding exprId, WholeStageCodegen IDs, and Exchange nodes is key to reading plans.

With this knowledge you can now:

• Explain why a query is slow by reading the physical plan

• Force broadcast joins, control aggregation strategy

• Write better Spark SQL that benefits from predicate pushdown

• Debug shuffle explosions or codegen failures