Seeing It with explain()

Concepts covered: Catalyst Optimizer

Everything so far happens inside Spark, out of sight. explain() prints the plan Catalyst produced for a DataFrame. It starts no job and reads no rows, so you can call it on a query over a huge table before you commit any cluster time.

from pyspark.sql import functions as F

orders = spark.read.parquet("/data/lake/orders")

west = (
    orders
    .groupBy("region")
    .agg(F.sum("profit").alias("total_profit"))
    .filter(F.col("region") == "West")
)
west.explain()

The filter is written last, after the grouping. The printed plan is a tree, drawn with the result at the top and the read at the bottom. Data flows upward, so start at the bottom line. The numbers after # are internal ids that change from run to run, and long lines are trimmed here to their first part.

Read	Plan line	What it tells you
6th	`AdaptiveSparkPlan isFinalPlan=false`	A wrapper from a feature a later lesson covers. Start reading on the line below it.
5th	`HashAggregate(keys=[region#2], functions=[sum(profit#3)])`	The final total per region, computed after rows are grouped.
4th	`Exchange hashpartitioning(region#2, 200)`	The shuffle. Rows with the same region go to the same task, across 200 partitions, the default.
3rd	`HashAggregate(keys=[region#2], functions=[partial_sum(profit#3)])`	A subtotal per partition before the shuffle. A later lesson explains this 2-step aggregation.
2nd	`Filter (isnotnull(region#2) AND (region#2 = West))`	Your last line, now running right after the read and before the shuffle, with a NULL check Catalyst added.
1st	`FileScan parquet [region#2,profit#3] ... ReadSchema: struct<region:string,profit:double>`	The read. Only 2 of the 4 columns load: `order_id` and `status` were pruned.

3 questions for every plan

Which columns does the scan read? The bracketed list on the scan line and its ReadSchema should hold only what your query uses. A long list means something above needs those columns, often the final result itself.
Where does each filter sit? A filter next to the scan is cheap. One that stays above an Exchange reads a column that only exists after it, such as an aggregate's output.
How many Exchange lines are there? Each is a shuffle. A groupBy adds a hash shuffle, an orderBy adds a range shuffle, and narrow steps add none.

You wrote a status filter at the end of a chain that groups by region. Which sign in the physical plan confirms Catalyst moved it ahead of the shuffle?

A Filter testing status that appears below the Exchange line, near the scan
A Filter testing status printed at the top, above the final aggregate
The Exchange line listing status inside its hashpartitioning(...)
The word Completed appearing anywhere in the plan's printed text

Before and after the rewrite

explain(mode="extended") prints 4 plans. The parsed plan comes first and is your chain as written. After the analyzed plan comes the optimized plan, which is what Catalyst turned it into, and the physical plan prints last. Comparing the parsed and optimized plans shows each rewrite in action.

On the left, the filter sits above the aggregate. On the right it has dropped below it and gained a NULL check, and a Project now keeps only region and profit. The relation still lists 4 columns at this level, and the physical planner turns that Project into the 2-column scan from the table above.

Spotting the RDD wall

Call explain() on the totals DataFrame from the escape-hatch section, and you get a short plan whose bottom line reads Scan ExistingRDD[region#20,total_profit#21]. The 3 lambdas and the shuffle inside reduceByKey are all missing, because Catalyst received the RDD as a finished source. A plan that starts at Scan ExistingRDD tells you 2 things: the steps between .rdd and toDF were not optimized and don't appear here, and counting its Exchange lines undercounts the job's shuffles. Any DataFrame work before .rdd was optimized on its own and isn't shown in this plan either.

How many shuffles does the physical plan for this query contain?

report = (
    orders
    .filter(F.col("status") == "Completed")
    .withColumn("profit_k", F.col("profit") / 1000)
    .groupBy("region")
    .agg(F.sum("profit_k").alias("total_k"))
    .orderBy(F.col("total_k").desc())
)
report.explain()

1, because Catalyst folds the sort into the grouping's shuffle
3, one each for the filter, the groupBy and the orderBy
2: a hash Exchange for groupBy and a range Exchange for orderBy
0, because explain() shows no shuffles until the job has run

About this lesson

This section is part of the Catalyst Optimizer: Beginner lesson on DataDriven, the home of data engineering. Lessons cover SQL, Python, Spark, data modeling and pipeline architecture, and each section pairs its explanation with worked examples and practice you run as you go.