The Evolution of Apache Spark Since Version 2: A Technical Deep Dive

• DataFrame became an alias for Dataset[Row]

• Strongly typed operations (via Dataset[T]) provided compile-time type safety in Scala/Java

• The Catalyst optimizer and Tungsten execution engine could now apply aggressive optimizations across both APIs

1. Analysis — resolves attribute references and relations against the catalog

2. Logical Optimization — applies rule-based rewrites (predicate pushdown, constant folding, etc.)

3. Physical Planning — selects execution strategies (e.g., BroadcastHashJoin vs SortMergeJoin)

4. Code Generation — uses Janino to compile query fragments into JVM bytecode

• Append — only new rows emitted

• Complete — entire result table re-emitted (useful for aggregations)

• Update — only changed rows emitted

• No support for arbitrary stateful operations with full user control → addressed by mapGroupsWithState and later flatMapGroupsWithState

• No native support for exactly-once end-to-end without careful sink design

1. Dynamically coalescing shuffle partitions — instead of setting spark.sql.shuffle.partitions statically (default 200, often wrong), Spark merges small partitions post-shuffle based on actual data sizes.

2. Dynamically switching join strategies — if runtime stats show that a table is small enough, Spark can switch from a planned SortMergeJoin to a BroadcastHashJoin on the fly.

3. Dynamically optimizing skew joins — detects skewed partitions and splits them, processing each sub-partition independently to avoid stragglers.

sql
SELECT * FROM orders o
JOIN dates d ON o.date_id = d.id
WHERE d.year = 2024

• Pandas UDFs (formerly called Vectorized UDFs) were re-categorized into clearer types: SCALAR, SCALAR_ITER, GROUPED_MAP, GROUPED_AGG

• Python type hints became the primary way to declare UDF semantics, replacing the explicit PandasUDFType enum

• Iterator-based Pandas UDFs allowed amortizing expensive setup costs (e.g., loading an ML model once per partition)

python
from pyspark.sql.functions import pandas_udf
import pandas as pd

@pandas_udf("double")
def predict(features: pd.Series) -> pd.Series:
    model = load_model()  # called once per partition in SCALAR_ITER
    return model.predict(features)

• mapInPandas and mapInArrow — new Pandas-on-Spark functions for efficient, arbitrary transformations on groups of rows using Arrow-backed buffers

• Node Decommissioning — graceful removal of executor nodes, enabling elastic scaling without data loss

• SQL standard compliance improvements: EXCEPT ALL, INTERSECT ALL, lateral subqueries

• Not all pandas operations map 1:1 (operations requiring global ordering are expensive)

• pyspark.pandas DataFrames are lazily evaluated and backed by Spark's optimizer

• spark.sql.execution.arrow.pyspark.enabled=true is recommended for performance

• Spark Connect (preview) — a new client-server architecture decoupling the driver from the cluster, enabling thin clients, better language support, and IDE integration

• CHAR/VARCHAR type support — long-awaited for compatibility with Hive and other SQL engines

• Error message improvements — structured error classes with error codes, much friendlier for debugging

• Histograms in cost-based optimizer — richer statistics for better join strategy decisions

• Lightweight Python/Scala/Rust clients

• Stable API versioning between client and server

• Running notebooks or IDEs against a remote Spark cluster without a fat driver dependency

• ArrowEvalPython operator replaces the old BatchEvalPython for Pandas UDFs

• spark.sql.execution.arrow.maxRecordsPerBatch controls Arrow batch sizes

• Zero-copy data sharing between JVM and Python processes via Arrow IPC format

spark.sql.adaptive.enabled=true
spark.sql.adaptive.coalescePartitions.enabled=true
spark.sql.optimizer.dynamicPartitionPruning.enabled=true

spark.sql.streaming.stateStore.providerClass=org.apache.spark.sql.execution.streaming.state.RocksDBStateStoreProvider