Apache Spark Internals: Catalyst, Tungsten, and the Shuffle Explained

Apache Spark presents a deceptively simple programming surface. A data engineer writes a few DataFrame transformations or a SQL query, calls an action, and a result appears. Underneath that surface sits a distributed query engine that parses the request into a logical plan, rewrites it through dozens of optimization rules, compiles fragments of it to Java bytecode, and coordinates hundreds of tasks across a cluster. When a job runs slowly, the cause is almost never the API call that was written. It lies in how the engine translated that call into stages, how much data crossed the network during a shuffle, and whether the optimizer chose a physical plan suited to the actual shape of the data.

This article traces that translation end to end. It assumes familiarity with running Spark jobs and concentrates on the execution internals that determine performance: the DAG scheduler and its stage boundaries, the Catalyst optimizer and its adaptive re-planning, the Tungsten engine and its code generation, the sort-based shuffle, and the join strategies that Spark selects. The version referenced throughout is the Spark 4.1.x line, current as of mid-2026, with Spark 4.1.2 released on 21 May 2026 and the 4.0 maintenance line still active through Spark 4.0.3 (Apache Spark news and downloads, as of 2026-07-14).

The Spark Execution Model: Jobs, Stages, and Tasks

A Spark application is coordinated by a single process called the driver, which holds the program logic, and a set of executor processes distributed across the cluster, which perform the actual computation on partitions of data. A partition is the unit of parallelism — a contiguous slice of a dataset that a single task processes. The driver builds the plan; the executors run it.

Spark evaluates transformations lazily. Operations such as map, filter, select, and join do not trigger computation; they extend a logical description of the work. Computation begins only when an action — for example count, collect, or write — is invoked. At that moment the driver submits a job, and the DAG scheduler translates the accumulated transformations into an execution plan structured as a directed acyclic graph (DAG) of stages.

Narrow and Wide Dependencies

The DAG scheduler divides a job into stages by examining the dependencies between partitions. A narrow dependency is one in which each parent partition contributes to at most one child partition. Transformations such as map and filter are narrow: a task can process its input partition and produce its output partition without consulting any other partition. Narrow dependencies are pipelined — chained together and executed within a single stage without materializing intermediate results.

A wide dependency is one in which a child partition depends on multiple parent partitions. Transformations such as groupByKey, reduceByKey, join, distinct, and repartition are wide, because a given output key may be assembled from records scattered across every input partition. A wide dependency cannot be pipelined; it requires data to be redistributed across the cluster so that all records sharing a key land in the same partition. That redistribution is the shuffle, and every shuffle defines a stage boundary (Apache Spark documentation; SparkInternals, as of 2026-07-14).

Figure 1: A job is split into stages at each shuffle. Narrow operators are pipelined inside a stage; each stage runs one task per partition.

Within a stage, the unit of execution is the task. Spark launches one task per partition, and tasks within a stage run in parallel across the available executor cores. The number of tasks in a stage therefore equals the number of partitions, which is why partition count directly governs both parallelism and per-task workload. A stage with too few partitions underuses the cluster; a stage with too many partitions incurs scheduling overhead for tasks that each process a trivial amount of data.

The Catalyst Optimizer and Adaptive Query Execution

Before any task runs, the query passes through Catalyst, Spark SQL’s query optimizer. Catalyst is an extensible optimizer built on functional tree-transformation constructs in Scala. It represents a query as a tree of nodes and rewrites that tree by applying rules — pattern-matching functions that transform one tree into an equivalent, cheaper tree. The DataFrame and SQL APIs both compile down to the same Catalyst representation, so an SQL query and its DataFrame equivalent are optimized identically.

The Four Phases

Catalyst structures optimization in four phases (Databricks, “Deep Dive into Spark SQL’s Catalyst Optimizer,” 2015-04-13):

1. Analysis. The parser produces an unresolved logical plan in which column and table names are still symbolic. The analyzer resolves these references against the catalog — the registry of tables, columns, and their types — turning the tree into an analyzed logical plan with fully typed attributes.
2. Logical optimization. A set of rule-based rewrites is applied to reduce work. Named optimizations include predicate pushdown (moving filters as close to the data source as possible), projection pruning (reading only the columns a query needs), constant folding (evaluating constant expressions once at plan time), and join reordering. The output is an optimized logical plan.
3. Physical planning. Catalyst generates one or more physical plans that specify concrete operators — which join algorithm to use, how to exchange data — and selects among them by cost. This is the only cost-based phase; all others are purely rule-based.
4. Code generation. The selected physical plan is compiled, in part, to JVM bytecode for execution, a step handled by the Tungsten engine and discussed in the next section.

