Skip to main content
PySparkSnowflakeDatabricks

Catalyst Optimizer & Tungsten Execution Engine in PySpark — Under the Hood

At NeoMart, queries must run efficiently even on billions of rows:

  • Aggregating sales by region
  • Joining large customer and order datasets
  • Filtering and transforming product catalogs

PySpark achieves high performance through the Catalyst Optimizer and Tungsten execution engine.

1. What is Catalyst Optimizer?​

Catalyst is Spark SQL’s query optimizer:

  • Automatically analyzes, rewrites, and optimizes logical query plans
  • Supports rule-based and cost-based optimization
  • Handles DataFrame and SQL queries

Example: Filter Pushdown​

df_filtered = df.filter(F.col("price") > 100)
df_filtered.explain()

Output (simplified)​

== Physical Plan ==
*(1) Filter (price > 100)
+- Scan Parquet [product_id, name, price]

✅ Catalyst automatically pushes filter to scan level, reducing data read.

2. Key Catalyst Optimizations​

  1. Predicate Pushdown → filter data early
  2. Column Pruning → read only required columns
  3. Constant Folding → compute constants at compile time
  4. Reordering Joins → optimize join order based on data size
  5. Code Generation → generate JVM bytecode for transformations

Story: NeoMart saves hours of compute on large joins thanks to Catalyst reordering and pushdowns.

3. Tungsten Execution Engine​

Tungsten focuses on physical execution:

  • Memory management → off-heap memory
  • Code generation → avoids Java object overhead
  • Efficient CPU utilization → batch processing of rows

Benefits​

FeatureBenefit
Off-heap memoryReduces GC pauses
Whole-stage codegenFewer Java objects, faster execution
Binary processingHigh CPU efficiency, lower memory footprint

4. Example: Whole-Stage Code Generation​

df_grouped = df.groupBy("category").agg(F.sum("price").alias("total_price"))
df_grouped.explain(True)

Output snippet​

== Physical Plan ==
*(2) HashAggregate(keys=[category], functions=[sum(price)])
+- *(2) ColumnarToRow
   +- FileScan parquet ...
  • Tungsten generates optimized bytecode for aggregation
  • Processing is vectorized, minimizing memory allocation

5. Combining Catalyst & Tungsten​

  • Catalyst → logical query plan optimization
  • Tungsten → physical execution efficiency

NeoMart Example:

  • Join customer and order tables
  • Catalyst optimizes join order and pushes filters
  • Tungsten executes aggregation in memory-efficient batches
  • Result → faster runtime, lower memory usage

6. Best Practices to Leverage Optimizer​

✔ Use DataFrame API over RDD → Catalyst optimizations enabled ✔ Avoid Python UDFs when possible → breaks Catalyst optimization ✔ Cache intermediate results for iterative pipelines ✔ Use Parquet/ORC → supports predicate pushdown ✔ Monitor explain() plans → identify wide transformations

Summary​

  • Catalyst Optimizer → smart logical query planning

  • Tungsten Execution Engine → optimized physical execution

  • Together, they provide:

    • Filter pushdown & column pruning
    • Off-heap memory & code generation
    • Efficient joins, aggregations, and scans

NeoMart’s high-scale pipelines rely on Catalyst + Tungsten to process terabytes of data quickly and reliably.

Next Topic → Performance Tuning: Partitions, Repartition, Coalesce