⚡ PySpark SQL Engine

DfSpark SQL Engine

The Spark SQL Engine is the computational backend that translates SQL queries and DataFrame operations into optimized RDD computations. It consists of the Catalyst optimizer (logical/physical plan optimization) and the Tungsten execution engine (memory management and code generation).

DfWhole-Stage Code Generation

Whole-stage code generation (Tungsten) compiles multiple operators into a single optimized JVM function, eliminating virtual function calls and leveraging CPU registers and SIMD instructions. This can improve query performance by 2x–10x.

Cost_{plan} = Cost_{scan} + Cost_{filter} + Cost_{project} + Cost_{join} + Cost_{agg}

Tungsten Memory Format

Row_{binary} = [null\\_bitmap | fixed\\_length\\_columns | variable\\_length\\_offsets | variable\\_length\\_columns]

Here,

$null\_bitmap$ =Bit array marking NULL values (1 bit per column)
$fixed\_length\_columns$ =Fixed-width values stored sequentially (8 bytes each)
$variable\_length\_offsets$ =Offset pointers to variable-length data
$variable\_length\_columns$ =Variable-width values (strings, arrays, maps)

Catalyst optimization rules include: Constant Folding, Boolean Simplification, Filter Pushdown, Column Pruning, Join Reordering, and Subquery Elimination. These rules are applied iteratively until no further improvements are found.

Enable AQE (Adaptive Query Execution) in Spark 3.x to automatically optimize shuffle partition count, convert sort-merge joins to broadcast joins at runtime, and handle data skew.

ThTungsten Memory Efficiency

Theorem: Tungsten's off-heap binary row format achieves memory efficiency ≥ 2x compared to Java object serialization by eliminating object headers, pointer indirection, and GC pressure. This allows processing datasets larger than heap size through managed off-heap allocation.

Spark SQL = Catalyst (optimizer) + Tungsten (execution engine)
Catalyst: Analysis → Optimization → Physical Planning → Code Generation
Tungsten: Off-heap binary format eliminates GC overhead, whole-stage codegen eliminates virtual calls
AQE (Adaptive Query Execution) provides runtime optimization
Predicate pushdown reduces data volume before shuffle/join operations

🏗️ Architecture Diagram

Architecture Diagram

┌─────────────────────────────────────────────────────────────────┐
│                 SPARK SQL ENGINE ARCHITECTURE                    │
├─────────────────────────────────────────────────────────────────┤
│                                                                 │
│  ┌─────────────────────────────────────────────────────────┐   │
│  │                  API Layer (DataFrame/SQL)               │   │
│  │  ┌─────────┐  ┌─────────┐  ┌─────────┐  ┌─────────┐  │   │
│  │  │DataFrame│  │  SQL    │  │  Hive   │  │  JDBC   │  │   │
│  │  │  API    │  │ Queries │  │  QL     │  │  JDBC   │  │   │
│  │  └────┬────┘  └────┬────┘  └────┬────┘  └────┬────┘  │   │
│  └───────┼────────────┼────────────┼────────────┼────────┘   │
│          │            │            │            │              │
│          ▼            ▼            ▼            ▼              │
│  ┌─────────────────────────────────────────────────────────┐   │
│  │              SQL Context / SparkSession                   │   │
│  │  ┌──────────┐  ┌──────────┐  ┌──────────┐             │   │
│  │  │  Parser  │→ │ Analyzer │→ │ Optimizer │             │   │
│  │  └──────────┘  └──────────┘  └──────────┘             │   │
│  └─────────────────────────────────────────────────────────┘   │
│                          │                                      │
│                          ▼                                      │
│  ┌─────────────────────────────────────────────────────────┐   │
│  │              Catalyst Optimizer Pipeline                  │   │
│  │                                                          │   │
│  │  ┌──────────┐  ┌──────────┐  ┌──────────┐             │   │
│  │  │ Logical  │→ │ Logical  │→ │ Physical │             │   │
│  │  │  Plan 1  │  │  Plan 2  │  │  Plans   │             │   │
│  │  └──────────┘  └──────────┘  └──────────┘             │   │
│  │       ↑              ↑              ↑                   │   │
│  │       │              │              │                   │   │
│  │  ┌────┴────┐  ┌──────┴──────┐  ┌───┴────┐            │   │
│  │  │Analysis │  │ Optimization│  │  Cost  │            │   │
│  │  │  Rules  │  │    Rules    │  │  Model │            │   │
│  │  └─────────┘  └─────────────┘  └────────┘            │   │
│  └─────────────────────────────────────────────────────────┘   │
│                          │                                      │
│                          ▼                                      │
│  ┌─────────────────────────────────────────────────────────┐   │
│  │              Tungsten Execution Engine                   │   │
│  │  ┌──────────┐  ┌──────────┐  ┌──────────┐             │   │
│  │  │  Code    │  │  Memory  │  │  Shuffle │             │   │
│  │  │Generation│  │ Manager  │  │  Manager │             │   │
│  │  └──────────┘  └──────────┘  └──────────┘             │   │
│  └─────────────────────────────────────────────────────────┘   │
│                          │                                      │
│                          ▼                                      │
│  ┌─────────────────────────────────────────────────────────┐   │
│  │              Spark Core (RDD Execution)                  │   │
│  │  ┌──────────┐  ┌──────────┐  ┌──────────┐             │   │
│  │  │  Task    │  │  Stage   │  │  DAG     │             │   │
│  │  │Scheduler │  │ Scheduler│  │ Scheduler│             │   │
│  │  └──────────┘  └──────────┘  └──────────┘             │   │
│  └─────────────────────────────────────────────────────────┘   │
└─────────────────────────────────────────────────────────────────┘

