Apache Spark: the complete cheat sheet

• 1. History and major releases
• 2. Architecture: the execution model
  • Worked example: counting jobs, stages and tasks
• 3. RDD, DataFrame and Dataset
• 4. RDD operations
• 5. spark-submit
• 6. Memory model
• 7. Reading and writing
• 8. The Catalyst optimizer
• 9. Joins
• 10. Shuffle and partitioning
• 11. Adaptive query execution
• 12. Caching and persistence
• 13. Spark SQL and ANSI mode
• 14. Structured Streaming
• 15. Useful functions and patterns
• 16. Diagnosing a slow job
• 17. Upgrading to Spark 4
• Conclusion
• References
• Trademarks

TL;DR

• A job splits into stages at shuffle boundaries, and a stage runs one task per partition. Most Spark performance questions turn into a question about partitions.
• ANSI mode is on by default in the Spark 4 line. spark.sql.ansi.enabled defaults to true unless the SPARK_ANSI_SQL_MODE environment variable is set to false, and it turns silent nulls into runtime errors.
• Adaptive query execution is on by default, and it coalesces partitions, splits skew and re-picks join strategies at runtime using real statistics.
• Defaults worth memorising: spark.executor.memory and spark.driver.memory are 1g, spark.executor.cores is 1, spark.sql.shuffle.partitions is 200, and broadcast joins kick in under 10 MB.
• The advisory shuffle partition size AQE aims for is 64 MB, not the 200-partition figure, so spark.sql.shuffle.partitions matters much less than it used to.

This is the Spark companion to the Hudi and Iceberg cheat sheets, in the same shape: every row gives you the concept, a literal example, and the sentence that tells you whether you want it.

Images captioned Source: Apache Spark documentation come from docs/img in the Apache Spark source tree, and those captioned Adapted from the Apache Spark documentation are my own redrawings of figures there, © The Apache Software Foundation, used under the Apache License 2.0; the rest are my own. Where a default changed in Spark 4, it is called out, because those are the ones that break a job on upgrade.

1. History and major releases

The 3.5, 4.0 and 4.1 lines are still receiving maintenance releases alongside 4.2, so “latest” and “the version you should be on” are not always the same question. Check which line your platform and connectors support first.

Why it displaced MapReduce. MapReduce gave you exactly two steps, map then reduce, and wrote to disk between every one of them. A multi-step pipeline became a chain of jobs, each paying a full write and read, and an iterative algorithm paid that on every pass.

flowchart TB
  subgraph MR["MapReduce: disk between every step"]
    direction LR
    M1["read HDFS"] --> M2["map"]
    M2 --> M3["write disk"]
    M3 --> M4["shuffle + sort"]
    M4 --> M5["reduce"]
    M5 --> M6["write HDFS"]
    M6 --> M7["next job reads<br/>it all back"]
  end

  subgraph SP["Spark: one DAG, memory between steps"]
    direction LR
    S1["read once"] --> S2["map, filter, join<br/>chained in memory"]
    S2 --> S3["shuffle only where<br/>the DAG needs one"]
    S3 --> S4["more in-memory steps"]
    S4 --> S5["write once"]
  end

  MR ~~~ SP

Two changes did it. Spark keeps intermediate results in memory instead of on disk, and it models the whole pipeline as one DAG rather than a fixed pair of steps, so it can see across the chain and shuffle only where the data genuinely has to move. The DAG is the more durable idea of the two: it is what lets Catalyst and adaptive execution reason about the query at all.

The through-line is that each major version moved work from the user to the engine. 1.x asked you to write RDD code and tune it by hand. 2.x gave the optimizer a schema to work with. 3.x let it re-plan from runtime statistics. 4.x turned the remaining silent-correctness footguns into errors.

Only the 4.0 row contains a change likely to break an existing job, and it is ANSI mode. The rest are additive.

2. Architecture: the execution model

Adapted from the Apache Spark documentation.

