Apache Spark Fundamentals: RDDs, DataFrames, Catalyst, and When Spark Still Wins

Apache Spark has spent more than a decade as the default choice for large-scale data processing. In that time the ecosystem has changed dramatically — DuckDB runs real analytical workloads on a laptop, Trino and Presto dominate ad-hoc querying, Flink owns streaming for many teams, and Polars has eaten a chunk of the single-node work Spark used to do. Yet Spark keeps growing. It is the compute engine under Databricks, AWS EMR, Google Dataproc, Azure Synapse, Snowflake’s Snowpark, and most on-prem lakehouse deployments. Knowing Spark well is still one of the highest-leverage skills a data engineer can have.

This post is a tour of the parts of Spark that matter: the execution model, how RDDs and DataFrames differ, what the Catalyst optimizer actually does, why shuffles are the enemy, how partitioning works, and where Spark genuinely beats lighter tools. I will assume you have seen a spark-submit command before and can read Scala or Python without flinching.

The mental model: a driver, some executors, and a DAG

A Spark application has one driver and many executors. The driver runs your code, holds the SparkSession, builds the query plan, and decides what to compute where. The executors run the work. Each executor is a JVM process with some number of cores and a chunk of memory. When you call collect(), data flows from executors back to the driver. When you call write.parquet(...), each executor writes its own partition.

The driver does not stream records through itself — it is an orchestrator, not a worker. This is why a Spark job can handle petabytes on a driver with 8 GB of memory, and also why collect() on a billion-row DataFrame will kill the driver with an OutOfMemoryError.

Spark is lazy. Calling df.filter(...), df.select(...), df.groupBy(...) doesn’t compute anything. It builds a logical plan. Computation happens only when an action runs: count, collect, show, write, take, foreach. Before that point, Spark can reorder, combine, and eliminate operations freely — which is exactly what the Catalyst optimizer does.

Jobs break into stages at shuffle boundaries. A stage is a sequence of transformations that can run inside a single executor without moving data between executors. Each stage has tasks, one per partition. A stage with 200 partitions runs 200 tasks. The scheduler assigns tasks to executor cores, and when one finishes it picks up another. This is why partition count matters so much: too few and you leave cores idle; too many and task overhead dominates.

RDDs, DataFrames, Datasets: what the difference actually is

The three APIs exist for historical reasons, and the answer to “which should I use” is almost always DataFrame / Dataset. But understanding why requires understanding what each gives up.

RDD (Resilient Distributed Dataset) is the original Spark primitive. An RDD is a distributed collection of objects — anything Java-serializable. You write rdd.map(x => x * 2) and Spark ships your closure to executors. The JVM executes your code on arbitrary objects.

This flexibility is the problem. Spark sees your lambda as a black box. It cannot push your filter into Parquet, it cannot prune columns, it cannot reorder your joins, it cannot skip rows using Parquet statistics. Every record has to be deserialized into a JVM object, passed through your closure, and reserialized.

DataFrame is the opposite. A DataFrame is a distributed collection of Row objects with a known schema, and the operations on it are not arbitrary functions but a relational algebra — select, filter, join, groupBy, agg. Because Spark knows what you are doing, it can optimize aggressively: Catalyst rewrites the plan, the physical planner picks join strategies, and the Tungsten engine processes data in a compact off-heap binary format instead of Java objects. For tabular workloads DataFrames are routinely 10× faster than equivalent RDD code.

Dataset is the Scala/Java-only typed version of DataFrame. Dataset[Person] is a DataFrame plus compile-time type safety. You get Catalyst optimization for the relational operations and type checking on your case class. In PySpark, DataFrame is all you have.

The rule: stay in DataFrame operations for as long as possible. Drop to RDDs only when you genuinely need a custom distributed algorithm the DataFrame API cannot express (graph algorithms, specific custom partitioners, very custom iterative algorithms). UDFs are usually the wrong answer — they force Spark to deserialize rows into Python/JVM objects, destroying the optimizations Catalyst built.

Catalyst: the optimizer doing most of the work

Catalyst is Spark’s query optimizer, and it is the reason DataFrame code is fast. When you write:

1
2
3
4
5
6
7

df = spark.read.parquet("s3://lake/events/")
result = (df
  .filter(df.event_date == "2026-04-19")
  .filter(df.user_country == "US")
  .groupBy("product_id")
  .agg(F.sum("revenue").alias("total")))
result.write.parquet("s3://lake/daily_revenue/")

Catalyst turns this into a pipeline of transformations:

1. Unresolved logical plan — the parse tree of your operations with column names as strings
2. Analyzed logical plan — column names resolved against the catalog, types inferred
3. Optimized logical plan — rule-based rewrites: combining filters, pushing predicates, constant folding, column pruning, reordering projections
4. Physical plans — multiple concrete execution strategies
5. Selected physical plan — chosen by cost-based optimizer using table statistics
6. RDDs of internal rows — the actual code that runs