Architecture Diagram

┌─────────────────────────────────────────────────────────────────┐
│              CATALYST OPTIMIZER RULES CATALOG                    │
├─────────────────────────────────────────────────────────────────┤
│                                                                 │
│  ┌─────────────────────────────────────────────────────────┐   │
│  │  ANALYSIS RULES (Schema Resolution)                     │   │
│  │  ├── ResolveRelations: Table → Catalog lookup           │   │
│  │  ├── ResolveReferences: Column → Attribute binding      │   │
│  │  ├── ResolveFunctions: Function → Registry lookup       │   │
│  │  ├── ResolveSubquery: Subquery → LogicalPlan            │   │
│  │  └── TypeCoercion: Implicit type casting                │   │
│  └─────────────────────────────────────────────────────────┘   │
│                                                                 │
│  ┌─────────────────────────────────────────────────────────┐   │
│  │  OPTIMIZATION RULES (Plan Rewriting)                    │   │
│  │  ├── PushDownPredicate: Filter → Scan                   │   │
│  │  ├── PruneColumns: Select only needed columns           │   │
│  │  ├── ConstantFolding: Evaluate constants at compile     │   │
│  │  ├── EliminateSorts: Remove unnecessary sorts           │   │
│  │  ├── CombineFilters: Merge adjacent filters             │   │
│  │  ├── CollapseProject: Merge adjacent projections        │   │
│  │  ├── PushThroughAggregate: Push predicate past agg      │   │
│  │  ├── PushDownJoinPredicates: Filter push to join sides  │   │
│  │  ├── ReorderJoin: Optimal join order                    │   │
│  │  └── ConvertToIn: Subquery → IN expression              │   │
│  └─────────────────────────────────────────────────────────┘   │
│                                                                 │
│  ┌─────────────────────────────────────────────────────────┐   │
│  │  PHYSICAL PLANNING RULES                                │   │
│  │  ├── JoinSelection: Broadcast/SortMerge/HashJoin        │   │
│  │  ├── EnsureRequirements: Add shuffles/sorts             │   │
│  │  └── Partitioning: Map partitions to physical layout    │   │
│  └─────────────────────────────────────────────────────────┘   │
│                                                                 │
│  ┌─────────────────────────────────────────────────────────┐   │
│  │  STRATEGIES (Rule Ordering)                             │   │
│  │  1. Resolution → 2. Optimization → 3. Planning         │   │
│  │  Each strategy applies rules until fixpoint             │   │
│  └─────────────────────────────────────────────────────────┘   │
└─────────────────────────────────────────────────────────────────┘

Architecture Diagram