flowchart LR
  subgraph ST1["stage 1: narrow, no shuffle"]
    direction LR
    R1["read trips"] --> F["filter, select"]
  end

  subgraph ST2["stage 2: after the shuffle"]
    direction LR
    AG["aggregate"] --> W["write, the action"]
  end

  F -->|"shuffle<br/>groupBy city_id"| AG

Everything between two shuffles is one stage, and a stage runs one task per partition. That is the whole scheduling model, and it is why “how many partitions” is the question behind most tuning.

Transformations are lazy and actions are eager. filter, select, join and withColumn add to a plan; count, collect, show and write execute it.

Narrow against wide is the distinction that predicts everything else. A narrow transformation needs only the partition in front of it, so it runs in place. A wide one needs rows from other partitions, so it forces a shuffle, and a shuffle is what ends a stage.

flowchart TB
  subgraph NW["narrow: each output partition reads one input partition"]
    direction LR
    N1["p1"] --> N1b["p1'"]
    N2["p2"] --> N2b["p2'"]
    N3["p3"] --> N3b["p3'"]
  end

  subgraph WD["wide: each output partition reads many input partitions"]
    direction LR
    W1["p1"] --> X1["p1'"]
    W1 --> X2["p2'"]
    W2["p2"] --> X1
    W2 --> X2
    W3["p3"] --> X1
    W3 --> X2
  end

  NW ~~~ WD

Two consequences worth holding on to. Counting the wide transformations in your job tells you how many stages it will have, which is the fastest way to predict a plan before running it. And a broadcast join is valuable precisely because it turns a wide transformation into a narrow one: the small side goes to every executor, so no rows have to move.

coalesce sits on the narrow side, which is exactly why it can starve upstream parallelism: avoiding the shuffle means the reduced partition count propagates backwards up the DAG. That is why a typo in a filter often surfaces at the write, several lines later.

Worked example: counting jobs, stages and tasks

Reading the counts off a piece of code is the skill this all builds to. Take this, on a table that reads as 8 input partitions:

from pyspark.sql.functions import col, when

trips = spark.read.parquet("s3a://lakehouse-prod/warehouse/trips")
# input partitions come from split size, not file count: check, do not assume
print(trips.rdd.getNumPartitions())                                  # say it is 8

by_city = (trips
    .filter(col("fare_amount") > 50)                                 # narrow
    .withColumn("fare_band",
                when(col("fare_amount") > 100, "high").otherwise("mid"))  # narrow
    .groupBy("city_id").count())                                     # WIDE: shuffle

by_city.write.parquet("s3a://lakehouse-prod/warehouse/trips_by_city")  # action

The rules underneath are short enough to memorise. Jobs equal actions. Stages equal wide transformations plus one. Tasks equal partitions, per stage. Narrow transformations never add a stage; they fuse into the tasks of the stage they sit in, which is also what whole-stage code generation collapses into a single generated method.

Watch for two things that break the arithmetic. Some operations hide an extra job: show() may run one job to peek and another to finish, and any operation that needs to sample or infer, such as reading JSON without a schema or a sortBy computing range boundaries, runs its own job first. And with adaptive execution on, the 200 in stage 2 is an upper bound rather than the number you will see, because coalescing runs after the shuffle statistics arrive.

3. RDD, DataFrame and Dataset

In Python, Dataset does not exist as a separate API: DataFrame is the typed and untyped API at once, because Python has no compile-time type checking to offer. In Scala, DataFrame is literally the alias Dataset[Row].

The practical rule is to stay in DataFrame or SQL. Every time you drop into an RDD or a typed lambda, Catalyst loses visibility into what you are doing and can no longer reorder, prune or generate code across that step. Reach for an RDD when you genuinely need something the structured API cannot express, not by habit.

4. RDD operations

Two kinds, and the distinction is the whole execution model: transformations build a plan lazily, actions run it.

Two habits worth forming. Prefer reduceByKey over groupByKey, because it combines on the map side and shuffles far less. And treat every action that returns data to the driver (collect, take, countByKey) as a driver memory question, not a cluster one.

