Spark Catalyst Optimizer
Catalyst is the engine inside Spark SQL that rewrites your query into a faster plan. You describe what result you want. Catalyst decides the best way to produce it. Every DataFrame and SQL query passes through Catalyst before any data moves.
The Layman View
A delivery driver receives a list of ten addresses. A smart navigation app reorders the stops, skips closed roads, and picks the shortest route. Catalyst plays the role of that app. Your query is the address list, and the plan is the route.
The Four Phases
Your query (SQL or DataFrame)
|
v
1. Analysis Resolve table names, column names, and types
|
v
2. Logical Optimization Rewrite the plan with smart rules
|
v
3. Physical Planning Choose join types and scan methods; compare costs
|
v
4. Code Generation Compile the winning plan into fast Java bytecode
|
v
Executors run the code
Phase 1: Analysis
Spark checks that every table and column exists and attaches data types. A misspelled column name stops the query here with an AnalysisException.
Phase 2: Logical Optimization
Catalyst applies rules that reduce work without changing the result.
- Predicate pushdown moves filters as close to the data source as possible.
- Column pruning reads only the columns that the query uses.
- Constant folding calculates fixed expressions once, such as 2 * 60 becoming 120.
- Filter combining merges several filters into one pass.
Written plan: Optimized plan: Scan(all columns, all rows) Scan(columns: name, age | rows: age > 30 only) Join with departments | | Join with departments Filter(age > 30) | | Project(name) Project(name)
The optimized plan filters before the join, so the join handles fewer rows.
Phase 3: Physical Planning
Catalyst produces several physical plans and picks the cheapest. A small table triggers a broadcast join. Two large tables trigger a sort-merge join.
Phase 4: Code Generation
Spark merges several operators into one tight loop of generated code. This technique, called whole-stage code generation, removes the cost of calling one operator after another for every row. The Tungsten engine supports it with compact memory layouts that avoid slow Java object handling.
Reading a Plan with explain
df = spark.read.parquet("/data/people")
result = df.filter(df.age > 30).select("name")
result.explain(True)
== Parsed Logical Plan == what you wrote == Analyzed Logical Plan == names and types resolved == Optimized Logical Plan == after Catalyst rules == Physical Plan == what executors run
Look for PushedFilters and ReadSchema in the physical plan. PushedFilters shows the conditions that reached the file reader. ReadSchema lists the columns that Spark reads.
FileScan parquet [name#1] PushedFilters: [IsNotNull(age), GreaterThan(age,30)] ReadSchema: struct<name:string,age:int>
What Blocks Optimization
- Python UDFs hide logic from Catalyst. Prefer built-in functions.
- Casting a column inside a filter can stop pushdown. Compare columns with literals of the same type.
- Reading CSV or JSON gives fewer pushdown benefits than Parquet.
- Converting a DataFrame to an RDD leaves the optimizer behind.
Helping Catalyst
Run ANALYZE TABLE to collect statistics such as row counts. Catalyst uses these numbers to choose better join strategies.
spark.sql("ANALYZE TABLE sales COMPUTE STATISTICS")
A Quick Experiment
Write the same query in two orders and compare the plans. The filter before the join and the filter after the join produce identical optimized plans, because Catalyst moves the filter on its own.
a = orders.join(customers, "customer_id").filter("country = 'IN'")
b = orders.join(customers.filter("country = 'IN'"), "customer_id")
a.explain()
b.explain()
Run both and read the PushedFilters line. Matching output proves that you can write clear code and trust the optimizer to handle the order of steps.