DataFrame Catalyst Query Plan Tracing
Series
Data Engineering & Distributed Systems Series
Estimated Time
~30 Mins Lab
Lab Objective
Interpret PySpark DataFrame `.explain(True)` plan outputs — identifying Parsed, Analyzed, Optimized, and Physical plan stages for a grouped aggregation query.
Target Query & Execution Code
df = spark.read.json("products.json")
result_df = df.select("category", "price") \
.filter("price > 100") \
.groupBy("category") \
.count()
result_df.explain(True)
Catalyst Plan Breakdown
1. Parsed Logical Plan (AST Syntax Tree):
== Parsed Logical Plan ==
'Aggregate ['category], ['category, 'count(1) AS 'count]
+- 'Filter ('price > 100)
+- 'Project ['category, 'price]
+- 'UnresolvedRelation [products.json]
2. Analyzed Logical Plan (Catalog Types Resolved):
== Analyzed Logical Plan ==
Aggregate [category#1], [category#1, count(1) AS count#2L]
+- Filter (price#2 > 100)
+- Project [category#1, price#2]
+- Relation [category#1,price#2] json
3. Optimized Logical Plan (Predicate Pushdown & Projection Pruning):
== Optimized Logical Plan ==
Aggregate [category#1], [category#1, count(1) AS count#2L]
+- Project [category#1]
+- Filter (price#2 > 100)
+- Relation [category#1,price#2] json
4. Physical Plan (Whole-Stage Code Gen Execution):
== Physical Plan ==
*(2) HashAggregate(keys=[category#1], functions=[count(1)], output=[category#1, count#2L])
+- Exchange hashpartitioning(category#1, 200), ENSURE_REQUIREMENTS, [id=#10]
+- *(1) HashAggregate(keys=[category#1], functions=[partial_count(1)], output=[category#1, count#12L])
+- *(1) Project [category#1]
+- *(1) Filter (isnotnull(price#2) AND (price#2 > 100))
+- FileScan json [category#1,price#2] Batched: false, Format: JSON
Find this content helpful?
☕ Buy me a coffee