5. spark-submit

SparkPi ships with every distribution, so this runs as pasted on any cluster and is the fastest way to prove a submit configuration works before pointing it at your own jar:

spark-submit \
  --master yarn \
  --deploy-mode cluster \
  --name spark-pi \
  --class org.apache.spark.examples.SparkPi \
  --num-executors 20 \
  --executor-cores 4 \
  --executor-memory 16g \
  --driver-memory 8g \
  --conf spark.sql.shuffle.partitions=400 \
  $SPARK_HOME/examples/jars/spark-examples_2.13-4.1.3.jar \
  1000

The trailing 1000 is the application’s own argument, not Spark’s: SparkPi reads it as the number of slices to parallelise across.

Dynamic allocation is deliberately absent above, because it and --num-executors are mutually exclusive: turning on spark.dynamicAllocation.enabled makes the fixed count irrelevant. If you want it, drop --num-executors, set spark.dynamicAllocation.minExecutors and maxExecutors, and give it either an external shuffle service or spark.dynamicAllocation.shuffleTracking.enabled.

Cluster mode against client mode decides where the driver runs, and that decides more than it sounds like.

flowchart TB
  subgraph CL["cluster mode: driver runs inside the cluster"]
    direction LR
    C1["spark-submit<br/>from a gateway"] --> C2["cluster manager"]
    C2 --> C3["driver<br/>in a container"]
    C3 --> C4["executors"]
    C1 -.->|"can disconnect"| C3
  end

  subgraph CI["client mode: driver runs where you submitted"]
    direction LR
    I1["spark-submit<br/>driver lives here"] --> I2["cluster manager"]
    I2 --> I3["executors"]
    I1 -->|"driver traffic"| I3
    I1 -.->|"kill this and<br/>the job dies"| I3
  end

  CL ~~~ CI

The one that bites: an interactive session is client mode, so the driver is on the machine you are sitting at. A collect() that works in production can exhaust the driver heap on your laptop, and closing the laptop kills the job.

On Kubernetes the same roles map onto pods, with the driver pod creating and owning the executor pods:

Adapted from the Apache Spark documentation.

Note: Arguments after the application jar go to your main, not to Spark. Anything Spark-facing must come before the jar. Options after it are passed to your application instead.

6. Memory model

flowchart TB
  C["container the cluster manager allocates"] --> H["spark.executor.memory<br/>JVM heap"]
  C --> O["spark.executor.memoryOverhead<br/>off-heap, 384m or 10 percent"]
  H --> RES["reserved<br/>300 MB"]
  H --> UP["unified pool<br/>memory.fraction 0.6"]
  H --> USR["user memory<br/>the remaining 0.4"]
  UP --> EX["execution<br/>shuffles, joins, sorts"]
  UP --> ST["storage<br/>cached blocks<br/>floor at storageFraction 0.5"]
  %% execution can evict storage down to the floor; storage can never evict execution
  EX -.->|"evicts"| ST

The container your cluster manager sees is spark.executor.memory plus overhead, so a 16g executor with the default factor asks for about 17.6g. When a job is killed for exceeding its container limit and the heap looked fine, the overhead is the first thing to check.

The bands are drawn in proportion to each other, except reserved memory, which is 1.8% of the heap and is given a minimum height so it can be labelled.

Worked example, --executor-memory 16g on the defaults. The formula is in UnifiedMemoryManager: subtract the reserved 300 MB, then take spark.memory.fraction of what is left.

Three things fall out of that arithmetic. Only about 9.4 GB of a 16 GB executor is available for shuffles, joins, sorts and cache combined, which is less than the container is given. The container is 10 percent larger than the heap you asked for, so a cluster sized to the nearest gigabyte will fail to schedule. And with --executor-cores 4, those 9650 MB are shared by four concurrent tasks, so one task spilling is often a cores-per-executor problem rather than a memory-per-executor one.

