Topolearnexplain() is how you verify that Spark will do what you expect before burning cluster hours.
Every DataFrame or SQL query goes through Catalyst, Spark's optimizer:
1. Parsed logical plan: what you wrote. 2. Analyzed plan: names and types resolved against the catalog. Unknown columns fail here. 3. Optimized logical plan: rule-based rewrites such as predicate pushdown, column pruning, constant folding, combining filters and simplifying expressions. 4. Physical plan: concrete operators chosen, like scans, HashAggregate, SortMergeJoin or BroadcastHashJoin, and Exchange (shuffle), with cost-based choices.
df.explain() prints the physical plan; df.explain("formatted") gives a numbered, readable version; "extended" shows all four; "cost" adds statistics.
What to look for:
- `Exchange` nodes are shuffles. Count them; each one is a stage boundary.
- `PushedFilters` and `ReadSchema` on a file scan show filter pushdown and column pruning into Parquet.
- `PartitionFilters` show partition pruning on partitioned tables.
- The join operator chosen (broadcast versus sort-merge).
*(n)prefixes orWholeStageCodegenmark operators fused into generated code.- `AdaptiveSparkPlan isFinalPlan=false` means AQE may still change the plan at runtime.
Python UDFs show up as BatchEvalPython or ArrowEvalPython, and filters can't be pushed through them.
Example
from pyspark.sql import SparkSession, functions as F
spark = SparkSession.builder.master("local[2]").getOrCreate()
(spark.range(1000).withColumn("country", F.when(F.col("id") % 2 == 0, "NO").otherwise("SE"))
.withColumn("amount", F.col("id") * 1.5)
.write.mode("overwrite").parquet("sales_pq"))
q = (spark.read.parquet("sales_pq")
.filter(F.col("country") == "NO")
.groupBy("country").agg(F.sum("amount").alias("total")))
q.explain("formatted")
# == Physical Plan ==
# AdaptiveSparkPlan (...)
# +- HashAggregate (...)
# +- Exchange (...) <- shuffle
# +- HashAggregate (...) <- partial aggregation
# +- Filter (...)
# +- Scan parquet (...)
# ...
# PushedFilters: [IsNotNull(country), EqualTo(country,NO)]
# ReadSchema: struct<country:string,amount:double> (id is pruned)
print(q.first().total) # 374250.0Remember
- Catalyst: parsed, analyzed, optimized logical, then physical plan
- Exchange = shuffle; PushedFilters and ReadSchema = pushdown and pruning
- Use
explain("formatted")before running costly jobs
Watch out for
- Assuming a filter is pushed into the scan without checking PushedFilters, e.g. when a UDF sits in between.