The design of Catalyst was first described in the academic literature by Armbrust and colleagues in “Spark SQL: Relational Data Processing in Spark,” presented at SIGMOD 2015. That paper introduced the tree-transformation framework that remains the foundation of Spark SQL a decade later.

Figure 2: Catalyst’s four phases. Predicate pushdown into Parquet scans and projection pruning are rule-based; join-algorithm selection is the cost-based decision.

Predicate pushdown and projection pruning are the phases where storage format matters most. When Spark reads a columnar file, it can skip entire column chunks and row groups that a query does not touch, so the physical layout of the source determines how much of Catalyst’s logical optimization can be realized as reduced I/O. The mechanics of that pushdown are covered in the companion discussion of Apache Parquet and Apache Arrow internals, which explains how row-group statistics and column pruning let the scan read only the bytes a query requires.

Adaptive Query Execution

Catalyst’s cost-based decisions historically relied on statistics gathered before execution — table sizes, column histograms — which are often stale, missing, or wrong for intermediate results. A join planned as a sort-merge because both sides looked large may, after a filter, involve one tiny side. Adaptive Query Execution (AQE) addresses this by re-optimizing the plan at runtime using statistics measured from completed shuffles. AQE has been enabled by default since Spark 3.2.0 through the configuration flag spark.sql.adaptive.enabled (change tracked in SPARK-33679); it existed but was off by default in Spark 3.0 and 3.1 (Spark SQL Performance Tuning documentation, as of 2026-07-14).

AQE has three headline capabilities. It dynamically coalesces shuffle partitions, merging contiguous small partitions after a shuffle so that a fixed partition count no longer produces hundreds of tiny tasks. It switches join strategies, upgrading a planned sort-merge join to a broadcast hash join once a shuffle reveals that one side is small enough. And it handles skew by splitting oversized partitions in a sort-merge join at runtime. Because AQE inserts these decisions between stages — precisely where actual data sizes become known — it corrects the very estimation errors that make static planning brittle.

Figure 3: AQE re-optimizes at each shuffle boundary — coalescing partitions, switching join strategies, and splitting skewed partitions using measured statistics.

Tungsten and Whole-Stage Code Generation

Where Catalyst decides what to execute, Project Tungsten governs how efficiently each task runs on a single core. Tungsten began in Spark 1.6 as an initiative to bring Spark’s execution closer to bare-metal CPU and memory efficiency. Its stated goals were four: explicit off-heap memory management, cache-aware computation, code generation, and reduced virtual function calls (Databricks, “Project Tungsten,” 2015-04-28). The second generation of Tungsten, including whole-stage code generation, shipped in Spark 2.0 (Databricks, “Apache Spark as a Compiler,” 2016-05-23).

The UnsafeRow Binary Format

A conventional JVM program represents each row as a tree of Java objects, each carrying object headers, pointers, and padding, all managed by the garbage collector. For a dataset of billions of rows, that overhead dominates both memory consumption and CPU time spent in garbage collection. Tungsten replaces this with UnsafeRow, a compact binary layout in which a row is stored as a contiguous block of bytes. Because a Dataset has a known schema, Tungsten can lay each field out at a fixed offset and manage the backing memory off-heap — outside the region the garbage collector scans — sidestepping JVM object overhead and garbage-collection pressure entirely (The Internals of Spark SQL; Databricks, “Project Tungsten”).

Figure 4: UnsafeRow stores a row as a contiguous off-heap byte block with a null bitset, a fixed-length region, and a variable-length region, avoiding JVM object overhead.

Whole-Stage Code Generation