The minimum is enforced: spark.executor.memory below 450 MB is rejected, because the reserved 300 MB times 1.5 is the floor UnifiedMemoryManager requires.

Execution memory (shuffles, joins, sorts) and storage memory (cached blocks) share one pool and borrow from each other. Execution can evict cached blocks down to the storage floor; storage can never evict execution. That is why caching a large DataFrame can quietly make a shuffle-heavy stage spill.

7. Reading and writing

# The single most useful read option for JDBC: parallelism
trips = (spark.read.format("jdbc")
    .option("url", "jdbc:postgresql://db.internal:5432/rides")
    .option("dbtable", "public.trips")
    .option("user", "etl")
    # partitionColumn must be numeric, date or timestamp. A string key will not do.
    .option("partitionColumn", "trip_seq")
    .option("lowerBound", "1")
    .option("upperBound", "40000000")
    .option("numPartitions", "16")
    .load())

Note: spark.sql.sources.partitionOverwriteMode defaults to STATIC, so INSERT OVERWRITE on a partitioned table replaces every partition, not just the ones in your DataFrame. Set it to dynamic for partition-scoped overwrites.

8. The Catalyst optimizer

Catalyst is why a DataFrame beats hand-written RDD code: you declare what you want and it decides how. Every query passes through four phases, each defined as rules over a tree.

flowchart TB
  SQL["SQL query<br/>or DataFrame code"] --> PARSE["parser"]
  PARSE --> UL["unresolved logical plan<br/><i>relations and columns are just names</i>"]

  UL --> AN["<b>Analyzer</b><br/>resolve against the catalog"]
  AN --> AL["analyzed logical plan<br/><i>every column typed and bound</i>"]
  AN -.->|"name not found"| ERR["AnalysisException"]
  CAT[("catalog<br/>tables, views,<br/>functions")] -.-> AN

  AL --> OPT["<b>Optimizer</b><br/>rule batches to fixed point"]
  OPT --> OL["optimized logical plan"]
  OPT -.- R1["predicate pushdown"]
  OPT -.- R2["column pruning"]
  OPT -.- R3["constant folding"]
  OPT -.- R4["limit pushdown"]
  OPT -.- R5["boolean simplification"]

  OL --> PLAN["<b>SparkPlanner</b><br/>strategies produce candidates"]
  PLAN --> PP["one or more<br/>physical plans"]
  STATS[("statistics<br/>sizes, row counts")] -.-> PLAN
  PP --> PICK["cost model selects one"]
  PICK --> SP["selected physical plan<br/><i>join strategy now fixed</i>"]

  SP --> CG["<b>WholeStageCodegenExec</b><br/>fuse operators into one Java method"]
  CG --> RDD["RDD[InternalRow]<br/>executed on the cluster"]

  RDD -.->|"real runtime statistics"| AQE["<b>AdaptiveSparkPlanExec</b><br/>re-plan the next stage"]
  AQE -.-> PLAN

The dashed loop at the bottom is the part that changed Spark 3 onwards: after a stage finishes, adaptive execution feeds its real statistics back into planning, so the join strategy and partition count for the next stage are chosen from measured sizes rather than estimates.

-- See every phase for a query, which is the fastest way to learn what Catalyst did
EXPLAIN EXTENDED
SELECT c.city_name, count(*)
FROM trips t JOIN cities c ON t.city_id = c.city_id
WHERE t.fare_amount > 50
GROUP BY c.city_name;

The reason this matters in practice: Catalyst can only optimise what it can see. A SQL expression or a DataFrame column expression is a tree it can rewrite. A Python UDF or an RDD lambda is an opaque function it must call as-is, which blocks pushdown across that point. That is the real cost of a UDF, and it is usually larger than the cost of the function itself.

9. Joins

SELECT /*+ BROADCAST(c) */ t.trip_id, t.fare_amount, c.city_name
FROM trips t JOIN cities c ON t.city_id = c.city_id;

The join that hurts is the one where both sides are large and the key is skewed. AQE handles the common case automatically; see the skew settings below.