┌─────────────────────────────────────────────────────────────────┐
│                  QUERY EXECUTION FLOW                            │
├─────────────────────────────────────────────────────────────────┤
│                                                                 │
│  SQL Query: "SELECT dept, AVG(salary) FROM emp WHERE age>30    │
│              GROUP BY dept HAVING AVG(salary) > 75000"         │
│                                                                 │
│  ┌─────────────────────────────────────────────────────────┐   │
│  │ Step 1: PARSING                                          │   │
│  │ Input: SQL string → AST (Abstract Syntax Tree)          │   │
│  │ Output: Unresolved Logical Plan                         │   │
│  └─────────────────────────────────────────────────────────┘   │
│       │                                                         │
│       ▼                                                         │
│  ┌─────────────────────────────────────────────────────────┐   │
│  │ Step 2: ANALYSIS                                         │   │
│  │ • Resolve table "emp" from catalog                      │   │
│  │ • Bind columns "dept", "salary", "age"                  │   │
│  │ • Resolve AVG function                                   │   │
│  │ Output: Resolved Logical Plan                           │   │
│  └─────────────────────────────────────────────────────────┘   │
│       │                                                         │
│       ▼                                                         │
│  ┌─────────────────────────────────────────────────────────┐   │
│  │ Step 3: OPTIMIZATION                                     │   │
│  │ • Push filter (age > 30) to scan                        │   │
│  │ • Prune unused columns                                   │   │
│  │ • Optimize aggregation order                             │   │
│  │ Output: Optimized Logical Plan                          │   │
│  └─────────────────────────────────────────────────────────┘   │
│       │                                                         │
│       ▼                                                         │
│  ┌─────────────────────────────────────────────────────────┐   │
│  │ Step 4: PHYSICAL PLANNING                                │   │
│  │ • Choose hash aggregate vs sort aggregate               │   │
│  │ • Decide partitioning strategy                          │   │
│  │ • Select exchange (shuffle) strategy                    │   │
│  │ Output: Physical Plan (multiple candidates)             │   │
│  └─────────────────────────────────────────────────────────┘   │
│       │                                                         │
│       ▼                                                         │
│  ┌─────────────────────────────────────────────────────────┐   │
│  │ Step 5: CODE GENERATION (Tungsten)                       │   │
│  │ • Whole-stage code generation                           │   │
│  │ • Generate JVM bytecode                                 │   │
│  │ • Compile to native code                                │   │
│  │ Output: RDD[InternalRow] execution plan                 │   │
│  └─────────────────────────────────────────────────────────┘   │
│       │                                                         │
│       ▼                                                         │
│  ┌─────────────────────────────────────────────────────────┐   │
│  │ Step 6: EXECUTION                                         │   │
│  │ • Execute RDD plan on cluster                           │   │
│  │ • Schedule tasks across executors                       │   │
│  │ • Return results to driver or write to storage          │   │
│  └─────────────────────────────────────────────────────────┘   │
└─────────────────────────────────────────────────────────────────┘

📚 Detailed Explanation

1. Spark SQL Engine Overview

Spark SQL is a Spark module for structured data processing. It provides:

A programming abstraction called DataFrames and Datasets
A SQL query engine that can run SQL/HiveQL queries
Integration with Spark's ecosystem (Spark Streaming, MLlib)
Connectivity to various data sources (Parquet, JSON, JDBC, Hive)

The engine consists of three main components:

Catalyst Optimizer: Transforms logical plans into optimized physical plans
Tungsten Execution Engine: Efficient runtime execution with code generation
Data Source API: Integration with various storage systems

2. Catalyst Optimizer Deep Dive

The Catalyst optimizer is a framework for manipulating query plans. It uses:

Trees: Representations of query plans
Rules: Transformations applied to trees
Strategies: Ordering of rule application

Optimization Techniques:

Predicate Pushdown:

Architecture Diagram

Before: Project(name) → Filter(age > 30) → Scan(all columns)
After:  Project(name) → Scan(name, age) → Filter(age > 30)

Column Pruning:

Architecture Diagram

Before: Project(name, age) → Scan(id, name, age, salary, dept)
After:  Project(name, age) → Scan(name, age)

Constant Folding:

Architecture Diagram

Before: Filter(age > 25 + 5)
After:  Filter(age > 30)

Join Reordering:

Architecture Diagram

Before: (A ⋈ B) ⋈ C
After:  A ⋈ (B ⋈ C)  -- if B⋈C is smaller

3. Tungsten Execution Engine

Tungsten focuses on three areas:

Memory Management:

Off-heap memory to avoid GC overhead
Binary row format for data storage
Cache-aware algorithms (hash joins, sorts)

Code Generation:

Whole-stage code generation (since Spark 2.0)
Generates tight JVM bytecode loops
Eliminates virtual function calls
Enables CPU pipeline optimization

Columnar Processing:

Vectorized operations (batch processing)
SIMD-friendly data layout
Efficient compression (dictionary, run-length, delta encoding)

4. Physical Plan Strategies

Join Selection:

Broadcast Hash Join: For small tables (< 10MB default)
Sort-Merge Join: For large tables with sorted data
Shuffle Hash Join: For medium-sized tables
Cartesian Product: For cross joins (avoid!)

Aggregation Strategies:

Hash Aggregation: Fast, uses hash table
Sort Aggregation: Better for sorted input
Tungsten Aggregation: Optimized with code generation

5. Shuffle and Exchange Operations

Shuffles occur when data needs to be redistributed across partitions:

Wide transformations (groupBy, join, repartition)
Data exchange between stages
Most expensive operation in Spark

Shuffle Optimization:

Use broadcast joins to avoid shuffle
Partition data to minimize shuffle
Use bucketing for repeated joins
Tune shuffle partition count

6. Data Source API

Spark supports multiple data sources:

File formats: Parquet, ORC, JSON, CSV, Text
JDBC: Relational databases
Hive: Hive tables
Custom: Through DataSource API v2

Predicate Pushdown to Sources:

Architecture Diagram

Parquet: Row group filtering via statistics
ORC: Bloom filter pushdown
JDBC: SQL WHERE clause pushdown

7. Performance Monitoring

Spark UI Metrics:

SQL tab: Query plans, execution times
Stages tab: Task distribution, shuffle read/write
Storage tab: Cached data, memory usage

Key Metrics:

scanTime: Time to read data
shuffleReadTime: Time to read shuffle data
shuffleWriteTime: Time to write shuffle data
taskDuration: Time per task
gcTime: Garbage collection time

🔑 Key Concepts Table

Concept	Description	Optimization
Catalyst	Query optimizer for logical/physical plans	Automatic plan optimization
Tungsten	Execution engine with code generation	Whole-stage code generation
Predicate Pushdown	Filter pushed to data source	Reduce I/O
Column Pruning	Read only required columns	Reduce memory/I/O
Constant Folding	Evaluate constants at compile time	Reduce runtime computation
Broadcast Join	Small table broadcast to executors	Avoid shuffle
Sort-Merge Join	Join sorted partitions	Efficient for large tables
Hash Aggregation	Use hash table for aggregation	Fast for groupBy
Shuffle	Data redistribution across partitions	Most expensive operation
Whole-Stage Codegen	Generate tight JVM bytecode	Eliminate virtual calls

💻 Code Examples

Example 1: SQL Queries with Catalyst

from pyspark.sql import SparkSession

spark = SparkSession.builder \
    .appName("SparkSQLEngine") \
    .config("spark.sql.adaptive.enabled", "true") \
    .config("spark.sql.adaptive.coalescePartitions.enabled", "true") \
    .getOrCreate()

# Create tables
employees = spark.createDataFrame([
    (1, "Alice", "Engineering", 30, 75000),
    (2, "Bob", "Marketing", 25, 65000),
    (3, "Charlie", "Engineering", 35, 90000),
    (4, "Diana", "Marketing", 28, 70000),
    (5, "Eve", "Engineering", 32, 85000),
    (6, "Frank", "Sales", 29, 60000),
    (7, "Grace", "Sales", 31, 68000)
], ["id", "name", "department", "age", "salary"])

employees.createOrReplaceTempView("employees")

# Simple query
result = spark.sql("""
    SELECT department, 
           COUNT(*) as emp_count,
           AVG(salary) as avg_salary,
           MAX(age) as max_age
    FROM employees
    WHERE age > 25
    GROUP BY department
    HAVING COUNT(*) > 1
    ORDER BY avg_salary DESC
""")

result.show()

# Explain the query plan
result.explain(True)

Example 2: Catalyst Optimizer Rules

# Show optimized plan
df = spark.sql("""
    SELECT e.name, e.salary, d.location
    FROM employees e
    JOIN departments d ON e.department = d.name
    WHERE e.age > 30
    AND e.salary > 70000
""")

# Print physical plan with optimizations
df.explain(True)

# Output shows:
# == Parsed Logical Plan ==
# ...
# == Analyzed Logical Plan ==
# ...
# == Optimized Logical Plan ==
# ... (with predicate pushdown, column pruning)
# == Physical Plan ==
# ... (with join strategy, exchange)

Example 3: Adaptive Query Execution (Spark 3.0+)

# Enable AQE
spark.conf.set("spark.sql.adaptive.enabled", "true")
spark.conf.set("spark.sql.adaptive.coalescePartitions.enabled", "true")
spark.conf.set("spark.sql.adaptive.skewJoin.enabled", "true")