The optimized logical plan in the example above collapses the two filters into one, pushes that filter into the Parquet scan (so Spark reads only files whose statistics indicate a match), and prunes columns to only the ones used (event_date, user_country, product_id, revenue). If the Parquet files are partitioned on event_date, Spark uses the partition filter to skip entire directories without opening any file.

You can see this at work with explain:

1
2
3

result.explain(mode="formatted")
result.explain(mode="cost")       # includes statistics
result.explain(mode="extended")   # shows all four plan stages

Reading the physical plan is the single most useful debugging skill in Spark. Things to look for:

• PushedFilters: [...] — filters that made it into the data source scan. Empty when your filter cannot push down (e.g., wrapped in a UDF).
• ReadSchema — the columns actually read. If you selected three columns but this shows thirty, column pruning failed.
• PartitionFilters — partition-column filters used to skip directories. This is where partitioning pays off.
• Exchange — a shuffle. Counting Exchanges is counting the expensive steps.
• BroadcastHashJoin vs SortMergeJoin — join strategy picked by the planner.

The Adaptive Query Execution revolution

Pre-Spark-3, the physical plan was chosen once at the start of a job using statistics that might be wildly wrong. In 3.0, Spark introduced Adaptive Query Execution (AQE), which re-plans the query at runtime using real shuffle statistics. AQE is on by default in recent versions, and it matters:

• Dynamic coalesce of shuffle partitions — the classic problem: you set spark.sql.shuffle.partitions=200 and end up with 200 tiny partitions after a filtering heavy query. AQE merges small partitions automatically.
• Dynamic join strategy — a join planned as SortMergeJoin gets re-planned as BroadcastHashJoin if one side ends up small enough after earlier stages.
• Dynamic skew handling — detects skewed partitions and splits them into multiple tasks.

Key settings:

spark.sql.adaptive.enabled=true
spark.sql.adaptive.coalescePartitions.enabled=true
spark.sql.adaptive.skewJoin.enabled=true
spark.sql.adaptive.advisoryPartitionSizeInBytes=128MB

If you are on Spark 3.2+, leave AQE on and tune advisoryPartitionSizeInBytes (usually 64–256 MB). AQE alone has retired a huge amount of manual tuning that 2.x-era engineers spent their careers doing.

Shuffles: the thing you are actually paying for

Almost every performance problem in Spark comes back to shuffles. A shuffle happens when a transformation needs data from multiple partitions on different executors to be reorganized — groupBy, join, distinct, repartition, window functions without a compatible partition.

During a shuffle, each executor writes its output to local disk, partitioned by the shuffle key. Downstream tasks fetch those blocks over the network. Shuffle data is written uncompressed (by default) and can easily be 10× the size of input data. This is why:

• A 100 GB job that fits fine in a map-only pipeline can OOM immediately when you add a groupBy.
• Network bandwidth is frequently the bottleneck, not CPU.
• Spot instance reclamations during shuffles trigger expensive re-computation.

The ways to fight shuffles:

Broadcast joins. If one side of a join is small (the default threshold is spark.sql.autoBroadcastJoinThreshold=10MB — raise it to 100–200 MB in most production jobs), Spark ships a copy of the small side to every executor and performs a hash join locally. No shuffle on either side.

1

large_df.join(F.broadcast(small_df), "user_id")

Force the broadcast when you know one side is small. The planner estimates table size from statistics; if your small side is a filtered-down version of a large table, the planner often underestimates.

Bucketing. If you write two tables bucketed on the same key with the same bucket count, joins between them can skip the shuffle entirely (bucketed tables are effectively pre-shuffled). Works well for star-schema fact/dimension joins where you always join on the same key.

Partition pruning for pre-partitioned sources. If your Parquet lake is partitioned by date, every query that filters on date scans only matching partitions. This is not the same as Spark partitions — it is storage-level partitioning. Get this right and you read 1 GB instead of 10 TB.

Avoid unnecessary repartition. Developers sprinkle repartition(200) around without thinking. Each one is a full shuffle. Use coalesce (which doesn’t shuffle) when reducing partition count. Use repartition only when you need an even distribution or a specific partitioning for a downstream join.

The partitioning mental model

Partitions show up in three distinct senses in Spark, and confusing them leads to a lot of bad tuning. Keep them separate:

• Input partitions: how Spark reads the source. Parquet files are typically split into 128 MB chunks per partition by default (spark.sql.files.maxPartitionBytes). Small-file problems happen here: a thousand 1 MB Parquet files become a thousand tasks doing almost no work each.
• Shuffle partitions: spark.sql.shuffle.partitions (default 200) — the number of partitions after a shuffle. AQE coalesces this adaptively; without AQE, 200 is almost never the right number for your data.
• Output partitions / files: how many files you write. If your last DataFrame has 2,000 partitions, you write 2,000 files per output directory. This matters for downstream consumers (too many files slows down metadata operations on S3/ADLS/GCS; too few hurts parallel read).

The general shape of a well-tuned Spark job:

1. Read with enough parallelism to keep all cores busy (aim for partitions ≈ 2–3× total core count).
2. Let AQE manage the middle of the pipeline.
3. Before writing, repartition or coalesce to a target file count (aim for 128 MB – 1 GB per output file).

For very large writes where you want both a sensible file count and partition column layout, use partitionBy with repartition:

1
2
3
4
5

(df
  .repartition("event_date", "country")  # align Spark partitions with output partitions
  .write
  .partitionBy("event_date", "country")
  .parquet("s3://lake/events/"))

Without the repartition, every Spark partition may contain rows for every (event_date, country) combination, and you end up writing tens of thousands of tiny files.

Memory and the OOM death spiral

Executor memory in Spark is carved up several ways. The simplified view:

• Reserved memory: 300 MB, untouchable.
• Unified memory region: spark.memory.fraction (default 0.6) of remaining heap. Used for execution (shuffles, sorts, joins) and storage (cache). The two regions borrow from each other.
• User memory: the other 40%. JVM overhead, UDFs, broadcast variables, driver-side structures on executors.

Common OOM causes in practice:

1. Skewed keys. One groupBy key has 99% of the rows. One task ends up with 100 GB of data trying to fit in an 8 GB executor. AQE skew join handling fixes many cases; for others, salt the key (append a random number, aggregate, then aggregate again).

2. collect on the driver. Obvious but still the #1 OOM. If you need to see sample data, use .show(20, truncate=False) or .limit(1000).toPandas().

3. Explosive joins. A many-to-many join silently produces row counts that dwarf inputs. Always sanity-check join cardinality before deploying.

4. Over-cached DataFrames. df.cache() keeps a DataFrame in memory, but if you cache everything, storage memory eats execution memory. Cache only DataFrames read multiple times, and unpersist() when done.

5. Python UDFs on wide rows. PySpark has to serialize each row to Python and back. For wide DataFrames this is both slow and memory-heavy. Use the vectorized Arrow-backed pandas UDFs (@F.pandas_udf) which operate on batches of columns at a time and are typically 10–100× faster.

When diagnosing memory issues, the Spark UI is irreplaceable. The Executors tab shows GC time per executor (more than 10% means memory pressure). The SQL tab shows per-stage data sizes and row counts. The Storage tab shows cached DataFrame sizes.

Join strategies and when each wins

Spark has several join strategies; understanding them lets you write queries that plan to the strategy you want.

Broadcast Hash Join — one side fits in memory; broadcast to all executors; hash lookup. Fastest for joins where one side is a dimension. Watch for planner underestimates; force with broadcast().

Sort Merge Join — default for large-large joins. Both sides shuffle on join key, sort, merge. Predictable memory usage. Scales to very large joins but pays the shuffle cost.

Shuffle Hash Join — both sides shuffle on key, but the smaller side is hashed instead of sorted. Faster than SMJ when the smaller side fits in memory but is too big to broadcast. Disabled by default; enable with spark.sql.join.preferSortMergeJoin=false.

Broadcast Nested Loop Join — cross join or non-equi join. Can explode catastrophically; usually a sign of a missing join condition.

If your join plan shows SortMergeJoin and one side is small, you are leaving performance on the floor. If it shows BroadcastNestedLoopJoin, check for a missing == condition.

Where Spark still beats DuckDB, Polars, and friends

It is fair to ask why anyone should still reach for Spark. DuckDB runs a few hundred gigabytes on a laptop. Polars eats single-node dataframes. Trino queries your lake interactively. The answer: scale, and a specific kind of workload.

Spark wins when:

• Your data is genuinely large (10 TB+) and needs horizontal scaling across a cluster.
• Your jobs include both SQL and imperative transformations (Scala/Python control flow between queries).
• You need a unified engine across batch, streaming, and ML — Spark’s streaming and MLlib share APIs with the batch DataFrame API.
• You work in a Databricks / EMR / Dataproc shop where Spark is the lingua franca and the ecosystem (Delta Lake, Unity Catalog, workflows) is built on it.
• You need a mature ecosystem of connectors — Spark has readers/writers for nearly every storage system imaginable.

Reach for something else when:

• Data fits on one large machine. DuckDB and Polars are dramatically simpler, faster for small data, and don’t require a cluster. A 500 GB single-node job in DuckDB often finishes before a Spark driver has finished JIT-warming.
• The workload is purely ad-hoc SQL over a lake. Trino, Presto, or Starburst are purpose-built for this and have better query latency.
• Streaming with tight tail latency. Flink beats Spark Structured Streaming on pure stream processing; Spark micro-batches inherently have higher minimum latency.

A healthy modern stack often has multiple engines: Spark for the heavy daily batch ETL, DuckDB or Polars for analyst notebooks, and Trino for interactive BI queries. These are complements, not competitors.

Running Spark in production: the concrete details

A few settings and practices that distinguish production Spark from notebook Spark:

Deploy mode: cluster for production (driver runs on the cluster, survives client disconnect). client for interactive work.

Resource sizing: a common heuristic is 4–5 cores per executor, 16–32 GB memory per executor. More cores per executor increases GC pressure; fewer wastes resources on JVM overhead. Total executors = available cluster cores / cores-per-executor.

Dynamic allocation: spark.dynamicAllocation.enabled=true scales executors up and down based on pending tasks. Pair with spark.shuffle.service.enabled=true so executors can be removed without losing their shuffle files. Essential for multi-tenant clusters.

Speculative execution: spark.speculation=true re-runs slow tasks on other executors. Valuable on clusters with hardware heterogeneity, harmful when the slow task is slow because of data skew (just runs the same skew twice).

Event logging: spark.eventLog.enabled=true writes event logs for the History Server. Non-negotiable for production — without it, debugging failed jobs after the fact is impossible.

Serialization: spark.serializer=org.apache.spark.serializer.KryoSerializer — faster and more compact than Java serialization for shuffle data. Register your custom classes with Kryo.

Shuffle service on Kubernetes: k8s deployments should use the external shuffle service or Spark 3.2+’s shuffle tracking to handle executor churn gracefully.

Observability: ship Spark metrics (JMX, graphite, Prometheus sink) to your monitoring stack. Per-stage duration, GC time, shuffle read/write, and executor count are the signals that catch problems before users notice.

Structured Streaming in one breath

Structured Streaming treats a stream as an unbounded table, and your batch DataFrame code mostly just works on streams. The key differences from pure batch:

• readStream / writeStream instead of read / write.
• A required trigger (processing time, continuous, available-now).
• A checkpoint location for fault tolerance.
• Stateful operations (aggregations, joins with streams) require watermarks to bound state growth.

The hidden cliff: state size. A streaming aggregation over user events with no watermark grows state forever. Define watermarks (.withWatermark("event_time", "1 hour")) and state eviction becomes possible. Without them, state hits the RocksDB state store backend’s limits and jobs slow to a crawl.

For most teams, a scheduled batch Spark job every 5–15 minutes is simpler and cheaper than Structured Streaming, and should be the first choice unless you truly need sub-minute latency.

Debugging: the workflow that actually works

When a Spark job is slow or failing:

1. Open the Spark UI. Every question below is answered there. Skip this step and you are guessing.
2. Stages tab: which stage is slow, and is the time uniform or skewed across tasks? Skewed = data skew. Uniformly slow = data size or compute.
3. SQL tab: click the query. The DAG shows exchanges and scan sizes. A 50 GB scan that should be 1 GB means pushdown failed.
4. Executors tab: GC time > 10%? Executors OOMing? Adjust memory.
5. Event timeline: tasks running in clean parallel blocks = healthy. Gaps = scheduler overhead or straggler waits.

For deeper issues: .explain(mode="extended") on the query, compare the analyzed plan to the optimized plan, and check the physical plan for the specific operators that are expensive.

A pragmatic rollout

For a team new to Spark, an order that usually works:

1. Get a managed Spark environment (Databricks, EMR Serverless, Glue, Dataproc). Do not start by building a YARN cluster.
2. Write DataFrame code. Stay out of RDDs. Treat UDFs as a last resort; use pandas UDFs when you need them.
3. Partition your data. At the lake level (Parquet partitioning) and at the Spark shuffle level (AQE does most of this).
4. Use Delta Lake or Iceberg for anything that needs updates, schema evolution, or time travel. Plain Parquet is fine for write-once pipelines.
5. Instrument everything. History Server, metrics sink, job-level ownership tags.
6. Migrate heavy batch jobs first, then anything ad-hoc. Don’t migrate things that fit on a single node unless you need unified tooling — DuckDB will serve you better.

Spark rewards investment. The first week feels heavy — the Spark UI is dense, the knobs are many, and the error messages can be brutal. But every hour spent understanding the execution model, the optimizer, and the shuffle pays off for years. Distributed computation is not going away, and Spark remains the best all-round tool in that space.