10. Shuffle and partitioning

The coalesce trap is worth spelling out: df.repartition(1000).coalesce(1) does not give you 1000-way parallelism then one file. Because coalesce avoids the shuffle, it pushes the narrow partition count up the DAG, and the upstream work runs with one task. Use repartition(1) when you genuinely want the shuffle.

11. Adaptive query execution

AQE re-optimises the plan mid-flight using statistics from completed stages, which is why it beats anything you can set by hand ahead of time.

flowchart LR
  P["logical plan"] --> ST1["run stage 1"]
  ST1 --> STATS["real statistics<br/>partition sizes, row counts"]
  STATS --> RE["re-optimise"]
  RE --> C1["coalesce small<br/>partitions"]
  RE --> C2["split skewed<br/>partitions"]
  RE --> C3["switch sort merge<br/>to broadcast"]
  C1 --> ST2["run stage 2"]
  C2 --> ST2
  C3 --> ST2

Both skew conditions must hold: a partition is treated as skewed only when it is larger than 5 times the median and larger than 256 MB. On a job whose partitions are all under 256 MB, skew handling never fires no matter how uneven they are. That is worth checking before concluding skew handling is broken.

12. Caching and persistence

Note: In Python, objects are always serialized with Pickle, so the _SER levels do not exist as a separate choice. The levels available from PySpark are MEMORY_ONLY, MEMORY_ONLY_2, MEMORY_AND_DISK, MEMORY_AND_DISK_2, DISK_ONLY, DISK_ONLY_2 and DISK_ONLY_3.

The two defaults differ and it catches people: rdd.cache() is MEMORY_ONLY, so anything that does not fit is silently recomputed every time. df.cache() is MEMORY_AND_DISK, so it spills instead. If an RDD cache seems to do nothing, that asymmetry is usually why.

The documentation’s own selection order is worth following: stay on the default if the data fits; move to a serialized level to save space before you move to disk; and only spill to disk when recomputing would be more expensive than reading it back, which is the case when the computation was expensive or filtered a lot of data away.

Cache when a DataFrame is used more than once and producing it was expensive. Caching something read once makes the job slower, because you pay the write and give up memory that execution wanted.

13. Spark SQL and ANSI mode

The biggest behavioural change in the Spark 4 line is ANSI mode. spark.sql.ansi.enabled now defaults to true, and the default is computed as “true unless the SPARK_ANSI_SQL_MODE environment variable is set to false”.

Treat an ANSI failure on upgrade as a finding rather than a blocker: the error is usually pointing at data that was silently becoming NULL before. Reach for try_cast and try_divide at the specific expression before turning the flag off everywhere.

14. Structured Streaming

The model to hold in your head is that a stream is an unbounded table, with each arriving batch appended as new rows:

Adapted from the Apache Spark documentation.

A query over that table produces a result table, recomputed incrementally each trigger, and the output mode decides which of its rows get written out:

Adapted from the Apache Spark documentation.

from pyspark.sql.functions import col, from_json, window, count
from pyspark.sql.types import StructType, StringType, TimestampType, DoubleType

schema = (StructType()
    .add("trip_id", StringType())
    .add("city_id", StringType())
    .add("fare_amount", DoubleType())
    .add("started_at", TimestampType()))

events = (spark.readStream.format("kafka")
    .option("kafka.bootstrap.servers", "kafka-broker1:9092")
    .option("subscribe", "trips")
    .option("startingOffsets", "latest")
    .load()
    # Kafka's own `timestamp` is broker ingest time. Event time comes from the
    # payload, which is what a watermark should be based on.
    .select(from_json(col("value").cast("string"), schema).alias("t"))
    .select("t.*"))

# A plain pass-through sink: no state, so no watermark is needed.
(events.writeStream
    .format("parquet")
    .option("checkpointLocation", "s3a://lakehouse-prod/checkpoints/trips_raw")
    .option("path", "s3a://lakehouse-prod/warehouse/trips_raw")
    .trigger(processingTime="1 minute")
    .outputMode("append")
    .start())