# Complex query that benefits from AQE
result = spark.sql("""
    SELECT 
        department,
        COUNT(*) as cnt,
        AVG(salary) as avg_sal,
        PERCENTILE(salary, 0.5) as median_sal
    FROM employees
    GROUP BY department
    HAVING COUNT(*) >= 2
""")

# AQE will:
# 1. Coalesce partitions after shuffle
# 2. Handle data skew automatically
# 3. Optimize join strategies based on runtime stats
result.show()

# Check AQE optimizations in UI
print(f"Number of partitions: {result.rdd.getNumPartitions()}")

Example 4: Data Source Optimization

# Read Parquet with predicate pushdown
parquet_df = spark.read.parquet("employee_data.parquet")

# Filter pushed to Parquet reader
filtered = parquet_df.filter(
    (parquet_df.age > 30) & 
    (parquet_df.salary > 70000)
)

# Only reads relevant row groups
filtered.explain()

# Write with bucketing for future joins
employees.write \
    .bucketBy(10, "department") \
    .sortBy("id") \
    .saveAsTable("bucketed_employees")

# Bucketed join avoids shuffle
spark.sql("""
    SELECT /*+ MAPJOIN(e2) */
        e1.name, e2.name as manager
    FROM bucketed_employees e1
    JOIN employees e2 ON e1.department = e2.department
""").show()

📊 Performance Metrics

Query Pattern	Without Catalyst	With Catalyst	Improvement
Simple SELECT	150ms	45ms	3.3x
Filter + Agg	320ms	85ms	3.8x
JOIN + WHERE	850ms	210ms	4.0x
Subquery	1200ms	280ms	4.3x
Window Function	450ms	120ms	3.8x
Read Parquet	200ms	60ms	3.3x
Write Parquet	300ms	150ms	2.0x
Shuffle	500ms	350ms	1.4x

✅ Best Practices

1. Enable Adaptive Query Execution

spark.conf.set("spark.sql.adaptive.enabled", "true")
spark.conf.set("spark.sql.adaptive.coalescePartitions.enabled", "true")
spark.conf.set("spark.sql.adaptive.skewJoin.enabled", "true")

2. Use Broadcast Joins for Small Tables

from pyspark.sql.functions import broadcast
result = large_df.join(broadcast(small_df), "key")
# Or configure threshold
spark.conf.set("spark.sql.autoBroadcastJoinThreshold", "10MB")

3. Cache Frequently Used DataFrames

df.cache()  # For DataFrames reused multiple times
# Or persist with specific storage level
from pyspark import StorageLevel
df.persist(StorageLevel.MEMORY_AND_DISK_SER)

4. Use Appropriate File Formats

# Parquet for columnar storage
df.write.parquet("output/", compression="snappy")

# ORC for Hive integration
df.write.orc("output/")

# Avoid CSV/JSON for large datasets

5. Partition Data Strategically

# Partition by frequently filtered columns
df.write.partitionBy("year", "month").parquet("output/")

# Use bucketing for join optimization
df.write.bucketBy(100, "user_id").saveAsTable("users_bucketed")

6. Monitor Query Plans

# Always check explain plans
df.explain(True)

# Look for:
# - BroadcastHashSortMergeJoin (good for small tables)
# - FileScan with pushed filters (predicate pushdown)
# - Project with only needed columns (column pruning)

PySpark SQL Engine: Execution, Catalyst Optimizer, and Tungsten

⚡ PySpark SQL Engine

DfSpark SQL Engine

DfWhole-Stage Code Generation

Tungsten Memory Format

ThTungsten Memory Efficiency

🏗️ Architecture Diagram

📚 Detailed Explanation

1. Spark SQL Engine Overview

2. Catalyst Optimizer Deep Dive

3. Tungsten Execution Engine

4. Physical Plan Strategies

5. Shuffle and Exchange Operations

6. Data Source API

7. Performance Monitoring

🔑 Key Concepts Table

💻 Code Examples

Example 1: SQL Queries with Catalyst

Example 2: Catalyst Optimizer Rules

Example 3: Adaptive Query Execution (Spark 3.0+)

Example 4: Data Source Optimization

📊 Performance Metrics

✅ Best Practices

1. Enable Adaptive Query Execution

2. Use Broadcast Joins for Small Tables

3. Cache Frequently Used DataFrames

4. Use Appropriate File Formats

5. Partition Data Strategically

6. Monitor Query Plans

See Also

Need Expert PySpark Help?