The classic way to execute a query plan is the Volcano iterator model, in which each operator implements a next() method and pulls rows one at a time from its child. This model is general but slow: every row crosses a virtual function call at every operator, intermediate values are boxed and materialized, and the CPU cannot keep data in registers. Whole-stage code generation (WSCG), tracked as SPARK-12795, collapses an entire chain of operators within a stage into a single JVM function. Instead of many operators each calling the next, Spark generates one tight loop that applies the whole chain of narrow operations to each row, eliminating virtual dispatch and keeping intermediate values in CPU registers rather than materializing rows. The generated Java source is compiled to bytecode at runtime by the Janino compiler (Databricks, “Apache Spark as a Compiler,” 2016-05-23; The Internals of Spark SQL).

Alongside WSCG, Spark 2.0 introduced vectorized columnar reads for Parquet (SPARK-12992), in which the reader decodes a batch of column values at a time rather than row by row, matching the columnar layout of the file to the batch-oriented execution engine. The scan and the compute path both operate on batches, which keeps the CPU’s instruction and data caches warm.

Figure 5: WSCG replaces a chain of Volcano-model operators, each with a virtual next() call, with a single generated loop that keeps intermediate values in registers.

Databricks demonstrated the effect of these techniques with a benchmark headlined as joining “one billion rows per second on a laptop.” That figure is a specific micro-benchmark result rather than a general guarantee, and it should be read as an illustration of what fused execution can achieve on a favorable workload, not a number any arbitrary job will reach.

The Shuffle and Data Skew

The shuffle is the operation that redistributes data across the cluster so that all records sharing a key reside on the same partition. It is triggered by every wide dependency and is, in most jobs, the single largest consumer of time and resources, because it combines serialization, disk writes, network transfer, and disk reads. Spark has used a sort-based shuffle as the default since Spark 1.2, when spark.shuffle.manager=sort replaced the earlier hash-based shuffle. The SortShuffleManager is now the only shuffle manager in vanilla Spark; it dispatches to three internal write paths — BypassMergeSortShuffleHandle, SerializedShuffleHandle, and BaseShuffleHandle — depending on the number of partitions and whether map-side aggregation is needed (apache/spark source, SortShuffleManager.scala, as of 2026-07-14).

Map Side and Reduce Side

A shuffle has two halves. On the map side, each task in the upstream stage processes its input partition and assigns every output record to a target partition, determined by hashing the key. Records accumulate in an in-memory structure — a PartitionedAppendOnlyMap when aggregation is required — grouped by target partition. When the structure exhausts its memory budget, the task spills a sorted run to disk, sorting with TimSort. At the end of the task, the spilled runs and any remaining in-memory records are merged into a single shuffle file with an accompanying index that marks where each partition’s bytes begin.

On the reduce side, each task in the downstream stage fetches the byte ranges destined for its partition from every map output across the cluster, then merges those streams on the fly using a min-heap. This on-the-fly merge distinguishes Spark’s shuffle from Hadoop MapReduce, which materializes a fully merged file on disk before the reduce begins. The volume of data fetched during this phase, multiplied by network latency, is what makes the shuffle expensive.

Figure 6: The sort-based shuffle. Map tasks partition and spill sorted runs; reduce tasks fetch their partition from every map output and merge on the fly.

The number of reduce-side partitions is governed by spark.sql.shuffle.partitions, whose default is 200. This fixed default is frequently wrong for a given data size: for a small result it creates 200 tiny tasks with more scheduling overhead than work, and for a large result it creates 200 oversized tasks that spill repeatedly to disk (Spark 4.1.2 SQL Performance Tuning documentation, as of 2026-07-14). AQE’s partition coalescing exists precisely to correct the small-partition case automatically, merging contiguous partitions toward a target size rather than requiring the value to be tuned by hand for every job.

Data Skew

Data skew occurs when records are distributed unevenly across keys, so that a few partitions hold far more data than the rest. Because a stage completes only when its slowest task finishes, a single oversized partition can hold up an entire job while the cluster sits idle. Skew is the most common reason a stage that processes a moderate dataset runs for an unexpectedly long time. It typically arises from a highly frequent key — a null join key, a default value, or a dominant category.

Two mitigations are standard. The first is salting: a random suffix is appended to the skewed key so that its records spread across many partitions, the join or aggregation is performed on the salted key, and the results are combined afterward. Salting is explicit and always available, but it requires rewriting the query. The second is AQE skew-join handling, enabled by default through spark.sql.adaptive.skewJoin.enabled, which detects an oversized partition at runtime and splits it into sub-partitions automatically, replicating the matching side of the join. A partition is treated as skewed when it exceeds spark.sql.adaptive.skewJoin.skewedPartitionThresholdInBytes — default 256 MB — and also exceeds a configurable factor times the median partition size.