# A stateful aggregation, which is where the watermark actually does work:
# it finalises windows and lets Spark drop their state.
(events
    .withWatermark("started_at", "10 minutes")
    .groupBy(window(col("started_at"), "5 minutes"), col("city_id"))
    .agg(count("*").alias("trips"))
    .writeStream
    .format("parquet")
    .option("checkpointLocation", "s3a://lakehouse-prod/checkpoints/trips_by_window")
    .option("path", "s3a://lakehouse-prod/warehouse/trips_by_window")
    .outputMode("append")
    .start())

The two queries above are deliberately different. A watermark on the first would be inert: it only has an effect on stateful operations, which a pass-through sink is not. Each query also gets its own checkpoint location, which is a hard requirement rather than a convention.

Event-time windows come in three shapes, and the choice changes how many windows a single record lands in:

Adapted from the Apache Spark documentation.

The watermark is what lets Spark finalise a window and drop its state. Anything arriving behind the watermark is too late to be counted:

Adapted from the Apache Spark documentation.

Without a watermark, a streaming aggregation keeps state forever and the job degrades over days rather than failing outright. The watermark is what lets Spark drop state it will never need again.

15. Useful functions and patterns

from pyspark.sql import Window
from pyspark.sql.functions import row_number, desc, col

# Latest row per trip, which is the shape most CDC dedup takes
latest = (trips
    .withColumn("rn", row_number().over(
        Window.partitionBy("trip_id").orderBy(desc("updated_at"))))
    .where(col("rn") == 1)
    .drop("rn"))

16. Diagnosing a slow job

The stages page is where skew and spill show up, because it gives you the task duration percentiles rather than just an average:

Source: Apache Spark documentation.

And the SQL tab lists each query with its duration and plan:

Source: Apache Spark documentation.

Opening one gives you the physical plan annotated with row counts per node, which answers “did my filter push down” and “which side got broadcast” directly:

Source: Apache Spark documentation.

The Spark UI SQL tab is the most useful page and the least used. It shows the physical plan with row counts per node, which answers “did my filter push down” and “which side got broadcast” directly, rather than by inference.

17. Upgrading to Spark 4

Section 1 lists what arrived. This is what breaks, which is a shorter and more useful list.

-- The one-line rehearsal: run your suite on the OLD runtime with ANSI on.
-- Every failure here is a Spark 4 failure you can fix before upgrading.
SET spark.sql.ansi.enabled = true;

Doing that first splits the upgrade in two. ANSI failures are data and expression problems you can fix on the version you already run; everything else is runtime and dependency work. Meeting both at once is what makes a Spark 4 upgrade feel hard.

Conclusion

Most Spark tuning reduces to three questions: how many partitions, how large, and is the data evenly spread across them. Executor sizing and memory fractions matter, but they matter because they decide how much work a single task can hold before it spills.

Adaptive query execution has absorbed most of what used to be manual tuning. It coalesces small partitions, splits skew and re-picks join strategies from real statistics, which means spark.sql.shuffle.partitions is far less load-bearing than the tuning guides of a few years ago suggest. Set it roughly and let AQE adjust.

The one thing to plan for deliberately on the 4.x line is ANSI mode. It is a good default and it will surface data problems that the 3.x line hid, so give it a dedicated test pass rather than meeting it during an upgrade.

References

• Spark configuration reference for every property, its default and the version it appeared in
• SQL performance tuning for adaptive query execution, join hints and partition coalescing
• Structured Streaming programming guide for triggers, watermarks and output modes
• ANSI compliance for the full list of behaviours the Spark 4 default changes
• Apache Hudi on Spark: the complete cheat sheet for the lakehouse layer on top of this

Trademarks

Apache Spark, Apache Hudi, Apache Iceberg, Apache Kafka, Apache Parquet, Apache Hadoop, Apache Hive and Apache are either registered trademarks or trademarks of The Apache Software Foundation in the United States and other countries.