Appendix F: Spark Reference and Tuning Guide

Pinned to PySpark 3.5.3 on JDK 17 (Appendix A).

Before this appendix, Chapter 22's question: does your data actually need Spark? Below a few hundred gigabytes, DuckDB or Polars on one machine is usually faster and always simpler. Spark's value starts where one machine stops.

F.1 The Mental Model

job     one action (write, collect, count)
  stage a set of tasks with no shuffle between them
    task one partition of one stage -- the unit of parallelism

Stage boundaries are shuffle boundaries. Everything expensive in Spark is a shuffle, and reading a plan is mostly counting them.

narrow  map, filter, withColumn, union         no shuffle. Cheap.
wide    groupBy, join, distinct, repartition   shuffle. Expensive.

F.2 A Session, Configured

from pyspark.sql import SparkSession, functions as F

spark = (SparkSession.builder
    .appName("kestrel_sessionize")
    .config("spark.sql.adaptive.enabled", "true")                    # AQE: on
    .config("spark.sql.adaptive.coalescePartitions.enabled", "true")
    .config("spark.sql.adaptive.skewJoin.enabled", "true")
    .config("spark.sql.shuffle.partitions", "200")                   # AQE tunes down from here
    .config("spark.sql.files.maxPartitionBytes", "134217728")        # 128 MB
    .config("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
    .config("spark.sql.sources.partitionOverwriteMode", "dynamic")   # idempotent writes
    .getOrCreate())

partitionOverwriteMode=dynamic is the one that makes a rerun idempotent — insertInto with overwrite then replaces only the partitions present in the data, rather than the whole table.

F.3 Reading and Writing

df = (spark.read
        .schema(schema)                       # ALWAYS. Inference reads the data twice.
        .parquet("s3a://bronze/orders/"))

# partition pruning: the predicate must be on the partition column, unmodified
df.filter(F.col("event_date") == "2026-11-27")            # prunes
df.filter(F.to_date(F.col("event_ts")) == "2026-11-27")   # DOES NOT PRUNE -- ch1's bug

(df.write
   .mode("overwrite")
   .partitionBy("event_date")
   .option("compression", "zstd")
   .parquet("s3a://silver/orders/"))

Schema inference on JSON or CSV reads the whole dataset to infer, then again to load. On Kestrel's 4.19 TB of clickstream that is the difference between a job and an incident.

F.4 Reading the Plan

df.explain(True)          # parsed, analyzed, optimized, physical
df.explain("formatted")   # the readable one
df.explain("cost")        # with statistics, if they exist

Four things to look for:

FileScan parquet [order_id,net_cents]
  PartitionFilters: [(event_date = 2026-11-27)]      <- pruning worked
  PushedFilters:    [IsNotNull(customer_id)]         <- predicate pushdown worked
  ReadSchema:       struct<order_id:bigint,...>      <- projection worked

Exchange hashpartitioning(customer_id, 200)          <- a SHUFFLE. Count these.
BroadcastHashJoin                                    <- good: no shuffle for the join
SortMergeJoin                                        <- a shuffle on both sides

An empty PartitionFilters on a partitioned table is Chapter 1's $3,840 job.

F.5 Joins

from pyspark.sql.functions import broadcast

# broadcast: the single most effective optimization when it applies
big.join(broadcast(small), "customer_id", "left")

spark.conf.set("spark.sql.autoBroadcastJoinThreshold", 100 * 1024 * 1024)   # 100 MB
spark.conf.set("spark.sql.autoBroadcastJoinThreshold", -1)                  # disable

The hazard: broadcast() on a table that is not small produces a driver OOM, and "not small" means after decompression and after Java object overhead — often 3–5× the Parquet size.

F.6 Skew

The symptom: 199 tasks finish in seconds and one runs for forty minutes (ch4, ch21).

# 1. let AQE handle it -- usually enough on 3.x
spark.conf.set("spark.sql.adaptive.skewJoin.enabled", "true")
spark.conf.set("spark.sql.adaptive.skewJoin.skewedPartitionFactor", "5")

# 2. salt the hot key when AQE cannot
N = 16
left  = df.withColumn("salt", (F.rand() * N).cast("int"))
right = dim.withColumn("salt", F.explode(F.array(*[F.lit(i) for i in range(N)])))
out   = left.join(right, ["customer_id", "salt"])

# 3. find the skew first
df.groupBy("customer_id").count().orderBy(F.desc("count")).show(20)

High cardinality does not prevent skew. Kestrel has 1.9 million customers and one of them was 8.1% of the rows, which took a 22-minute job to 71.

F.7 Partitioning and File Size

df.rdd.getNumPartitions()

df.repartition(200)                    # full shuffle, even sizes
df.repartition("event_date")           # shuffle by column -- for a partitioned write
df.coalesce(10)                        # no shuffle, uneven sizes, only reduces

# the small-file fix on write
(df.repartition("event_date")
   .write.partitionBy("event_date").parquet(path))

Target 128 MB–1 GB per output file (ch9). Too small and you pay per-request and per-task overhead; too large and you lose parallelism and skipping.

coalesce(1) before a write is the standard way to accidentally make one task do everything.

F.8 Caching

df.cache()                              # MEMORY_AND_DISK
df.persist(StorageLevel.MEMORY_ONLY_SER)
df.unpersist()                          # do this; cached data is not free

Cache when a DataFrame is used more than once and is expensive to compute. Not otherwise: caching a DataFrame used once costs memory and buys nothing, and it is the most common misapplied Spark advice.

count() after cache() materializes it, which is sometimes what you want and is a full pass.

F.9 Memory

spark.executor.memory        heap
spark.executor.memoryOverhead  off-heap: shuffle buffers, Python workers
spark.driver.memory          the driver; a collect() lands here
spark.executor.cores         tasks per executor

Rules of thumb that survive contact:

4–5 cores per executor. More produces HDFS/S3 throughput contention; fewer wastes JVM overhead.

memoryOverhead at 10–15% — and PySpark needs more, because Python workers live outside the heap.

Never collect() a large DataFrame. Use write, or limit().collect(), or toLocalIterator().

OutOfMemoryError in the driver is almost always a collect() or an over-large broadcast. In an executor it is usually skew or too few partitions.

F.10 Structured Streaming

stream = (spark.readStream
    .format("kafka")
    .option("kafka.bootstrap.servers", os.environ["KAFKA_BROKERS"])
    .option("subscribe", "clickstream")
    .option("startingOffsets", "latest")
    .load())

out = (stream
    .select(F.from_json(F.col("value").cast("string"), schema).alias("e"))
    .select("e.*")
    .withWatermark("event_ts", "30 minutes")           # ch29: a business decision
    .groupBy(F.window("event_ts", "5 minutes"), "customer_id")
    .agg(F.count("*").alias("events")))

query = (out.writeStream
    .outputMode("append")
    .option("checkpointLocation", "s3a://scratch/ckpt/sessionize")   # REQUIRED
    .trigger(processingTime="1 minute")
    .start())

The checkpoint location is not optional and cannot be moved. It holds the offsets and the state; lose it and you reprocess from startingOffsets.

Output modes: append (final rows only, needs a watermark), update (changed rows), complete (the whole result — only for bounded aggregations).

Idle partitions stall the watermark (ch29 CS1): the global watermark is the minimum across partitions, so one silent partition holds every window open. Spark's equivalent of Flink's withIdleness is spark.sql.streaming.noDataMicroBatches.enabled plus a source-level setting — check your source's options, because the default is to stall.

F.11 Cost

On this book's frozen rate card, r6i.8xlarge at $2.400/node-hour:

24 nodes x 1.3 h  =  31.2 node-hours  =  $74.88     Kestrel's nightly sessionization
160 nodes x 10 h  = 1,600 node-hours  = $3,840.00   the same job with the CAST bug

Cost is node-hours, and node-hours track work. Autoscaling changes the wall clock and not the bill (ch36 CS2, ch38) — which also means it removes the latency signal you were unknowingly using to detect a cost regression.

Three things worth measuring before optimizing:

Is the cluster idle? A job that takes 90 minutes with 20 of them spent waiting on a source is a scheduling problem, not a Spark problem.

Is it one task? The Spark UI's stage view, sorted by duration, answers skew in ten seconds.

Is it reading what you think? The scan node's byte count.

F.12 Local Mode, for This Book

spark = (SparkSession.builder
    .master("local[*]")
    .config("spark.driver.memory", "4g")
    .config("spark.sql.shuffle.partitions", "8")     # 200 is absurd locally
    .config("spark.ui.enabled", "true")              # localhost:4040
    .getOrCreate())

spark.sql.shuffle.partitions=200 on a laptop creates 200 tasks over a few megabytes, and the overhead dominates completely. Set it to roughly your core count locally, and leave AQE to handle it in production.

The Spark UI at localhost:4040 is the tool, and it is the fastest way to learn what the plan actually did — the SQL tab shows the physical plan with row counts and timings per node.