Figure 7: A single hot key stalls a stage. Salting spreads the key manually; AQE splits the skewed partition automatically at the 256 MB threshold.

Join Strategies and Practical Performance Tuning

The join is where the cost of Catalyst’s physical planning, the shuffle, and AQE all converge. Spark chooses among three principal join algorithms, and the choice is governed largely by data size.

Broadcast, Sort-Merge, and Shuffle-Hash

A broadcast hash join is chosen when one side of the join is small — below spark.sql.autoBroadcastJoinThreshold, whose default is 10485760 bytes (10 MB). The small side is collected to the driver and broadcast to every executor, which builds a hash table from it in memory. The large side is then joined locally, partition by partition, with no shuffle of the large side at all. This is the cheapest join when it applies, because it avoids redistributing the large table.

A sort-merge join is the general default for two large inputs. Both sides are shuffle-partitioned on the join key and sorted, and the sorted partitions are then merged. It requires a full shuffle of both sides, which makes it expensive, but it scales to inputs far larger than memory because it never builds a full hash table.

A shuffle-hash join shuffles both sides on the join key but then builds an in-memory hash table on one side rather than sorting. AQE can convert a sort-merge join into a shuffle-hash join when all post-shuffle partitions are small enough to build hash tables locally, governed by spark.sql.adaptive.maxShuffledHashJoinLocalMapThreshold. AQE can also upgrade a planned sort-merge join to a broadcast hash join once the actual shuffle-output size of one side proves to be below the broadcast threshold (Spark 4.1.2 SQL Performance Tuning documentation, as of 2026-07-14).

Figure 8: Join-strategy selection. A side under the 10 MB threshold triggers a broadcast; otherwise Spark shuffles both sides and chooses shuffle-hash or sort-merge, with AQE able to switch at runtime.

The following table summarizes the trade-offs. A broadcast join is preferred whenever a side genuinely fits the threshold, because it eliminates the most expensive part of the join.

Configuration Defaults That Govern Performance

A small set of configuration values controls most of the behavior described above. The defaults below are those documented for the Spark 4.1.2 line (Spark SQL Performance Tuning documentation, as of 2026-07-14). Understanding them is more useful than memorizing them, because AQE now adjusts several of these dynamically.

Practical Tuning Priorities

Several practical conclusions follow directly from the mechanics above. The first is that reducing the amount of data read is the cheapest optimization available, because bytes never read are bytes never shuffled. Storing data in a columnar format with row-group statistics lets Catalyst’s predicate pushdown and projection pruning skip most of the file before any compute begins; the same principle underlies the design of modern open table formats such as Iceberg, Delta Lake, and Hudi, which add partition pruning and file-level statistics on top of the columnar layer.

The second is that the shuffle should be minimized, not merely tuned. Filtering before a join, pre-aggregating where possible, and enabling broadcast joins for small dimension tables all reduce the volume of shuffled data. Where a shuffle is unavoidable, AQE’s partition coalescing removes most of the need to hand-tune spark.sql.shuffle.partitions, though a very large job may still benefit from a higher explicit value so that individual tasks do not spill.

The third is that skew deserves direct attention. Because a stage is bounded by its slowest task, a job that appears to under-use the cluster is often waiting on one hot partition. The Spark UI’s stage view, which shows the distribution of task durations and shuffle read sizes, is the fastest way to confirm skew before applying salting or verifying that AQE skew handling engaged.

Spark occupies a specific position in the broader landscape of data systems. It is a distributed, disk-spilling engine designed for datasets that exceed the memory of any single machine, which is a different design point from the single-node vectorized engines examined in the comparison of lakehouse query engines such as Trino, StarRocks, and DuckDB. It is also predominantly a batch and micro-batch engine rather than a low-latency streaming system, a distinction developed in the discussion of streaming versus batch processing architectures. In practice Spark is frequently the compute layer beneath a SQL-first transformation workflow, and the models produced by tools such as dbt for transformation pipelines often compile down to the very Catalyst plans and shuffles described here.

Frequently Asked Questions

What is the difference between Catalyst and Tungsten in Spark?

Catalyst and Tungsten operate at different levels. Catalyst is the query optimizer: it decides what to execute by transforming a logical plan through analysis, rule-based logical optimization, and cost-based physical planning. Tungsten is the execution engine: it decides how efficiently each task runs on a CPU by generating fused machine-friendly code, storing rows in a compact off-heap binary format called UnsafeRow, and managing memory outside the garbage collector. Catalyst produces the plan; Tungsten executes it fast.

Why is the shuffle so expensive in Spark?

A shuffle redistributes data across the cluster so that all records sharing a key land on the same partition. It combines four costly operations: serializing records, spilling sorted runs to disk when memory is exhausted, transferring bytes across the network, and merging the fetched streams. Because it is triggered by every wide dependency and moves data proportional to dataset size, the shuffle is usually the dominant cost of a Spark job, which is why reducing the amount of shuffled data is the highest-value optimization.

Is Adaptive Query Execution enabled by default?

Yes. Adaptive Query Execution has been enabled by default since Spark 3.2.0 through the configuration flag spark.sql.adaptive.enabled. It was available but off by default in Spark 3.0 and 3.1. AQE re-optimizes a plan at runtime using statistics from completed shuffles, coalescing small partitions, switching a sort-merge join to a broadcast join when a side turns out small, and splitting skewed partitions automatically.

What does spark.sql.shuffle.partitions control, and why is 200 often wrong?

It sets the number of post-shuffle partitions for SQL and DataFrame joins and aggregations, and its default is 200. The value is fixed regardless of data size, so it is frequently a poor fit: for a small result it creates 200 tiny tasks whose scheduling overhead exceeds their work, and for a large result it creates 200 oversized tasks that spill to disk. AQE’s partition coalescing corrects the small-partition case automatically by merging contiguous partitions toward a target size, which reduces the need to tune the value by hand.

When does Spark use a broadcast join instead of a sort-merge join?

Spark chooses a broadcast hash join when one side of the join is smaller than spark.sql.autoBroadcastJoinThreshold, whose default is 10 MB. The small side is broadcast to every executor and the large side is joined locally without being shuffled, which is the cheapest join when it applies. Two large inputs default to a sort-merge join, which shuffles and sorts both sides. With AQE, a planned sort-merge join can be upgraded to a broadcast join at runtime once the actual size of one side is known to fall below the threshold.

Conclusion

Spark’s performance is best understood as the product of three cooperating layers. Catalyst decides the shape of the computation, rewriting a query through rule-based logical optimization and one cost-based physical-planning phase, and since Spark 3.2.0 it re-optimizes that plan at runtime through Adaptive Query Execution. Tungsten then executes each task efficiently, fusing operators into generated code that shipped in Spark 2.0 and storing rows in a compact off-heap format that avoids garbage-collection overhead. Between the stages sits the shuffle, the sort-based data-redistribution mechanism that has been the default since Spark 1.2 and that dominates the cost of most jobs.

For a practicing data engineer, the practical consequences are consistent. The largest gains come from reading less data, shuffling less data, and eliminating skew — in that order — because each addresses a cost that the engine cannot optimize away on its own. The configuration defaults matter less than the model behind them: once the boundary between narrow and wide dependencies, the role of the broadcast threshold, and the behavior of AQE are clear, the Spark UI and the physical plan become sufficient to diagnose almost any slow job. Reasoning about how the engine executes a query, rather than adjusting parameters in isolation, is what turns Spark performance work from guesswork into engineering.

References

• Apache Spark, “Spark SQL Performance Tuning” (official documentation, latest / 4.1.x). spark.apache.org/docs/latest/sql-performance-tuning.html
• Apache Spark, “News and Releases” (version verification, 4.1.2 / 4.0.3, mid-2026). spark.apache.org/news
• Databricks, “Deep Dive into Spark SQL’s Catalyst Optimizer” (2015-04-13). databricks.com/blog
• Databricks, “Project Tungsten: Bringing Apache Spark Closer to Bare Metal” (2015-04-28). databricks.com/blog
• Databricks, “Apache Spark as a Compiler: Joining a Billion Rows per Second on a Laptop” (2016-05-23). databricks.com/blog
• Armbrust et al., “Spark SQL: Relational Data Processing in Spark,” SIGMOD 2015 — the paper that introduced Catalyst.