DEV Community: Aniketh Deshpande

What Actually Happens When You Call spark.read? One Line of Python, a Thousand Tasks

Aniketh Deshpande — Thu, 01 Oct 2026 15:43:21 +0000

TL;DR: spark.read.parquet(path) reads almost nothing. The real work starts at the first action, when Spark
turns your DataFrame into four plans, slices your files into tasks, ships those tasks to executors and, depending on
what you asked for, shuffles data across the network. This article follows one DataFrame through all of that, in
five scenarios: a plain read, a read with filters, a sort, a join, and finally the whole pipeline end to end.

⏱️ 30-minute read. Every scenario comes with the code, the explain() output, the Spark UI you'd see, and a
diagram of what the cluster is doing. All facts were checked against Spark 3.5 and Spark 4.0; defaults are
the same in both unless noted.

Prologue: Maya's first ticket
The setup: one cluster, one dataset
Chapter 1: The line that does (almost) nothing
Chapter 2: The first action, or the anatomy of a job
Chapter 3: Reading less: filters, pruning and pushdown
Chapter 4: Putting things in order: sort and shuffle
Chapter 5: Meeting other tables: joins
Chapter 6: The whole pipeline, end to end
Cheat sheets
Try it on your laptop
Further reading

Prologue: Maya's first ticket

It's Monday. Maya has been a data engineer at RideHub for exactly four days, and she has just been handed her first real ticket:

FIN-2231: Finance needs the total fares for March, counting only trips over $100, broken down by driver and pickup zone. Top 10 drivers. By Thursday.

She knows PySpark. She writes six lines:

trips   = spark.read.parquet("s3a://ridehub/trips/")
drivers = spark.read.parquet("s3a://ridehub/drivers/")
zones   = spark.read.parquet("s3a://ridehub/zones/")

report = (trips
    .filter((F.year("pickup_ts") == 2026) & (F.month("pickup_ts") == 3))
    .filter(F.col("fare") > 100)
    .join(drivers, "driver_id")
    .join(zones, "zone_id")
    .groupBy("driver_name", "zone_name")
    .agg(F.sum("fare").alias("total_fare"), F.count("*").alias("trips"))
    .orderBy(F.desc("total_fare"))
    .limit(10))

report.write.mode("overwrite").parquet("s3a://ridehub/reports/fin-2231/")

She hits run. Fourteen minutes later, the job finishes. The numbers look right. She's about to close the ticket when Ravi, the staff engineer who sits across from her, rolls his chair over.

"Fourteen minutes for a month of data?" he says. "Do you know what those six lines actually did?"

"They... read the data?"

"That's the one thing they almost didn't do." He pulls up the Spark UI. There are five jobs, one of them with 16,384 tasks, and half a terabyte read from S3 to produce a report with ten rows. "Let's walk through it. By Thursday you'll be able to read this screen like a story."

This article is that walk. We'll take Maya's pipeline apart one operation at a time, and for each one answer the same question: what does the cluster actually do when this line runs?

The setup: one cluster, one dataset

So the numbers in the story mean something, here is what Maya is working with.

The cluster

 ┌─────────────────────────────┐
 │  Driver                     │   8 GB RAM, 2 cores
 │  (runs Maya's Python +      │   Holds the SparkSession, the plans,
 │   the JVM that plans jobs)  │   the scheduler and the Spark UI
 └──────────────┬──────────────┘
                │  schedules tasks over the network
   ┌────────────┼────────────┬─────────── ... ──────────┐
   ▼            ▼            ▼                          ▼
 ┌──────┐    ┌──────┐    ┌──────┐                   ┌──────┐
 │Exec 1│    │Exec 2│    │Exec 3│       ...         │Exec10│   10 executors
 │4 core│    │4 core│    │4 core│                   │4 core│   16 GB RAM each
 │16 GB │    │16 GB │    │16 GB │                   │16 GB │   = 40 task slots
 └──────┘    └──────┘    └──────┘                   └──────┘

The data lives on object storage, as Parquet, partitioned by day:

s3a://ridehub/trips/                      2.0 TB, 3.2 billion rows, 18 columns
├── dt=2025-10-01/
│   ├── part-00000-....snappy.parquet     ~128 MB each
│   ├── part-00001-....snappy.parquet
│   └── ...  (~44 files per day)
├── dt=2025-10-02/
├── ...
└── dt=2026-09-30/                        365 days ≈ 16,000 files

s3a://ridehub/drivers/   50 MB,  600,000 rows       (driver_id, driver_name, licence, payout, ...)
s3a://ridehub/zones/     20 KB,  263 rows           (zone_id, zone_name, borough)

The trips schema, trimmed to the columns the story uses:

column	type	note
`trip_id`	string
`driver_id`	long	joins to `drivers`
`zone_id`	int	pickup zone, joins to `zones`
`pickup_ts`	timestamp
`fare`	double
`dt`	date	partition column: not stored in the files, it comes from the directory name

And the session, with every setting at its default so the defaults are what we learn:

from pyspark.sql import SparkSession, functions as F

spark = (SparkSession.builder
    .appName("fin-2231")
    .getOrCreate())

💡 All the explain() output below is lightly trimmed (expression IDs like #12 and plan IDs removed) so it fits on
a screen. The shape is exactly what you'll see.

Chapter 1: The line that does (almost) nothing

trips = spark.read.parquet("s3a://ridehub/trips/")

Maya runs this in a notebook and it returns in 1.8 seconds. For a 2 TB dataset. Her first instinct is that Spark is fast. Ravi's correction: Spark hasn't read any trips yet. It did do three things, though, and one of them is a real Spark job.

1.1 Resolving the data source

spark.read gives you a DataFrameReader. .parquet(path) asks it to build a data source. Spark has two data source APIs. The newer one (DataSource V2) is used for Delta, Iceberg, Kafka and most connectors; but for the built-in file formats, Spark still defaults to the original V1 path, because spark.sql.sources.useV1SourceList lists parquet, orc, csv, json, text, avro. So Parquet resolves to ParquetFileFormat wrapped in a HadoopFsRelation. This matters later, because V1 and V2 differ in exactly which optimizations they support.

1.2 Listing the files, which is Job 0

To build that relation, Spark needs to know which files exist. It creates an InMemoryFileIndex and walks the directory tree.

The top level has 365 directories. That's more than spark.sql.sources.parallelPartitionDiscovery.threshold (default 32), so Spark doesn't list them from the driver one by one. It launches a Spark job that lists directories in parallel on the executors. If Maya opens the Spark UI right now, she sees it:

Jobs
 Job 0   Listing leaf files and directories for 365 paths   365 tasks   1.4 s   ✔

The result is a list of ~16,000 FileStatus objects (path, size, modification time) cached in driver memory. While walking the tree, Spark also runs partition discovery: it sees dt=2026-03-14 in every path and turns dt into a column. Because spark.sql.sources.partitionColumnTypeInference.enabled is on, the value 2026-03-14 is inferred as a date, not a string.

⚠️ This listing is why "just calling spark.read" on a bucket with a million small files can take minutes before
any query runs. The fix is on the write side: fewer, bigger files, and a real table format (Hive metastore, Delta,
Iceberg) that stores the file list in metadata instead of walking S3.

1.3 Reading one footer for the schema

Maya didn't give a schema, so Spark infers one. For Parquet that's cheap: every Parquet file ends with a footer that contains the full schema and statistics for every column chunk. With spark.sql.parquet.mergeSchema at its default false, Spark opens one file, seeks to the end, reads the footer, and takes that schema as the schema of the whole dataset.

 One Parquet file, from the back:

 ┌──────┬─────────────┬─────────────┬─────┬─────────────────────┬─────┬──────┐
 │ PAR1 │ row group 0 │ row group 1 │ ... │  footer (metadata)  │ len │ PAR1 │
 └──────┴─────────────┴─────────────┴─────┴──────────┬──────────┴─────┴──────┘
                                                     ▲
                                                     │  schema, row counts, min/max
                                      Spark reads only this (a few KB)

Compare that with CSV or JSON with inferSchema=True: there is no footer, so Spark runs a full pass over the data just to guess types. That pass is a job too, and on 2 TB it's an expensive one. That's reason number one to prefer columnar formats, and reason number one to pass an explicit schema when you can't.

1.4 What `trips` actually is

After 1.8 seconds, the variable trips holds a DataFrame. In Spark 3.x/4.x a DataFrame is a Dataset[Row], and a Dataset is a thin wrapper around a logical plan: a tree that describes what you want, not how to get it. Right now that tree has exactly one node:

trips.explain(True)

== Parsed Logical Plan ==
Relation [trip_id,driver_id,zone_id,pickup_ts,fare,...,dt] parquet

== Analyzed Logical Plan ==
trip_id: string, driver_id: bigint, zone_id: int, pickup_ts: timestamp, fare: double, ..., dt: date
Relation [trip_id,driver_id,zone_id,pickup_ts,fare,...,dt] parquet

== Optimized Logical Plan ==
Relation [trip_id,driver_id,zone_id,pickup_ts,fare,...,dt] parquet

== Physical Plan ==
*(1) ColumnarToRow
+- FileScan parquet [trip_id,driver_id,zone_id,pickup_ts,fare,...,dt]
     Batched: true, Format: Parquet,
     Location: InMemoryFileIndex(1 paths)[s3a://ridehub/trips],
     PartitionFilters: [], PushedFilters: [],
     ReadSchema: struct<trip_id:string,driver_id:bigint,zone_id:int,pickup_ts:timestamp,fare:double,...>

Here's what's in driver memory after spark.read, and what isn't:

 DRIVER MEMORY after spark.read.parquet(...)
 ┌─────────────────────────────────────────────────────────┐
 │  DataFrame `trips`                                      │
 │   └─ LogicalPlan: Relation(HadoopFsRelation)            │
 │        ├─ schema         (from one footer)              │
 │        ├─ partition cols (dt: date, from paths)         │
 │        └─ InMemoryFileIndex                             │
 │             └─ ~16,000 FileStatus (path, size, mtime)   │
 └─────────────────────────────────────────────────────────┘
 EXECUTOR MEMORY: nothing. Zero trip rows anywhere.

1.5 Why laziness is the whole design, not a trick

Every method Maya calls next (filter, select, join, groupBy, orderBy) is a transformation. Each one returns a new DataFrame whose plan is the old plan with one more node on top. Nothing executes. Only an action (count, collect, show, write, toPandas, foreach) hands the plan to the engine.

This is deliberate. Because Spark sees the entire tree before it runs anything, it can push filters below joins, drop columns nobody reads, and decide how to join based on the size of what's left after filtering. An eager engine that executed spark.read.parquet as "read 2 TB" could never do that. Hold on to this idea: almost every optimization in the rest of this article is possible only because the read was lazy.

📝 PySpark note. Nothing Maya wrote runs in Python on the executors. The DataFrame API is a thin client: each
call is forwarded to the JVM (over Py4J in classic mode, or over gRPC with Spark Connect in 3.4+) and builds the plan
there. Python workers only appear on the executors when you use a Python UDF, rdd.map, or mapInPandas. Keep
that in mind when we hit the UDF trap in Chapter 3.

Chapter 2: The first action, or the anatomy of a job

Before touching the ticket, Maya wants a feel for the data:

trips.count()
# 3,214,906,117

It takes 3 minutes 50 seconds. Two terabytes in under four minutes on 40 cores felt impossibly fast to her, and Ravi says that's the second lesson: it was fast because of what it didn't read. But first, the journey. A count() goes through seven stops between Maya's notebook and the number on her screen.

2.1 Stop 1: four plans (Catalyst)

The action creates a QueryExecution, and the plan Maya built goes through Spark's optimizer, Catalyst, in four phases:

   trips.count()
        │
        ▼
 ┌──────────────────────┐   "Unresolved": names are just strings.
 │ Parsed logical plan  │   count(1) over Relation
 └──────────┬───────────┘
            │  ANALYZER: resolve column names against the catalog/schema,
            │  check types, expand *, resolve functions
            ▼
 ┌──────────────────────┐   Every column now has a type and an ID (fare#12)
 │ Analyzed logical plan│
 └──────────┬───────────┘
            │  OPTIMIZER: ~100 rule-based rewrites, applied until nothing changes
            │  ColumnPruning, PushDownPredicates, ConstantFolding,
            │  CombineFilters, ReorderJoin, ...
            ▼
 ┌──────────────────────┐   Still "what", but a cheaper "what"
 │ Optimized logical    │
 └──────────┬───────────┘
            │  SPARK PLANNER: strategies turn logical nodes into physical
            │  operators (Aggregate → HashAggregate, Join → BroadcastHashJoin...)
            ▼
 ┌──────────────────────┐   "How": real operators with real algorithms
 │ Physical plan        │
 └──────────┬───────────┘
            │  PREPARATION: EnsureRequirements inserts Exchanges (shuffles),
            │  CollapseCodegenStages fuses operators, AQE wraps the whole thing
            ▼
 ┌──────────────────────┐
 │ Executed plan        │ ──▶  RDDs  ──▶  DAGScheduler
 └──────────────────────┘

For count() the final physical plan is:

AdaptiveSparkPlan isFinalPlan=false
+- HashAggregate(keys=[], functions=[count(1)])                      ← final sum, 1 partition
   +- Exchange SinglePartition, ENSURE_REQUIREMENTS                  ← a SHUFFLE
      +- HashAggregate(keys=[], functions=[partial_count(1)])        ← one count per partition
         +- ColumnarToRow
            +- FileScan parquet []                                   ← no columns!
                 Batched: true, Format: Parquet,
                 Location: InMemoryFileIndex(1 paths)[s3a://ridehub/trips],
                 ReadSchema: struct<>

Three things to read off this plan, bottom to top:

FileScan parquet [] with ReadSchema: struct<>. The optimizer's ColumnPruning rule noticed that count(1) needs no columns at all. The Parquet reader will open each file, read the row count from each row group header, and return empty batches that carry only a row count. That's why 2 TB took four minutes: Spark read footers and headers, not data.
Two HashAggregate nodes with an Exchange between them. Aggregation is split into a partial step that runs where the data is, and a final step that merges the partials. The Exchange is a shuffle: data moving between tasks over the network. Every Exchange in a plan is a stage boundary.
AdaptiveSparkPlan isFinalPlan=false. Adaptive Query Execution (on by default since 3.2) will re-optimize this plan at each shuffle boundary using real statistics. We'll see it earn its keep in Chapters 4 and 5.

2.2 Stop 2: how many tasks? (file splitting)

The physical plan is turned into RDDs. The FileScan becomes a FileScanRDD, and its number of partitions is the number of tasks in the first stage. Spark decides that with a small piece of arithmetic in FilePartition.maxSplitBytes:

# Pseudocode of what the driver computes. Defaults in comments.
max_partition_bytes = 128 * MB            # spark.sql.files.maxPartitionBytes
open_cost           = 4 * MB              # spark.sql.files.openCostInBytes
min_partitions      = default_parallelism # spark.sql.files.minPartitionNum (= total cores = 40)

total = sum(f.size + open_cost for f in files)             # 2.0 TB + 16,000 × 4 MB
bytes_per_core = total / min_partitions                    # ≈ 51 GB
max_split = min(max_partition_bytes, max(open_cost, bytes_per_core))   # = 128 MB

Then Spark bin-packs: files are sorted by size (largest first), split into chunks of at most max_split, and chunks are packed into partitions until adding the next one would exceed max_split. Each resulting FilePartition is one task. With 16,000 files of ~128 MB:

	value
files	~16,000
bytes per task	128 MB
tasks in stage 0	16,384
task slots	40
waves of tasks	~410

Two consequences worth remembering:

Small files are expensive. The open_cost of 4 MB is a fake size added to every file so the packer doesn't put thousands of 10 KB files into one task. But each file is still a separate S3 GET for the footer. 16,000 files of 128 MB is fine; 2 million files of 1 MB would be a disaster at the same total size.
Splittable vs not. Parquet and ORC can be split at row-group boundaries, so one 1 GB file becomes 8 tasks. A gzip-compressed CSV or JSON file cannot be split: one file, one task, no matter how big. That's the classic "one executor is busy, nine are idle" symptom.

2.3 Stop 3: stages and the DAGScheduler

count() ends in sc.runJob(...). The DAGScheduler receives the final RDD and walks its lineage backwards. Every time it crosses a ShuffleDependency (the RDD form of an Exchange), it cuts a stage:

   RDD lineage (read bottom-up)                 Stages

   ResultRDD  (final HashAggregate)             ┌────────────────────────┐
       ▲  ShuffleDependency ◀──── stage cut ────│ Stage 1 (ResultStage)  │  1 task
       │                                        └────────────────────────┘
   MapPartitionsRDD (partial HashAggregate)     ┌────────────────────────┐
       ▲                                        │ Stage 0                │
   MapPartitionsRDD (ColumnarToRow)             │ (ShuffleMapStage)      │  16,384 tasks
       ▲                                        │                        │
   FileScanRDD (16,384 partitions)              └────────────────────────┘

Stage 0 has no missing parents, so it's submitted first as a TaskSet of 16,384 ShuffleMapTasks. Stage 1 waits until every task of Stage 0 has reported back.

2.4 Stop 4: shipping tasks to executors

The TaskSchedulerImpl takes the TaskSet and matches it against resource offers from executors ("I have 4 free cores"). The scheduling is FIFO by default, respects locality (PROCESS_LOCAL → NODE_LOCAL → RACK_LOCAL → ANY, waiting spark.locality.wait = 3 s per level), and, because the data is on S3 and not on local disks, every task here is ANY: there's no local copy to prefer.

Each LaunchTask message is small. The task's code (the whole RDD chain plus the generated Java for the operators) is serialized once into a taskBinary and sent to executors as a broadcast variable, not 16,384 times. The per-task message only carries the partition description: which files, which byte ranges.

 DRIVER                                               EXECUTOR 3
 ┌───────────────────────────┐                       ┌───────────────────────────┐
 │ DAGScheduler              │                       │ CoarseGrainedExecutor     │
 │   stage 0: 16,384 tasks   │                       │   Backend                 │
 │ TaskSchedulerImpl         │  LaunchTask(task 812: │   ┌─ task 812 ─┐          │
 │   TaskSetManager          │  file X[0:128MB])     │   ├─ task 813  │ 4 cores  │
 │ SchedulerBackend          │ --------------------> │   ├─ task 814  │ busy     │
 │   (RPC endpoint)          │                       │   └─ task 815 ─┘          │
 │ MapOutputTracker          │  <--------------------│ BlockManager              │
 │   (where shuffle          │  StatusUpdate(        │   shuffle files on local  │
 │    blocks live)           │  FINISHED, MapStatus) │   disk, broadcast cache   │
 └───────────────────────────┘                       └───────────────────────────┘

2.5 Stop 5: inside one task

Task 812 lands on executor 3, core 2. From its point of view:

Deserialize the task binary (first task on this executor only; afterwards it's cached) and the partition: dt=2026-01-19/part-00031.parquet, bytes 0..134217728.
Open the file through the Hadoop S3AFileSystem, seek to the end, read the footer. Pick the row groups whose midpoint falls inside this task's byte range. That midpoint rule is how two tasks that split one file never double-count a row group that straddles the boundary.
Read with the vectorized reader. VectorizedParquetRecordReader produces ColumnarBatches of up to 4,096 rows (spark.sql.parquet.columnarReaderBatchSize). For this query the batch has zero columns and a row count.
Run the generated code. The *(1) in explain marks a whole-stage codegen unit: Spark fused ColumnarToRow and the partial HashAggregate into one hand-written-looking Java class, compiled by Janino. For this plan the loop body is essentially count += batch.numRows().
Shuffle write. The partial result, one row holding one long, goes through the shuffle writer: it's written to a file under blockmgr-*/ on the executor's local disk, with a companion .index file, and the task reports a MapStatus (where the block is, how big) back to the driver's MapOutputTracker.

Multiply by 16,384. Average task time ~0.5 s, 40 at a time, roughly 3.5 minutes.

2.6 Stop 6: the second stage and the result

When the last MapStatus arrives, the DAGScheduler submits Stage 1: one ResultTask. It asks the MapOutputTracker where the 16,384 shuffle blocks are, fetches all of them (each is a few dozen bytes, but it's still 16,384 network requests, which is why this tiny stage takes a couple of seconds), sums them, and returns the single row to the driver.

The task result is small, so it comes back inline in the status update (DirectTaskResult, anything under spark.task.maxDirectResultSize = 1 MB). Larger results are written to the executor's BlockManager and the driver fetches them; the total is capped by spark.driver.maxResultSize (1 GB), which is the limit you hit when a collect() is too greedy.

The driver converts the row to a Python int, Py4J hands it to the notebook, and Maya sees 3214906117.

2.7 Stop 7: what the Spark UI shows

Jobs
 Job 0   Listing leaf files and directories for 365 paths     365 tasks      1.4 s
 Job 1   count at <cell>:1   Stage 0   16384 tasks   Input 2.0 TB   Shuffle Write 1.2 MB   3.7 min
 Job 2   count at <cell>:1   Stage 1       1 task    Shuffle Read 1.2 MB                     2 s

SQL / DataFrame tab → query 1
  Scan parquet:  number of files read 16,384 · size of files read 2.0 TB
                 number of output rows 3,214,906,117 · scan time 2.1 h (summed over tasks)
  HashAggregate: number of output rows 16,384 → Exchange → HashAggregate: 1

Two jobs for one count()? That's AQE: it submits each shuffle map stage as its own job so it can look at the real output sizes before planning the next stage. The SQL tab groups them back into one query.

"Input 2.0 TB" is the size of files the tasks were assigned. The bytes actually pulled from S3 were a tiny fraction, because of that empty ReadSchema. Here's the whole job as a timeline:

 time ──▶
 driver   │plan│sched│                                                        │fetch+sum│→ 3214906117
 exec 1   │    │     │▓▓▓▓▓▓▓▓▓▓▓▓▓▓▓▓▓▓▓▓▓▓▓▓▓▓▓▓▓▓▓▓▓▓▓▓▓▓▓▓▓▓▓▓▓▓▓▓▓▓▓▓▓▓▓│         │
 exec 2   │    │     │▓▓▓▓▓▓▓▓▓▓▓▓▓▓▓▓▓▓▓▓▓▓▓▓▓▓▓▓▓▓▓▓▓▓▓▓▓▓▓▓▓▓▓▓▓▓▓▓▓▓▓▓▓▓▓│         │
  ...     │    │     │  stage 0: 16,384 tasks, 4 per executor at a time       │         │
 exec 10  │    │     │▓▓▓▓▓▓▓▓▓▓▓▓▓▓▓▓▓▓▓▓▓▓▓▓▓▓▓▓▓▓▓▓▓▓▓▓▓▓▓▓▓▓▓▓▓▓▓▓▓▓▓▓▓▓▓│         │
                     ◀──────────────────── ~3.5 min ─────────────────────────▶◀─ 2 s ──▶
                                                                              stage 1 (1 task)

Maya now has the mental model that the rest of the story builds on:

Action → plans → stages (cut at shuffles) → tasks (cut at file splits) → executors → results.

Every remaining chapter changes just one of those arrows.

Chapter 3: Reading less: filters, pruning and pushdown

Now the ticket. March only, fares over $100, and Maya only needs a few columns:

march = (trips
    .filter((F.col("dt") >= "2026-03-01") & (F.col("dt") <= "2026-03-31"))
    .filter(F.col("fare") > 100)
    .select("trip_id", "driver_id", "zone_id", "fare", "dt"))

march.count()
# 41,882,310

9 seconds. From 3 minutes 50 to 9 seconds, on the same cluster, with no cache. That's not one optimization, it's three stacked on top of each other, and each one operates at a different granularity. Ravi draws a funnel:

                 16,384 files, 2.0 TB         ← what spark.read "pointed at"
   ┌─────────────────────────────────────────┐
   │ 1. PARTITION PRUNING  (driver, planning)│   dt between 03-01 and 03-31
   └────────────────────┬────────────────────┘   → 31 directories
                        ▼
                  1,364 files, 170 GB
   ┌─────────────────────────────────────────┐
   │ 2. COLUMN PRUNING  (executor, per file) │   read 4 of 18 column chunks
   └────────────────────┬────────────────────┘   → ~38 GB of column data
                        ▼
   ┌─────────────────────────────────────────┐
   │ 3. PREDICATE PUSHDOWN (executor, Parquet│   fare > 100 vs row-group stats,
   │    reader: row groups → pages)          │   dictionaries, page indexes
   └────────────────────┬────────────────────┘   → skip chunks that can't match
                        ▼
   ┌─────────────────────────────────────────┐
   │ 4. ROW FILTER  (executor, Spark)        │   fare > 100 on every row that
   └────────────────────┬────────────────────┘   survived, in generated code
                        ▼
                 41,882,310 rows

3.1 Partition pruning: never open the file

The plan tells the story:

march.explain()

*(2) HashAggregate(keys=[], functions=[count(1)])
+- Exchange SinglePartition, ENSURE_REQUIREMENTS
   +- *(1) HashAggregate(keys=[], functions=[partial_count(1)])
      +- *(1) Project
         +- *(1) Filter (isnotnull(fare) AND (fare > 100.0))
            +- *(1) ColumnarToRow
               +- FileScan parquet [fare,dt]
                    Batched: true, Format: Parquet,
                    Location: InMemoryFileIndex(1 paths)[s3a://ridehub/trips],
                    PartitionFilters: [isnotnull(dt), (dt >= 2026-03-01), (dt <= 2026-03-31)],
                    PushedFilters:    [IsNotNull(fare), GreaterThan(fare,100.0)],
                    DataFilters:      [isnotnull(fare), (fare > 100.0)],
                    ReadSchema: struct<fare:double>

Look at the FileScan line. The optimizer split Maya's two filters into two very different buckets:

PartitionFilters reference only partition columns. They are evaluated on the driver, at planning time, against the InMemoryFileIndex from Chapter 1. Spark walks its in-memory list of 365 directory values, keeps the 31 that satisfy dt between ..., and builds FilePartitions only from the 1,364 files inside them. The other 14,600 files are never opened, never even mentioned to an executor. Stage 0 now has 1,364 tasks instead of 16,384. No S3 calls are needed for this, because the listing was already cached.
DataFilters reference regular columns. They can't eliminate files by name. They're sent to the executors in two forms, which brings us to the next two layers.

💡 Since the file listing is reused, calling spark.read.parquet once and filtering many times is cheaper than
calling it once per query. In a notebook, keep the un-filtered DataFrame around.

3.2 Column pruning: open the file, skip most of it

ReadSchema: struct<fare:double>. Maya selected five columns, but count() needs only the one the filter touches, and dt is a partition column that comes from the path, not from the file. So each task asks the Parquet reader for exactly one column chunk per row group.

That's where Parquet's layout pays off:

 One Parquet file (~128 MB)

 ┌───────────────────────────────────────────────────────────────────────────┐
 │ Row group 0  (~1M rows)                                                   │
 │  ┌──────────┬──────────┬─────────┬───────────┬──────────┬─── ... ──────┐ │
 │  │ trip_id  │ driver_id│ zone_id │ pickup_ts │  fare    │  (13 more)   │ │
 │  │ chunk    │ chunk    │ chunk   │ chunk     │  chunk   │              │ │
 │  │ 22 MB    │ 4 MB     │ 1 MB    │ 6 MB      │  3 MB    │              │ │
 │  └──────────┴──────────┴─────────┴───────────┴──────────┴──────────────┘ │
 │    each chunk = pages of ~1 MB, each page: header + (dictionary) + values │
 ├───────────────────────────────────────────────────────────────────────────┤
 │ Row group 1 ...                                                           │
 ├───────────────────────────────────────────────────────────────────────────┤
 │ Footer: schema, and for every chunk: offset, size, num values,            │
 │         min, max, null count, (optional) column index per page            │
 └───────────────────────────────────────────────────────────────────────────┘

 This query reads:  [footer] + [fare chunk of each row group]   ≈ 3 MB per row group
 A row-oriented format (CSV, JSON, Avro) would have to read all 128 MB.

The reader seeks straight to the fare chunk's offset (it got that from the footer), reads ~3 MB, and moves on. The S3 bill for this query is dominated by footers and fare chunks: roughly 20–25% of the file bytes.

3.3 Predicate pushdown: let Parquet skip row groups and pages

PushedFilters: [IsNotNull(fare), GreaterThan(fare,100.0)]. These are Maya's filters translated into Parquet's own FilterPredicate API and handed to the Parquet reader (spark.sql.parquet.filterPushdown, on by default). The reader uses them at three levels, cheapest first:

level	what Parquet checks	skips
Row group statistics	footer says `fare: min=3.10, max=412.00` for this chunk. Can `fare > 100` be true? Yes → read. If `max` were 95 → skip the whole row group	whole row groups (~1M rows)
Dictionary filtering	for dictionary-encoded chunks, test the predicate against the dictionary entries. No entry > 100 → skip	whole row groups
Column index (page skipping)	Parquet ≥ 1.11 writes per-page min/max. Spark 3.2+ reads them and skips pages that can't match	pages (~1 MB)

Here's the catch Ravi wants Maya to notice: fares in a RideHub file are in pickup order, so every row group has fares from $3 to $400. The min/max check never skips anything. The pushdown is correct and free, but it's not doing work for this data layout.

This is why "pushdown is enabled" and "pushdown is effective" are different statements, and it's where writing matters as much as reading. If the files had been written with sortWithinPartitions("fare"), each row group would cover a narrow fare range, and the reader would skip 80% of them. Delta's ZORDER and Iceberg's sort orders exist for exactly this reason.

3.4 Why there's still a `Filter` operator

Maya spots it: *(1) Filter (isnotnull(fare) AND (fare > 100.0)) sits above the scan, even though the same predicate was pushed into the scan. Isn't that double work?

It's deliberate. Statistics, dictionaries and page indexes all work at chunk granularity: they can prove that a chunk contains no match, but a chunk that survives still contains rows that don't match (the $40 fares next to the $140 ones). The vectorized Parquet reader does not evaluate predicates per row, so Spark keeps its own row-level Filter to finish the job. That operator runs inside the same codegen stage (*(1)), one tight loop over the column vectors, so its cost is small.

Where each filter form actually runs:

 ┌──────────────┬────────────────────┬────────────────┬──────────────────────────┐
 │ form         │ where              │ when           │ granularity              │
 ├──────────────┼────────────────────┼────────────────┼──────────────────────────┤
 │ Partition    │ DRIVER             │ planning       │ directory / file         │
 │ filter       │ (FileIndex)        │                │ never opens the file     │
 ├──────────────┼────────────────────┼────────────────┼──────────────────────────┤
 │ Pushed       │ EXECUTOR           │ while reading  │ row group, page          │
 │ filter       │ (Parquet reader)   │                │ via min/max, dictionary  │
 ├──────────────┼────────────────────┼────────────────┼──────────────────────────┤
 │ Data filter  │ EXECUTOR           │ after reading  │ row                      │
 │ (Filter op)  │ (generated code)   │                │ final correctness check  │
 └──────────────┴────────────────────┴────────────────┴──────────────────────────┘

3.5 The Spark UI after pruning

SQL / DataFrame tab → query 2
  Scan parquet:  number of files read 1,364          (was 16,384)
                 number of partitions read 31        (dt values)
                 size of files read 170.5 GB
                 dynamic partition pruning time 0 ms
                 number of output rows 273,015,992   ← rows that left the scan
  Filter:        number of output rows 41,882,310    ← rows that passed fare > 100

That gap between the scan's output rows (273 M) and the filter's output rows (42 M) is the pushdown not helping: 85% of the rows the reader produced were thrown away by the row-level filter. On a fare-sorted layout, that first number would have been close to the second.

3.6 The five ways to lose pruning by accident

Each of these looks harmless and quietly turns the 9-second query back into a 4-minute one. The tell is always in explain(). The first one is exactly what Maya's Monday code did.

1. Filtering the wrong column (Maya's Monday mistake)

# ✗ pickup_ts is a data column, not the partition column.
#   No partition pruning, and the month() wrapper means no pushdown either:
#   every row group of every file is decoded just to test the predicate.
trips.filter((F.year("pickup_ts") == 2026) & (F.month("pickup_ts") == 3))
#   PartitionFilters: []
#   PushedFilters:    []
#   DataFilters:      [(year(pickup_ts) = 2026), (month(pickup_ts) = 3)]

# ✓ filter on the partition column, as a plain range
trips.filter(F.col("dt").between("2026-03-01", "2026-03-31"))
#   PartitionFilters: [isnotnull(dt), (dt >= 2026-03-01), (dt <= 2026-03-31)]

The two columns mean the same thing to a human and completely different things to the file index. (A function around the partition column itself, like F.month("dt") == 3, still prunes on a plain file source: Spark evaluates it against the 365 directory values on the driver. Hive-metastore tables are less forgiving, because only simple comparisons can be pushed to the metastore.)

2. A function around a data column

# ✗ nothing pushed: Parquet has no stats for round(fare)
trips.filter(F.round("fare") > 100)
#   PushedFilters: []

3. A type mismatch that casts the column instead of the literal

# driver_id is a long. Comparing it to a string makes Spark cast the COLUMN.
trips.filter(F.col("driver_id") == "4471")
#   DataFilters: [(cast(driver_id as string) = 4471)]   PushedFilters: []

# ✓ compare like with like
trips.filter(F.col("driver_id") == 4471)
#   PushedFilters: [IsNotNull(driver_id), EqualTo(driver_id,4471)]

4. A Python UDF in the filter

@F.udf("boolean")
def is_premium(fare):
    return fare is not None and fare > 100

trips.filter(is_premium("fare"))

*(2) Project
+- *(2) Filter pythonUDF0: boolean
   +- BatchEvalPython [is_premium(fare)]      ← rows leave the JVM, go to a Python
      +- *(1) ColumnarToRow                      worker process, and come back
         +- FileScan parquet [fare] PushedFilters: []

Catalyst can't see inside a Python function, so it pushes nothing and ships every row through a Python worker. The same filter as a native expression stays in the JVM, pushes down, and is 50–100× faster.

5. A LIKE '%...%'

startswith becomes StringStartsWith and is pushed; a contains / '%x%' can't use min/max statistics and isn't. Not a bug, just physics: there's no cheap way to prove a chunk contains no substring.

📝 Rule of thumb. After writing a filter, run .explain() and read the FileScan line. If the predicate you
wrote isn't in PartitionFilters or PushedFilters, Spark is doing it the hard way.

Maya adds a line to her notes: "The filter I write is a wish. The FileScan line is what Spark actually does."

Chapter 4: Putting things in order: sort and shuffle

Finance asked for a ranked report, so Maya tries the obvious thing on her March subset and writes it out:

(march
    .orderBy(F.desc("fare"))
    .write.mode("overwrite").parquet("s3a://ridehub/scratch/march_sorted/"))

It takes 2 minutes 10 seconds. The count on the same data took 9. The Spark UI shows something new: three jobs for one action, the first of which she never asked for, and 1.6 GB of shuffle between the last two.

Jobs
 Job 3   parquet at <cell>:3   (RangePartitioner sampling)   1,364 tasks                           11 s
 Job 4   parquet at <cell>:3   Stage 3   1364 tasks   Input 170 GB   Shuffle Write 1.6 GB      1.6 min
 Job 5   parquet at <cell>:3   Stage 4     25 tasks   Shuffle Read 1.6 GB   Output 25 files      14 s

4.1 What "sorted" means on a cluster

On one machine, sorting is one algorithm. On a cluster, globally sorted output means: every row in output file 1 is greater than every row in file 2, which is greater than every row in file 3, and each file is sorted internally. To arrange that, Spark has to decide which rows go to which file before it can sort anything. That is a range partitioning problem, and it needs to know the distribution of fare to pick good boundaries. The plan shows both halves:

*(2) Sort [fare DESC NULLS LAST], true, 0                              ← local sort, per partition
+- AQEShuffleRead coalesced
   +- ShuffleQueryStage 0
      +- Exchange rangepartitioning(fare DESC NULLS LAST, 200), ENSURE_REQUIREMENTS   ← the shuffle
         +- *(1) Project [trip_id, driver_id, zone_id, fare, dt]
            +- *(1) Filter (isnotnull(fare) AND (fare > 100.0))
               +- *(1) ColumnarToRow
                  +- FileScan parquet [trip_id,driver_id,zone_id,fare,dt] ... (same pruning as before)

Sort [...], true, 0: the true means global. The Exchange rangepartitioning(..., 200) is what makes global possible, and 200 is spark.sql.shuffle.partitions, the default number of partitions after any shuffle.

4.2 Job 3: the sampling pass nobody asked for

Before Spark can shuffle by range it needs the range boundaries. The RangePartitioner gets them by sampling the input:

target sample: 100 × 200 partitions = 20,000 rows (spark.sql.execution.rangeExchange.sampleSizePerPartition = 100),
spread across the 1,364 input partitions, so each task reservoir-samples about 44 rows (Spark asks for 3 × the target to tolerate uneven partitions),
the samples are collected to the driver, sorted, and 199 boundary values are chosen so that each of the 200 ranges should get roughly 1/200 of the rows.

That sampling is a real job over the real input: it opens the 1,364 files, applies the partition pruning, column pruning and the fare > 100 filter, and keeps a few dozen rows per task. On this data it costs 11 seconds, about the price of the count(). A global sort reads its input twice.

          sampling job (Job 3)                         boundaries on the driver

  task 0  ─▶ 44 random fares ─┐
  task 1  ─▶ 44 random fares ─┤                sorted sample: 100.2 ... 412.0
   ...                        ├─▶ collect() ─▶ cut into 200 ranges:
  task 1363 ▶ 44 random fares ┘                p0: fare ≥ 398.1
                                               p1: 391.6 ≤ fare < 398.1
                                               ...
                                               p199: fare < 101.3

4.3 Stage 3: the shuffle write (map side)

Now the main job. Each of the 1,364 map tasks reads its files (with all the pruning from Chapter 3), and for every surviving row:

picks a destination partition by binary-searching the 199 boundaries (fare = 215.40 → partition 87),
hands the row to the shuffle writer. Because Spark SQL rows are already in a compact binary format (UnsafeRow), and there's no map-side aggregation, it uses the UnsafeShuffleWriter: rows are serialized into memory pages and the sorter only moves 8-byte pointers tagged with the partition ID. When memory runs low it spills sorted runs to local disk,
at the end, merges the runs into exactly one data file and one index file per task. The index says where each of the 200 partitions starts inside the data file.

 map task 812 output, on executor 3's local disk

 shuffle_0_812_0.data                                  shuffle_0_812_0.index
 ┌────────┬────────┬────────┬─────────┬────────┐       ┌────────────────────┐
 │ p0     │ p1     │ p2     │  ...    │ p199   │       │ offset of p0 = 0   │
 │ 4 KB   │ 6 KB   │ 5 KB   │         │ 7 KB   │  ◀──  │ offset of p1 = 4096│
 └────────┴────────┴────────┴─────────┴────────┘       │ ...                │
   rows with fare ≥ 398.1  │                           └────────────────────┘
                           └─ 391.6 ≤ fare < 398.1

The task reports a MapStatus to the driver: "my output lives on executor 3, and partition i is n bytes" for all 200 values of i. That's compressed to a few bytes per partition, but with 1,364 maps × 200 reduces the driver is tracking 272,800 blocks for this one shuffle.

4.4 Stage 4: fetch, sort, write (reduce side)

Reduce task i now:

asks the MapOutputTracker for the locations of block i from every one of the 1,364 map outputs,
fetches them with the ShuffleBlockFetcherIterator, keeping up to spark.reducer.maxSizeInFlight (48 MB) of requests outstanding, from the executors' block managers (or from the external shuffle service, if one is running so that executors can be removed without losing their shuffle files),
runs SortExec: an UnsafeExternalSorter sorts rows in memory using an 8-byte prefix of the sort key for fast comparisons, and spills to disk if execution memory (its share of the unified spark.memory.fraction = 60% of heap) runs out,
writes its partition as one Parquet file.

 shuffle = every reduce talks to every map

   map 0    map 1    map 2    ...   map 1363          1,364 map outputs
    │ │ │    │ │ │    │ │ │          │ │ │
    │ │ └────┼─┼─┼────┼─┼─┼──────────┼─┼─┘ ─────▶ reduce 199  (fare < 101.3)
    │ └──────┼─┼──────┼─┼────────────┼─┘   ─────▶ reduce 1
    └────────┼────────┼──────────────┘     ─────▶ reduce 0    (fare ≥ 398.1)
             ...                                  200 reduce tasks
                                                  each sorts its own range
                                                  → output is globally sorted

4.5 Where did the 200 go? (AQE coalescing)

The plan said 200 partitions, the UI shows 25 tasks in Stage 4. That's AQEShuffleRead coalesced. When Stage 3 finished, AQE looked at the real MapStatus sizes, saw that 1.6 GB split 200 ways is 8 MB per reducer, and merged adjacent ranges until each was close to spark.sql.adaptive.advisoryPartitionSizeInBytes (64 MB). Because the ranges are contiguous, merging neighbours keeps the output globally sorted. Without AQE, Maya would have 200 tiny files and 200 tasks that spend more time fetching than sorting. With it, 25 files of ~64 MB. This is the single most useful thing AQE does, and it's why nobody tunes spark.sql.shuffle.partitions by hand any more for most jobs.

4.6 The twist: the top 10 took 10 seconds

Maya only needs the top 10 for the ticket, so she tries:

march.orderBy(F.desc("fare")).show(10)

10 seconds. No sampling job. No shuffle. The plan has a different operator entirely:

TakeOrderedAndProject(limit=11, orderBy=[fare DESC NULLS LAST], output=[trip_id,...])
+- *(1) Project ...
   +- *(1) Filter (isnotnull(fare) AND (fare > 100.0))
      +- *(1) ColumnarToRow
         +- FileScan parquet ...

The optimizer saw Sort directly under Limit and replaced both with TakeOrderedAndProject. Each of the 1,364 tasks keeps a bounded priority queue of the 11 largest fares it sees (11, not 10: show asks for one extra row to know whether to print "only showing top 10 rows"), sends those 11 rows to the driver, and the driver merges 1,364 × 11 rows. One stage, no shuffle, a few hundred KB over the network.

The same rewrite happens for .orderBy().limit(n) followed by a write or a collect(). It's a reminder that the physical plan depends on the whole tree, not on the last thing you typed. orderBy alone means a sampling job plus a full shuffle; orderBy + limit means a map-only top-k.

4.7 Three sorts, three costs

you write	physical plan	shuffle?	when to use
`orderBy(c)`	`Exchange rangepartitioning` + `Sort global=true`	yes, plus a sampling job	you need globally ordered output (a report someone reads top to bottom, a sorted export)
`sortWithinPartitions(c)`	`Sort global=false`	no	you want each output file sorted, usually so that Parquet min/max stats become useful (Chapter 3)
`orderBy(c).limit(n)`	`TakeOrderedAndProject`	no	top-N / bottom-N
`repartitionByRange(n, c).sortWithinPartitions(c)`	`Exchange rangepartitioning` + `Sort global=false`	yes	the explicit form of a global sort with control over `n`

Maya's note for the day: "orderBy is the most expensive word in the API. Say limit right after it, or mean it."

Chapter 5: Meeting other tables: joins

The ticket wants names, not IDs. Two joins: zones (20 KB) and drivers (50 MB). Spark will treat them completely differently, and the reason is one number: spark.sql.autoBroadcastJoinThreshold = 10 MB.

5.1 The zones join: broadcast hash join

with_zone = march.join(zones, "zone_id")
with_zone.explain()

*(2) Project [trip_id, driver_id, fare, dt, zone_name, borough]
+- *(2) BroadcastHashJoin [zone_id], [zone_id], Inner, BuildRight, false
   :- *(2) Filter ((isnotnull(fare) AND (fare > 100.0)) AND isnotnull(zone_id))
   :  +- *(2) ColumnarToRow
   :     +- FileScan parquet [trip_id,driver_id,zone_id,fare,dt] ...    ← trips: 1,364 tasks
   +- BroadcastExchange HashedRelationBroadcastMode(List(cast(zone_id as bigint))), [plan_id=…]
      +- *(1) Filter isnotnull(zone_id)
         +- *(1) ColumnarToRow
            +- FileScan parquet [zone_id,zone_name,borough] ...        ← zones: 1 task

At planning time Spark asks each side for stats.sizeInBytes. For a file source that's simply the sum of file sizes (times spark.sql.sources.fileCompressionFactor, default 1.0). Zones is 20 KB, far under 10 MB, so the planner picks a BroadcastHashJoin with zones as the build side. Notice there's no Exchange on the trips side. The 170 GB never moves.

What happens at runtime, in order:

 DRIVER                                            EXECUTORS (×10)
 ┌───────────────────────────────┐
 │ 1. BroadcastExchange starts   │
 │    on a background thread     │
 │ 2. runs a small job: scan     │ ◀── 1 task reads zones.parquet, returns 263 rows
 │    zones, collect() rows      │
 │ 3. builds the hash table      │
 │    HERE, on the driver:       │
 │    LongHashedRelation         │
 │    (zone_id → row)            │
 │ 4. sc.broadcast(relation)     │
 │    TorrentBroadcast: split    │ ──▶ 5. first task on each executor fetches the
 │    into 4 MB blocks, store in │      blocks (from the driver, then from peers,
 │    driver BlockManager        │      BitTorrent-style), caches the relation
 └───────────────────────────────┘      in its BlockManager; all 4 cores share it
                                        6. every trips task: for each row, probe
                                           the hash table by zone_id → join row

Steps 1–4 are the extra "broadcast exchange" job you see in the UI, running concurrently with the main job's planning. The trips stage is then map-only: 1,364 tasks, each probing an in-memory hash table inside the generated code. The join costs about as much as a filter.

Three facts about broadcasts that bite later:

The hash table is built on the driver. A "small" table that's 10 MB on disk as Parquet can be 100 MB as Java objects, and it's collected into driver memory first. That's why the default threshold is a conservative 10 MB and why bumping it to 1 GB "because the executors have RAM" is a classic way to OOM the driver.
Hard limits: a broadcast relation can't exceed 8 GB or 512 million rows; Spark throws rather than tries. And the build must finish within spark.sql.broadcastTimeout (300 s), or you get the infamous "Could not execute broadcast in 300 secs".
You can force or forbid it. F.broadcast(df) or the SQL hint /*+ BROADCAST(z) */ overrides the size estimate; spark.sql.autoBroadcastJoinThreshold = -1 disables automatic broadcasting.

5.2 The drivers join: sort-merge join

with_driver = march.join(drivers, "driver_id")
with_driver.explain()

*(5) Project [trip_id, zone_id, fare, dt, driver_name]
+- *(5) SortMergeJoin [driver_id], [driver_id], Inner
   :- *(2) Sort [driver_id ASC NULLS FIRST], false, 0
   :  +- Exchange hashpartitioning(driver_id, 200), ENSURE_REQUIREMENTS       ← shuffle trips
   :     +- *(1) Filter (... AND isnotnull(driver_id))
   :        +- *(1) ColumnarToRow
   :           +- FileScan parquet [trip_id,driver_id,zone_id,fare,dt] ...
   +- *(4) Sort [driver_id ASC NULLS FIRST], false, 0
      +- Exchange hashpartitioning(driver_id, 200), ENSURE_REQUIREMENTS       ← shuffle drivers
         +- *(3) Filter isnotnull(driver_id)
            +- *(3) ColumnarToRow
               +- FileScan parquet [driver_id,driver_name] ...

Drivers is 50 MB on disk. That's above the threshold, so the planner falls back to the default for large equi-joins: SortMergeJoin. And a sort-merge join has a requirement that EnsureRequirements enforces by inserting two Exchanges: both inputs must be hash-partitioned on the join key, with the same number of partitions, and sorted by that key within each partition.

The trick that makes this work is co-partitioning. Both sides use the same function, hash(driver_id) mod 200, so every trip with driver_id = 4471 and the one driver row with driver_id = 4471 land in the same partition number, which means the same reduce task:

   trips (1,364 map tasks)                  drivers (1 map task)
   ┌──────┬──────┬──────┬───┐               ┌──────┬──────┬──────┬───┐
   │ p0   │ p1   │ p2   │...│               │ p0   │ p1   │ p2   │...│
   └──┬───┴──┬───┴──┬───┴───┘               └──┬───┴──┬───┴──┬───┴───┘
      │      │      │      hash(driver_id) mod 200 │      │      │
      ▼      │      │                              ▼      │      │
   ┌────────────────────────────────────────────────────────────────┐
   │ reduce task 0:  sort trips p0 by driver_id                     │
   │                 sort drivers p0 by driver_id                   │
   │                 walk both with two pointers, emit matches      │
   └────────────────────────────────────────────────────────────────┘
      ... × 200

The stage graph now has a shape Maya hasn't seen: two independent map stages that run in parallel, then a reduce stage that depends on both.

   Stage 5: scan trips → filter → shuffle write (1,364 tasks, 600 MB written)
   Stage 6: scan drivers        → shuffle write (1 task, 8 MB written)
            │                                     │
            └────────────────┬────────────────────┘
                             ▼
   Stage 7: fetch both, sort both, merge-join (200 tasks → AQE coalesces)

Every byte of the filtered trips (600 MB after column pruning and the fare filter) crosses the network, and then gets sorted. On the unfiltered dataset it would have been about 19 GB.

5.3 AQE changes its mind at runtime

Then something interesting happens. Maya runs it, opens the SQL tab, and the final plan isn't a sort-merge join at all:

AdaptiveSparkPlan isFinalPlan=true
+- *(3) Project [...]
   +- *(3) BroadcastHashJoin [driver_id], [driver_id], Inner, BuildRight, false
      :- AQEShuffleRead local                                  ← no network for trips
      :  +- ShuffleQueryStage 0
      :     +- Exchange hashpartitioning(driver_id, 200)
      :        +- ... FileScan parquet trips
      +- BroadcastQueryStage 2
         +- BroadcastExchange HashedRelationBroadcastMode(List(driver_id))
            +- AQEShuffleRead local
               +- ShuffleQueryStage 1
                  +- Exchange hashpartitioning(driver_id, 200)
                     +- ... FileScan parquet drivers [driver_id,driver_name]

Here's the sequence:

AQE executes the plan one shuffle stage at a time, leaf-first. Stages 5 and 6 are both submitted.
Stage 6 (drivers) finishes almost immediately. Its MapStatuses say the shuffle output is 8 MB: after column pruning (driver_id, driver_name, not the 18 columns on disk) and compression, drivers is a lot smaller than its 50 MB file size suggested.
8 MB is under spark.sql.adaptive.autoBroadcastJoinThreshold (defaults to the same 10 MB). AQE re-plans the join as a BroadcastHashJoin, reading the drivers side from its shuffle output and broadcasting it.
The trips side is awkward: its shuffle write is already running. Spark can't un-write those files, but it can avoid the expensive half of a shuffle. AQEShuffleRead local means each reduce task reads only the map outputs on its own executor, with no network transfer, no sort, and no 272,800-block fetch plan.

So the join ended up costing one cheap shuffle write and a broadcast. Two lessons Ravi draws on the whiteboard:

Planning-time size estimates are about files; runtime sizes are about what's left. The planner saw 50 MB of Parquet. The runtime saw 8 MB of two columns. AQE exists to close that gap.
AQE can't undo work already started. If Maya had written F.broadcast(drivers) herself, the trips side would never have been shuffled at all. When you know a side is small, say so.

5.4 What if drivers had been 5 GB? (skew)

Ravi's thought experiment: imagine drivers were too big to broadcast and the sort-merge join stuck. Now imagine, as is true at RideHub, that 35% of March trips have driver_id = -1, the placeholder for "driver unknown". Hash partitioning sends every one of those 14 million rows to one reduce task:

 reduce task sizes (bytes of shuffle read)

 p0   ████                                           8 MB
 p1   ███                                            7 MB
 ...
 p117 ████████████████████████████████████████████ 290 MB   ← hash(-1) mod 200
 ...
 p199 ████                                           8 MB

 199 tasks finish in 4 s. Task 117 runs for 6 minutes. The stage takes 6 minutes.

That's data skew, and it's the most common reason a Spark job has one task that won't finish. AQE has a rule for it, spark.sql.adaptive.skewJoin.enabled (on by default): after the map stages finish, any partition that is both larger than skewedPartitionFactor (5) × the median and larger than skewedPartitionThresholdInBytes (256 MB) is split into several reader tasks, each covering a slice of the map outputs. The other side's matching partition is read once by each slice, so every slice still sees all the rows it could match:

   before                              after AQE skew handling
   reduce 117: trips p117 (290 MB)     reduce 117a: trips p117 maps 0–340    ⋈ drivers p117
               ⋈ drivers p117          reduce 117b: trips p117 maps 341–680  ⋈ drivers p117
                                       reduce 117c: trips p117 maps 681–1020 ⋈ drivers p117
                                       reduce 117d: trips p117 maps 1021–1363⋈ drivers p117

The plan marks it SortMergeJoin ... isSkew=true with AQEShuffleRead skewed. Before AQE, you fixed this by hand with salting: append a random suffix to the hot key on the big side, explode the small side across all suffixes, join on the salted key. Still useful for aggregations, which AQE's skew rule doesn't cover.

And notice: a broadcast join has no skew problem at all, because the big side is never partitioned by the key. Another reason to broadcast when you can.

5.5 Dynamic partition pruning: the join that prunes the scan

One more join, because it's the one that surprises people. Finance's follow-up: "same report, but only for public holidays". Holidays are a tiny table of dates:

us_holidays = spark.read.parquet("s3a://ridehub/holidays/").filter(F.col("country") == "US")

trips.join(us_holidays, "dt").filter(F.col("fare") > 100).count()

There's no literal dt filter on trips, so by the rules of Chapter 3 Spark should scan all 365 partitions, then join. Instead:

*(2) HashAggregate(...)
+- ...
   +- *(2) BroadcastHashJoin [dt], [dt], Inner, BuildRight
      :- *(2) Filter (isnotnull(fare) AND (fare > 100.0))
      :  +- *(2) ColumnarToRow
      :     +- FileScan parquet [fare,dt]
      :          PartitionFilters: [isnotnull(dt), dynamicpruningexpression(dt IN dynamicpruning#88)]
      :             +- SubqueryBroadcast dynamicpruning#88, 0, [dt]
      :                +- BroadcastExchange HashedRelationBroadcastMode(List(dt))
      :                   +- ... FileScan parquet holidays [dt] PushedFilters: [EqualTo(country,US)]
      +- BroadcastExchange ... (reused)

dynamicpruningexpression(dt IN dynamicpruning#88): a partition filter whose value list isn't known until runtime. This is Dynamic Partition Pruning (Spark 3.0+, spark.sql.optimizer.dynamicPartitionPruning.enabled). Spark runs the holidays side first (it was going to broadcast it for the join anyway, so the result is reused), collects the 11 distinct dt values, and then decides which trips files to list into tasks. 11 partitions instead of 365; 484 tasks instead of 16,384. The SQL tab even has a metric for it: dynamic partition pruning time.

The conditions: the join key must be a partition column of the big side, the small side must have a filter that makes it selective, and the small side must be broadcastable (or Spark must judge the pruning worth a separate subquery). Star-schema fact tables partitioned by date, joined to a filtered dimension, hit this constantly, and it's why "filter the dimension, join the fact" can be as fast as filtering the fact directly.

5.6 How Spark picks a join

Putting the chapter together, this is the decision the planner makes, in order:

 is there a join hint?  ──yes──▶ obey it (BROADCAST / MERGE / SHUFFLE_HASH / SHUFFLE_REPLICATE_NL)
        │ no
        ▼
 equi-join (keys compared with =)?
        │ yes                                        │ no
        ▼                                            ▼
 one side ≤ autoBroadcastJoinThreshold?       one side broadcastable?
   yes → BroadcastHashJoin (map-only)           yes → BroadcastNestedLoopJoin
   no  ↓                                        no  → CartesianProduct (and a warning
 preferSortMergeJoin = false and                      about your life choices)
 one side much smaller than the other?
   yes → ShuffledHashJoin  (shuffle both, hash the small side, no sort)
   no  → SortMergeJoin     (shuffle both, sort both, merge)
        ↓ at runtime, AQE may still:
          • convert SMJ → BHJ if a shuffled side turns out small
          • split skewed partitions
          • coalesce small partitions

strategy	shuffles the big side?	memory risk	shines when
BroadcastHashJoin	no	driver + one copy per executor	one side fits in memory (≲ tens of MB on disk)
ShuffledHashJoin	yes	one partition of the small side per task	both large, one clearly smaller, keys not sortable
SortMergeJoin	yes	spills gracefully	both large, the safe default
BroadcastNestedLoopJoin	no	O(n × m) CPU	non-equi joins (`<`, `BETWEEN`) with a small side

Maya's note: "A join is a question about sizes. Answer it before Spark guesses."

Chapter 6: The whole pipeline, end to end

It's Wednesday afternoon. Maya rewrites FIN-2231 with everything from the week:

report = (trips
    .filter(F.col("dt").between("2026-03-01", "2026-03-31"))   # 1. prunable: a plain range on the partition column
    .filter(F.col("fare") > 100)                                # 2. native expression → pushed to Parquet
    .join(F.broadcast(zones), "zone_id")                        # 3. tiny: say so
    .join(F.broadcast(drivers), "driver_id")                    # 4. small after pruning: say so, skip the shuffle
    .groupBy("driver_name", "zone_name")
    .agg(F.sum("fare").alias("total_fare"), F.count("*").alias("trips"))
    .orderBy(F.desc("total_fare"))                              # 5. followed by limit → top-k, no sort shuffle
    .limit(10))

report.write.mode("overwrite").parquet("s3a://ridehub/reports/fin-2231/")

41 seconds. Same cluster, same data, same answer. Before she ships it, Ravi makes her read the final plan out loud, bottom to top.

6.1 The final plan, annotated

report.explain()

AdaptiveSparkPlan isFinalPlan=true
+- TakeOrderedAndProject(limit=10, orderBy=[total_fare DESC NULLS LAST],          ⑧ top-10, no range shuffle
     output=[driver_name, zone_name, total_fare, trips])
   +- *(3) HashAggregate(keys=[driver_name, zone_name],                           ⑦ final sum/count
            functions=[sum(fare), count(1)])
      +- AQEShuffleRead coalesced                                                 ⑥ 200 → 1 partition (62 MB)
         +- ShuffleQueryStage 2
            +- Exchange hashpartitioning(driver_name, zone_name, 200)             ⑤ THE one shuffle
               +- *(2) HashAggregate(keys=[driver_name, zone_name],               ④ partial sum/count per task
                        functions=[partial_sum(fare), partial_count(1)])
                  +- *(2) Project [fare, zone_name, driver_name]
                     +- *(2) BroadcastHashJoin [driver_id], [driver_id], Inner, BuildRight   ③ probe drivers table
                        :- *(2) Project [driver_id, fare, zone_name]
                        :  +- *(2) BroadcastHashJoin [zone_id], [zone_id], Inner, BuildRight ② probe zones table
                        :     :- *(2) Filter (((isnotnull(fare) AND (fare > 100.0))
                        :     :         AND isnotnull(zone_id)) AND isnotnull(driver_id))
                        :     :  +- *(2) ColumnarToRow
                        :     :     +- FileScan parquet [driver_id,zone_id,fare,dt]             ① the only big read
                        :     :          PartitionFilters: [isnotnull(dt), (dt >= 2026-03-01), (dt <= 2026-03-31)],
                        :     :          PushedFilters: [IsNotNull(fare), GreaterThan(fare,100.0),
                        :     :                          IsNotNull(zone_id), IsNotNull(driver_id)],
                        :     :          ReadSchema: struct<driver_id:bigint,zone_id:int,fare:double>
                        :     +- BroadcastQueryStage 0
                        :        +- BroadcastExchange HashedRelationBroadcastMode(List(cast(zone_id as bigint)))
                        :           +- *(1) ... FileScan parquet zones [zone_id,zone_name]
                        +- BroadcastQueryStage 1
                           +- BroadcastExchange HashedRelationBroadcastMode(List(driver_id))
                              +- ... FileScan parquet drivers [driver_id,driver_name]

Things Maya can now point at:

① ReadSchema has three columns out of eighteen, and dt isn't in it (it comes from the path). The optimizer worked out that trip_id is never used and dropped it, even though the joins and aggregates are five operators higher up.
The Filter has grown two extra predicates she never wrote: isnotnull(zone_id) and isnotnull(driver_id). That's InferFiltersFromConstraints: an inner join can never match a null key, so the optimizer filters them out before the join, and pushes those too.
② ③ Two broadcast joins, zero Exchanges under them. The 170 GB of March is read once and never shuffled.
④ ⑤ ⑦ The classic partial/exchange/final aggregate triplet. The shuffle carries pre-aggregated rows: at most one row per (driver, zone) per task, which is why it's 62 MB and not 600 MB.
⑥ AQE collapsed the 200 reducers into 1, because 62 MB fits in a single advisory partition.
⑧ TakeOrderedAndProject: because limit(10) sits directly on orderBy, there's no range partitioning and no sampling job.
The whole-stage codegen stage IDs: *(2) covers scan → filter → join → join → partial aggregate. Five operators, one generated Java loop per task. A row is read from a column vector, tested, probed twice, and folded into a hash map without ever being materialized as an object in between.

6.2 The stage map

 Jobs for one action (report.write)

 Job 6   broadcast exchange (zones)     1 task    0.4 s  ─┐ started by background threads
 Job 7   broadcast exchange (drivers)   1 task    2.1 s  ─┘ as soon as the plan is ready

 Job 8   parquet at <cell>:11  (AQE: the shuffle map stage runs as its own job)
         ┌──────────────────────────────────────────────────────────────────────────┐
         │ Stage 8   1,364 tasks                                           34 s     │
         │  scan (3 cols, 31 dirs) → filter → ⋈ zones → ⋈ drivers → partial agg     │
         │  → shuffle write by hash(driver_name, zone_name)                         │
         │  Input 170 GB · Shuffle Write 62 MB                                      │
         └───────────────────────────────────┬──────────────────────────────────────┘
                                             ▼
 Job 9   parquet at <cell>:11  (the final stages, planned after Stage 8's sizes were known)
         ┌──────────────────────────────────────────────────────────────────────────┐
         │ Stage 9   1 task (coalesced from 200)                            3 s     │
         │  shuffle read 62 MB → final agg (2.1 M groups) → local top-10            │
         └───────────────────────────────────┬──────────────────────────────────────┘
                                             ▼
         ┌──────────────────────────────────────────────────────────────────────────┐
         │ Stage 10  1 task                                                 1 s     │
         │  merge top-10 → write 1 Parquet file + _SUCCESS                          │
         └──────────────────────────────────────────────────────────────────────────┘

6.3 A day in the life of task 812

To make it concrete, here is one task from Stage 8, millisecond by millisecond, as reconstructed from the task metrics:

   0 ms   LaunchTask arrives at executor 3; core 2 is free
   2 ms   deserialize partition: dt=2026-03-14/part-00017.parquet, bytes [0, 134217728)
   4 ms   broadcast 0 (zones) and broadcast 1 (drivers): already in this executor's
          BlockManager, fetched by the first task that ran here. Zero network.
  41 ms   S3 GET: last 64 KB of the file → footer: 8 row groups, offsets of the
          driver_id / zone_id / fare chunks, min/max per chunk
  43 ms   PushedFilters vs stats: fare max is 388–412 in every row group → nothing skipped
          (the data is in pickup order; see Chapter 3)
  50 ms   row group 0: GET three column chunks (~7.5 MB total)
 120 ms     decode → ColumnarBatch(4096 rows) × ~250 batches
            generated loop per row:
              fare > 100 ?            → 13% survive
              zones.probe(zone_id)    → zone_name
              drivers.probe(driver_id)→ driver_name
              aggMap[(driver_name, zone_name)] += (fare, 1)
 ...        row groups 1–7, same story
 905 ms   aggregation map holds 38,412 groups
 906 ms   shuffle write: hash(keys) mod 200 → UnsafeShuffleWriter → 1 data file (1.1 MB) + index
 924 ms   MapStatus (executor 3, 200 block sizes) sent to the driver. Task FINISHED.
          Metrics: input 128 MB, records read 1,046,000, output rows 38,412,
                   shuffle write 1.1 MB, GC time 3 ms, spill 0

1,364 of those, 40 at a time, 34 seconds.

6.4 Before and after

line	Monday's version	what it made Spark do	Wednesday's version	what changed
date filter	`month(pickup_ts) == 3`	wrong column, wrapped in a function: no partition pruning, no pushdown. 16,384 tasks, ~500 GB of columns decoded	`dt.between(...)`	1,364 tasks, 31 directories, 170 GB assigned
fare filter	`fare > 100`	pushed, but to every file	same	pushed, after pruning
zones join	`.join(zones)`	broadcast (size estimate was enough)	`broadcast(zones)`	same, now explicit
drivers join	`.join(drivers)`	sort-merge planned: trips shuffled and sorted by `driver_id`; AQE converted to broadcast only after the shuffle write	`broadcast(drivers)`	no trips shuffle
ranking	`orderBy().limit(10)`	`TakeOrderedAndProject` (she got this right by accident)	same	same
total	14 min, 5 jobs		41 s, 4 jobs

Nothing about the cluster changed. Every second saved came from giving Catalyst something it could reason about: a comparable column instead of a function, a size hint instead of a guess.

6.5 Epilogue: Thursday

Maya closes FIN-2231 with a day to spare. Ravi asks her to explain the Spark UI back to him, and she does it in five sentences:

"spark.read built a plan and listed the files; that's Job 0. The write was the action, and it became a plan with
one shuffle, so two stages, plus a tiny third one for the output file. The two little jobs before it were the
broadcasts. Stage 8 had 1,364 tasks because March is 1,364 files of 128 MB, and each task read three columns,
filtered, probed two hash tables and pre-aggregated. Everything after that was 62 MB."

"Good," Ravi says. "Now: what do you think happens when you call .write? Because that part has its own story." But that's the next article.

Cheat sheets

Does it shuffle?

operation	physical operator	shuffle?	stage boundary?	notes
`spark.read.*`	`FileScan`	no		may run a listing job (> 32 dirs) and, for CSV/JSON with `inferSchema`, a full-scan job
`select`, `withColumn`, `filter`	`Project`, `Filter`	no		fused into the scan's codegen stage
`count()`	`HashAggregate` ×2	tiny	yes	reads zero columns from Parquet
`groupBy().agg()`	`HashAggregate` ×2	yes	yes	partial aggregation first, so shuffle size ≈ groups per task
`distinct()`, `dropDuplicates()`	`HashAggregate` ×2	yes	yes	same as groupBy on all columns
`orderBy()` / `sort()`	`Exchange rangepartitioning` + `Sort`	yes	yes	plus a sampling job
`sortWithinPartitions()`	`Sort global=false`	no
`orderBy().limit(n)`	`TakeOrderedAndProject`	no		map-only top-k
`limit(n)`	`CollectLimit`	no		reads partitions incrementally
`join` (small side ≤ 10 MB)	`BroadcastHashJoin`	no (big side)		small side collected to driver, broadcast
`join` (both large)	`SortMergeJoin`	yes, both sides	yes	AQE may convert to broadcast at runtime
`join` (non-equi)	`BroadcastNestedLoopJoin`	no		O(n×m); needs a broadcastable side
`repartition(n)` / `repartition(col)`	`Exchange roundrobin` / `hashpartitioning`	yes	yes
`coalesce(n)`	`Coalesce`	no		merges partitions in place; can reduce parallelism of the whole stage
`union`	`Union`	no		just concatenates partitions
window functions	`Window` + `Exchange`	yes (unless already partitioned)	yes	plus a sort by the window's order
`cache()` / `persist()`	`InMemoryTableScan`	no		lazy; materialized by the next action
`collect()`, `toPandas()`	`CollectLimit` / result task	no		bounded by `spark.driver.maxResultSize`
`write.*`	`InsertIntoHadoopFsRelation`	no		one file per task unless you repartition first

The settings that appeared in this story

setting	default	what it controls	chapter
`spark.sql.sources.parallelPartitionDiscovery.threshold`	32	directories before file listing becomes a Spark job	1
`spark.sql.parquet.mergeSchema`	false	read one footer vs all footers for the schema	1
`spark.sql.files.maxPartitionBytes`	128 MB	max bytes per scan task	2
`spark.sql.files.openCostInBytes`	4 MB	fake size added per file when bin-packing	2
`spark.sql.parquet.columnarReaderBatchSize`	4096	rows per `ColumnarBatch`	2
`spark.driver.maxResultSize`	1 GB	total size of results a single action may return to the driver	2
`spark.sql.parquet.filterPushdown`	true	hand predicates to the Parquet reader	3
`spark.sql.shuffle.partitions`	200	reducers after a shuffle (before AQE adjusts it)	4
`spark.sql.adaptive.enabled`	true (3.2+)	AQE as a whole	4, 5
`spark.sql.adaptive.coalescePartitions.enabled`	true	merge small post-shuffle partitions	4
`spark.sql.adaptive.advisoryPartitionSizeInBytes`	64 MB	target size when coalescing or splitting	4, 5
`spark.sql.execution.rangeExchange.sampleSizePerPartition`	100	sample points per output partition for range sorts	4
`spark.reducer.maxSizeInFlight`	48 MB	shuffle fetch requests outstanding per reducer	4
`spark.sql.autoBroadcastJoinThreshold`	10 MB	planning-time broadcast decision (file sizes)	5
`spark.sql.adaptive.autoBroadcastJoinThreshold`	= above	runtime broadcast decision (shuffle sizes)	5
`spark.sql.broadcastTimeout`	300 s	how long a broadcast build may take	5
`spark.sql.adaptive.skewJoin.enabled`	true	split skewed join partitions	5
`spark.sql.adaptive.skewJoin.skewedPartitionFactor` / `...ThresholdInBytes`	5 / 256 MB	what counts as skewed	5
`spark.sql.optimizer.dynamicPartitionPruning.enabled`	true	prune partitions using the other side of a join	5

Five questions to ask any plan

What's in PartitionFilters? If your date/region/tenant filter isn't there, you're scanning everything.
What's in ReadSchema? If it lists 40 columns and you use 3, something upstream (a UDF, a select("*") into a cache) is defeating column pruning.
How many Exchanges, and what's below each? Each is a stage boundary and a network copy. A Sort under one means a global sort; two under a join means sort-merge.
Which joins are BroadcastHashJoin? Anything that should be and isn't is a size-estimate problem; fix it with broadcast() or better statistics.
Is isFinalPlan=true? If not, you're reading the initial plan; run the query and look at the SQL tab for what AQE actually did.

Reading the Spark UI like a story

 Jobs tab        Unexpected jobs are the first clue:
                 "Listing leaf files"        → the read listed > 32 directories
                 1,364-task job before a sort → RangePartitioner sampling
                 1-task jobs before a join    → broadcast builds
                 full-scan job on spark.read  → CSV/JSON inferSchema

 Stages tab      Task count vs cores (16,384 tasks on 40 cores = 410 waves).
                 Task duration quartiles: max ≫ 75th percentile = skew.
                 Shuffle Read/Write columns: where the bytes moved.
                 Spill (memory / disk): sorter or aggregate ran out of memory.

 SQL tab         Click the query → the plan as a graph, with live metrics:
                 Scan: files read, partitions read, size, rows out
                 Filter: rows out (compare with scan rows out = pushdown effectiveness)
                 Exchange: shuffle bytes written / fetched
                 Open "Details" at the bottom for the text plan, including isFinalPlan.

 Executors tab   GC time (should be < 10% of task time), storage memory used by
                 broadcasts and caches, failed tasks per executor.

Try it on your laptop

Everything above reproduces on a single machine with local[4], just with smaller numbers. This generates a mini RideHub (about 300 MB) and sets you up to run every explain() in the article:

# pip install pyspark   (3.5.x or 4.0.x)
from pyspark.sql import SparkSession, functions as F

spark = (SparkSession.builder
    .master("local[4]")
    .appName("spark-read-story")
    .config("spark.sql.shuffle.partitions", "200")        # the default; set explicitly so it's visible
    .getOrCreate())

base = "/tmp/ridehub"

# trips: 5M rows over 60 days, partitioned by dt, fares in random order (like RideHub's)
(spark.range(0, 5_000_000)
    .select(
        F.concat(F.lit("t"), F.col("id")).alias("trip_id"),
        (F.rand(1) * 50_000).cast("long").alias("driver_id"),
        (F.rand(2) * 263).cast("int").alias("zone_id"),
        (F.rand(3) * 400 + 3).alias("fare"),
        F.expr("date_add(date'2026-02-01', cast(id % 60 as int))").alias("dt"))
    .repartition(2, "dt")                                  # ~2 files per day
    .write.partitionBy("dt").mode("overwrite").parquet(f"{base}/trips"))

# drivers: 50k rows with some padding so the file is bigger than the broadcast threshold
(spark.range(0, 50_000)
    .select(
        F.col("id").alias("driver_id"),
        F.concat(F.lit("Driver "), F.col("id")).alias("driver_name"),
        F.sha2(F.col("id").cast("string"), 512).alias("licence_hash"),
        F.sha2((F.col("id") * 7).cast("string"), 512).alias("payout_token"))
    .write.mode("overwrite").parquet(f"{base}/drivers"))

# zones: tiny
(spark.range(0, 263)
    .select(F.col("id").cast("int").alias("zone_id"),
            F.concat(F.lit("Zone "), F.col("id")).alias("zone_name"))
    .write.mode("overwrite").parquet(f"{base}/zones"))

Then walk the chapters, with the Spark UI open at http://localhost:4040:

trips   = spark.read.parquet(f"{base}/trips")       # Chapter 1: Jobs tab shows "Listing leaf files" (60 dirs > 32)
drivers = spark.read.parquet(f"{base}/drivers")
zones   = spark.read.parquet(f"{base}/zones")

trips.count()                                        # Chapter 2: ReadSchema struct<>, 2 stages
trips.explain(True)

march = trips.filter(F.col("dt").between("2026-03-01", "2026-03-31")).filter(F.col("fare") > 100)
march.explain()                                      # Chapter 3: PartitionFilters vs PushedFilters
trips.filter(F.month("dt") == 3).explain()           #            the trap: PartitionFilters: []

march.orderBy(F.desc("fare")).explain()              # Chapter 4: Exchange rangepartitioning + a sampling job
march.orderBy(F.desc("fare")).limit(10).explain()    #            TakeOrderedAndProject

march.join(zones, "zone_id").explain()               # Chapter 5: BroadcastHashJoin
spark.conf.set("spark.sql.autoBroadcastJoinThreshold", -1)
march.join(drivers, "driver_id").explain()           #            SortMergeJoin with two Exchanges
spark.conf.set("spark.sql.autoBroadcastJoinThreshold", "10MB")

# Run the join and compare the initial plan with the final one in the SQL tab (isFinalPlan=true)
march.join(drivers, "driver_id").count()

Two experiments worth doing on top:

Turn AQE off (spark.conf.set("spark.sql.adaptive.enabled", False)) and re-run the sort: you'll get 200 output files instead of a handful, and the join won't convert to broadcast.
Rewrite trips sorted by fare (.sortWithinPartitions("fare") before the write) and compare the scan's rows output metric in the SQL tab for fare > 300. That's Parquet statistics finally doing something.

The Distributed Compute Playground: Everything You Can Plug In and Tune in an Apache Spark Job

Aniketh Deshpande — Thu, 01 Oct 2026 09:34:42 +0000

TL;DR: An Apache Spark job is not one thing. It is a stack of nine layers, and almost every layer has
parts you can swap: where executors come from, how the code is written, which engine runs the plan, how joins
and shuffles are decided, where shuffle files live, how memory is divided, how the scheduler reacts to slow and
dying nodes, how bytes sit on disk, and how files are laid out so the planner can skip them. This article maps
all of those options by the problem each one solves, with the --conf line or PySpark call for every one of them.

⏱️ 40-minute read. It is the reference companion to
What actually happens when you call spark.read?, which followed one job through the
internals. This one is the parts catalogue. Everything here was checked against Spark 3.5 and Spark 4.0
documentation; where a platform (Databricks, EMR, Dataproc) owns a feature, it says so.

🎮 Rather break a Spark job than read about one? Every component in this article is a chip you can drag in the
free, open-source Distributed Compute Playground.
Build a job layer by layer and it shows you the explain() plan, the stages, the task counts and the runtime you
just caused, then hands you 25 on-call tickets to fix. If it clicks, a ⭐ means the world.
(More about it at the end)

How to read this article
Layer 1: Cluster manager
Layer 2: Code and APIs
Layer 3a: Execution engine
Layer 3b: Planner, joins and AQE
Layer 4a: Shuffle
Layer 4b: Memory, caching and serialization
Layer 5: Scheduling and resilience
Layer 6a: File and table formats
Layer 6b: Data layout and the read path
Combinations that are worth more than their parts
Six starter profiles
Cheat sheet: symptom to lever
Play it: the Distributed Compute Playground 🎮
Further reading

How to read this article

The previous article told a story: Maya, a new data engineer at RideHub, inherits a report that used to take six minutes and now takes fifty, and we followed one DataFrame through Catalyst, the file index, the scheduler and the shuffle to find out why. If you have not read it, you do not need to; this article stands alone. But the dataset is the same, so when an example says trips, it means 2 TB of Parquet partitioned by dt, 16,000 files, one year of ride-hailing history.

This time the question is different. Not what happens, but what can I change.

The nine layers

Every Spark job, whether it is a notebook cell or a 5 TB nightly join, sits on the same stack:

 ┌──────────────────────────────────────────────────────────────────────┐
 │  L1  CLUSTER MANAGER      local[*] · Standalone · YARN · K8s · serverless │
 ├──────────────────────────────────────────────────────────────────────┤
 │  L2  CODE & APIs          DataFrame · RDD · UDF ladder · Connect ·      │
 │                           Thrift · Structured Streaming                 │
 ├────────────────────────────────┬─────────────────────────────────────┤
 │  L3  EXECUTION ENGINE          │  L3  PLANNER, JOINS & AQE           │
 │      Catalyst+Tungsten (JVM)   │      AQE · broadcast · DPP · CBO ·  │
 │      Gluten · Comet · Photon · │      salting · bucketing ·          │
 │      RAPIDS                    │      shuffle.partitions · pushdown  │
 ├────────────────────────────────┼─────────────────────────────────────┤
 │  L4  SHUFFLE                   │  L4  MEMORY, CACHING, SERIALIZATION │
 │      ESS · tracking · Celeborn │      executor shape · overhead ·    │
 │      Uniffle · Magnet · disks  │      off-heap · cache levels · Arrow│
 ├────────────────────────────────┴─────────────────────────────────────┤
 │  L5  SCHEDULING & RESILIENCE  dynamic allocation · speculation · FAIR · │
 │                               retries · decommission · spot · history   │
 ├────────────────────────────────┬─────────────────────────────────────┤
 │  L6  FILE / TABLE FORMAT       │  L6  DATA LAYOUT & READ PATH        │
 │      Parquet · ORC · Avro ·    │      partitionBy · bucketBy ·       │
 │      CSV · JSON · Delta ·      │      Z-ORDER · compaction ·         │
 │      Iceberg · Hudi · Hive     │      schema · committers · stats    │
 └──────────────────────────────────────────────────────────────────────┘

Each section below takes one layer, lists every part you can plug into it, and for each part answers the same four questions:

What problem does it solve?
What does it cost? (Every lever has a price. Some are hidden.)
What does it need? (Prerequisites and conflicts. Dynamic allocation without a shuffle service is a config error, not a tuning choice.)
How do you turn it on? (The --conf or the PySpark call.)

The eleven dimensions

To compare parts that do completely different things, it helps to score them against the same questions. The playground uses eleven, and this article uses them informally too:

Dimension	The question it answers
Batch throughput	How many bytes per hour can the nightly job chew through?
Interactive latency	How fast does a dashboard query come back?
Shuffle efficiency	How much data crosses the network, and how painful is that?
Memory safety	How close is the driver or an executor to an OutOfMemoryError?
Skew resilience	What happens when one key holds 30% of the rows?
I/O and small-file efficiency	How much of the time is spent listing, opening and parsing files?
Fault tolerance	What happens when a node disappears mid-stage?
Cost efficiency	Dollars per terabyte processed.
Python friendliness	How much does writing it in Python cost compared with SQL?
Streaming fitness	Can it run forever with bounded state?
Operational simplicity	How many things can page you at 04:00?

Most of the dangerous settings in this article look great on one dimension and are catastrophic on another. spark.sql.autoBroadcastJoinThreshold = 2g is a shuffle-efficiency win and a memory-safety disaster. The point of scoring against all eleven is that you notice.

Three kinds of "dangerous"

Throughout, settings are marked three ways:

⚠️ Warning: a legitimate tool that is wrong more often than it is right. Example: MEMORY_ONLY persistence on a dataset bigger than RAM.
🧨 Danger: a setting that will eventually take production down, usually on the first busy day after the person who set it has left. Example: spark.task.maxFailures = 1.
ℹ️ Note: a prerequisite or platform limit that is easy to miss. Example: Kubernetes has no external shuffle service.

Layer 1: Cluster manager

The cluster manager answers one question: where do executors come from, and what happens to their disks when they go away? That second half decides more than people expect. It is the reason dynamic allocation works out of the box on YARN and needs a workaround on Kubernetes.

                      driver
                        │ asks for N executors
            ┌───────────┴────────────┐
            ▼                        ▼
   ┌─────────────────┐      ┌─────────────────┐
   │ cluster manager │      │ cluster manager │
   │  (YARN / K8s /  │ ...  │   decides which │
   │  standalone)    │      │   node, how big │
   └────────┬────────┘      └─────────────────┘
            │ launches
   ┌────────▼────────┐  ┌─────────────────┐  ┌─────────────────┐
   │ executor (JVM)  │  │ executor (JVM)  │  │ executor (JVM)  │
   │ cores · heap ·  │  │                 │  │                 │
   │ LOCAL DISK ◄────┼──┼── shuffle files live here, and die with the node
   └─────────────────┘  └─────────────────┘  └─────────────────┘

`local[*]`: one JVM

Driver and executors share one process on your laptop. * means one task slot per core.

Solves: running tests, reading explain() output, learning. Everything in this article can be observed on local[*] with a few hundred megabytes of Parquet.
Costs: no fault tolerance, one machine of RAM, one machine of disk for shuffle. Scores poorly on throughput and resilience by definition.
ℹ️ local[*] is a learning tool, not a deployment target. If the nightly job "works on my laptop" at 2 TB, it is because the laptop has the data cached and nobody is sharing it.

spark.master=local[*]

Spark Standalone

Spark's own master/worker cluster manager. No Hadoop, no Kubernetes, one process per node and a web UI.

Solves: a dedicated Spark cluster with the least moving parts. The worker can host the external shuffle service, so dynamic allocation works.
Costs: no queues, no multi-tenancy worth the name, no preemption. One application tends to take the whole cluster unless you cap spark.cores.max.

spark.master=spark://master:7077

Docs: Spark Standalone Mode.

Hadoop YARN

The classic. YARN runs a ResourceManager and a NodeManager per node; the NodeManager can host Spark's external shuffle service as an auxiliary service.

Solves: queues, capacity limits, preemption between teams, and shuffle files that survive executor removal. The reason the "Nightly ETL starter" profile at the end of this article uses YARN is that everything simply works: spark.shuffle.service.enabled=true, spark.dynamicAllocation.enabled=true, done.
Costs: a Hadoop installation to operate. Executor sizes are quantised by yarn.scheduler.minimum-allocation-mb. Container start-up is slower than a pod.
Deploy mode matters: cluster puts the driver inside the YARN application (survives your SSH session dying); client keeps it where you ran spark-submit.

spark.master=yarn
spark.submit.deployMode=cluster

Docs: Running Spark on YARN.

Kubernetes

Executors are pods. You bring a container image; Spark asks the API server for pods and tears them down when done.

Solves: elasticity, one platform for Spark and everything else, custom images with exactly the Python packages a job needs, namespaces and quotas.
Costs: ℹ️ there is no external shuffle service on Kubernetes. When an executor pod is removed, its shuffle files go with it. That means dynamic allocation needs either shuffle tracking (keep executors alive while their shuffle data is needed) or a remote shuffle service (Celeborn, Uniffle), both covered in Layer 4a. Expect to spend time on image builds, service accounts and volume mounts for spark.local.dir.

spark.master=k8s://https://k8s-api:6443
spark.kubernetes.container.image=registry/spark-py:3.5
spark.kubernetes.namespace=data-eng
spark.kubernetes.authenticate.driver.serviceAccountName=spark

Docs: Running Spark on Kubernetes.

Managed serverless Spark

Dataproc Serverless, EMR Serverless and Databricks serverless compute remove the cluster entirely: you submit a job, the platform sizes and scales it.

Solves: operational simplicity, start-up time, autoscaling you did not have to configure, and the "who left the 40-node cluster running over the weekend" bill.
Costs: ℹ️ a ceiling on what you may change. Executor shapes are chosen from a menu, some plugins are unavailable, and a few spark.* settings are silently overridden. Read the platform limits before relying on a knob from this article.
Platform-specific: it is one of the parts the playground marks as not on open-source Spark.

Docs: Dataproc Serverless, EMR Serverless, Databricks serverless compute.

Choosing

You want	Pick	Then also
To learn, test, read plans	`local[*]`	nothing
One team, one cluster, no Hadoop	Standalone	external shuffle service
Multi-tenant queues, battle-tested dynamic allocation	YARN	external shuffle service, push-based shuffle
Cloud-native, custom images, elasticity	Kubernetes	shuffle tracking or Celeborn / Uniffle
Nobody to operate a cluster	Serverless	accept the platform's limits

Layer 2: Code and APIs

How the job is written decides what the optimizer can see. This is the layer where a one-line change (a @udf decorator) can make a job 50× slower, and where the fix is usually a different decorator.

DataFrame / SQL API

Declarative plans. You describe what, Catalyst decides how: column pruning, predicate pushdown, join reordering, whole-stage code generation. Python, Scala, Java and SQL all compile to the same logical plan, so a PySpark DataFrame pipeline runs at JVM speed.

df = spark.read.parquet("s3a://ridehub/trips/")
report = (df.filter(F.col("dt") >= "2026-03-01")
            .groupBy("zone_id").agg(F.sum("fare")))

This is the default and the baseline. Everything else in this layer is measured against it.

RDD API

The original abstraction: a distributed collection of opaque objects with map, filter, reduceByKey.

Solves: full control over partitioning and per-record logic, access to things the DataFrame API hides (custom partitioners, mapPartitions with external state).
⚠️ Costs: RDDs skip Catalyst. No column pruning, no pushdown, no AQE, no codegen. In Python, every row is pickled from the JVM to a Python worker process and back. On a 2 TB Parquet scan that reads six columns out of forty, the RDD path reads all forty and deserializes them one object at a time.
Use it when you genuinely need it, and convert back to a DataFrame as early as possible.

Docs: RDD Programming Guide.

The Python UDF ladder

This is the single most important diagram in this layer. Every rung is "run my Python function on every row"; the difference is how the rows get to Python.

                 JVM executor                           Python worker
             ┌────────────────┐                      ┌────────────────┐
 row UDF     │ row ─► pickle ─┼──── one row ────────►│ unpickle ─► f()│   ~10–100× slower
 @udf        │ row ─► pickle ─┼──── one row ────────►│ unpickle ─► f()│   than native
             └────────────────┘                      └────────────────┘

 Arrow UDF   ┌────────────────┐                      ┌────────────────┐
 @udf(       │ 10,000 rows ───┼─── Arrow batch ─────►│ f() per value  │   same code,
 useArrow)   └────────────────┘                      └────────────────┘   batched transfer

 pandas UDF  ┌────────────────┐                      ┌────────────────┐
 @pandas_udf │ 10,000 rows ───┼─── Arrow batch ─────►│ f(pd.Series)   │   vectorized:
             └────────────────┘                      └────────────────┘   numpy speed

 mapInPandas ┌────────────────┐                      ┌────────────────┐
             │ whole partition┼─── Arrow batches ───►│ f(iter[pd.DF]) │   model.predict()
             └────────────────┘                      └────────────────┘   on a whole batch

Row-at-a-time Python UDF (@F.udf). The rung everyone starts on.

@F.udf("boolean")
def is_premium(fare):
    return fare is not None and fare > 100

df.filter(is_premium("fare"))        # BatchEvalPython in the plan

⚠️ Typically 10–100× slower than the equivalent F.col("fare") > 100. Worse, Catalyst treats it as a black box: nothing above it in the plan can be pushed down into the scan, so a UDF in a filter means the scan reads everything. Look for BatchEvalPython in explain(); it is the smoking gun.

Arrow-optimized Python UDF (Spark 3.5+, @F.udf(..., useArrow=True)). Same function, same row-by-row semantics, but rows cross the boundary in Arrow batches instead of pickle. A free 2–4× for the price of one keyword argument.

@F.udf("double", useArrow=True)
def tip_ratio(tip, fare):
    return tip / fare if fare else None

Pandas UDF (@F.pandas_udf). Your function receives a pd.Series (or several) per Arrow batch and returns one. This is where Python logic stops being slow, because pandas and numpy do the inner loop in C.

@F.pandas_udf("double")
def zscore(v: pd.Series) -> pd.Series:
    return (v - v.mean()) / v.std()

mapInPandas / applyInPandas. Hand an iterator of pandas DataFrames (a whole partition, or a whole group) to a function. The natural home for model scoring: load the model once per partition, call predict on a batch.

def score(batches):
    for pdf in batches:
        pdf["p"] = model.predict(pdf[FEATURES])
        yield pdf

df.mapInPandas(score, schema="trip_id string, p double")

All of the Arrow-based rungs still block pushdown for expressions above them. Put filters before the UDF in the pipeline, and make sure the UDF's inputs are the only columns that survive to that point.
All of them need memory outside the JVM heap for the Python worker and the Arrow buffers. See memoryOverhead in Layer 4b. The classic failure is Container killed by YARN for exceeding memory limits with a heap that is barely used.

Docs: Apache Arrow in PySpark.

pandas API on Spark

import pyspark.pandas as ps gives you pandas syntax on a Spark DataFrame. Good for migrating notebooks.

import pyspark.pandas as ps
psdf = ps.read_parquet("s3a://ridehub/trips/")
psdf.groupby("zone_id")["fare"].sum()

ℹ️ pandas has a sequential index; Spark does not. The default index type builds one, and some operations (iloc, shift) collapse to a single partition to do it. Set ps.set_option("compute.default_index_type", "distributed") unless you need pandas-exact ordering.

Docs: pandas API on Spark.

Scala / Java UDF

Runs inside the JVM, so there is no serialization tax. Still opaque to Catalyst: no pushdown through it, no codegen of the expression itself. Cheaper than any Python rung, more expensive than a built-in function.

Built-in functions only

The top of the ladder is not a UDF at all. pyspark.sql.functions has over 400 functions, and Spark 3.5 added F.call_function plus higher-order functions (transform, filter, aggregate) over arrays and maps. Almost every "I need a UDF for this" has a native expression, and the native expression is visible to Catalyst.

df.filter(F.col("fare") > 100).withColumn("tip_ratio", F.col("tip") / F.col("fare"))

Docs: Functions.

Spark Connect

Spark 3.4+ splits the client from the driver. Your notebook holds a thin gRPC client; the driver (and the JVM, and the dependency hell) runs on a server. Clients can be upgraded independently, one bad client cannot crash the driver, and pip install pyspark-connect is all a laptop needs.

spark.remote=sc://spark-connect:15002

ℹ️ RDD APIs and SparkContext calls are not available over Connect. If your code uses sc.parallelize, it is not a Connect job yet.

Docs: Spark Connect overview.

Thrift server (JDBC / BI)

A long-running SparkSession that speaks HiveServer2's protocol, so Tableau, Looker or beeline can run SQL against your lake.

Solves: dashboards. The session stays warm, caches survive between queries, and the file index for a table is built once.
Costs: one application shared by everyone. Without FAIR scheduler pools (Layer 5), one analyst's 30-minute query starves every dashboard.

Docs: Distributed SQL Engine.

Structured Streaming and its three companions

The DataFrame API over an unbounded source. Micro-batch by default (continuous mode exists but is rarely the right choice), checkpointed, exactly-once with the right sink.

events = (spark.readStream.format("kafka")
    .option("kafka.bootstrap.servers", "kafka:9092")
    .option("subscribe", "rides").load())

Three parts turn a streaming demo into a streaming pipeline. They only make sense together with Structured Streaming, which is why the playground refuses to activate them without it.

RocksDB state store. Streaming aggregations, stream-stream joins and dropDuplicates keep state. The default state store keeps it in the JVM heap, which works until the state is bigger than the heap and the executor dies in a GC storm on Friday evening. RocksDB keeps state off-heap on local disk with changelog checkpointing.

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

Watermark and trigger tuning. Without a watermark, a windowed aggregation keeps every window forever, because a late event could still arrive. withWatermark("event_ts", "10 minutes") tells Spark when a window can be finalised and its state dropped. The trigger interval decides the latency/throughput trade; pick it on purpose.

(events.withWatermark("event_ts", "10 minutes")
  .groupBy(F.window("event_ts", "5 minutes"), "zone_id").count()
  .writeStream.trigger(processingTime="1 minute") ...)

foreachBatch with an idempotent sink. File sinks and Delta/Iceberg are exactly-once by design. A JDBC table or a MERGE is not, unless you make each micro-batch idempotent, keyed by batchId, so a replay after a crash does not double-count.

def upsert(batch_df, batch_id):
    batch_df.createOrReplaceTempView("b")
    spark.sql("""MERGE INTO rides t USING b ON t.id = b.id
                 WHEN MATCHED THEN UPDATE SET * WHEN NOT MATCHED THEN INSERT *""")

agg.writeStream.foreachBatch(upsert).option("checkpointLocation", chk).start()

Docs: Structured Streaming Programming Guide, RocksDB state store.

Choosing

Situation	Use
Anything expressible in SQL	DataFrame API with built-in functions. Always first.
Python logic over a column	pandas UDF; Arrow UDF if the logic is inherently scalar
Model scoring	`mapInPandas`, model loaded once per partition
Migrating a pandas notebook	pandas API on Spark, with the distributed index
Custom partitioning or per-partition state in the JVM	RDD, then back to DataFrame
Many notebooks, one cluster	Spark Connect
BI tools	Thrift server + FAIR pools + cache
Kafka in, table out, forever	Structured Streaming + RocksDB + watermark + idempotent sink

Layer 3a: Execution engine

The engine is the one part of the stack you cannot leave empty. It is what turns a physical plan into CPU instructions. For fifteen years that meant one thing: Spark's own JVM engine. Since about 2022 there have been real alternatives, and they all work the same way: a Spark plugin intercepts the physical plan and replaces operators it knows how to run with native (C++, Rust or GPU) versions.

   Catalyst physical plan
            │
            ▼
   ┌─────────────────────┐        supported operator?
   │  engine plugin      │──────────────┬──────────────┐
   │  (Gluten / Comet /  │             yes             no
   │   Photon / RAPIDS)  │              │               │
   └─────────────────────┘              ▼               ▼
                               native columnar     JVM row engine
                               operator            (Tungsten codegen)
                                      │               ▲
                                      └── columnar→row conversion ──┘
                                          (the hidden cost of every fallback)

Catalyst + Tungsten (JVM): stock Spark

Whole-stage code generation (operators fused into one Java method per stage), a vectorized Parquet/ORC reader, the unified memory manager, and UnsafeRow binary rows that skip Java object overhead.

Solves: everything, everywhere, with every UDF and every data source. The safest default and the baseline every native engine benchmarks against.
Costs: CPU-bound SQL on wide tables runs 2–4× slower than the native engines. GC pauses on large heaps. That is the whole pitch of the alternatives.

Gluten + Velox

Apache Gluten (incubating) plugs Meta's Velox C++ library under the Spark API. It also ships a columnar shuffle manager so data stays in Arrow-like columnar batches across the exchange.

Solves: CPU-bound SQL and DataFrame ETL. Typical published numbers are 2–3× on TPC-H/DS-style workloads; the playground models it as about 2×.
Costs: ℹ️ operators and expressions Velox does not support fall back to the JVM, with a columnar-to-row conversion on each side of the fallback. A Python UDF in the middle of a Gluten plan is slower than the same UDF in a plain JVM plan. It also needs off-heap memory configured, because Velox allocates outside the heap; forgetting spark.memory.offHeap.size is the most common first-day failure.

spark.plugins=org.apache.gluten.GlutenPlugin
spark.shuffle.manager=org.apache.spark.shuffle.sort.ColumnarShuffleManager
spark.memory.offHeap.enabled=true
spark.memory.offHeap.size=16g

Apache Comet (DataFusion)

Apache DataFusion Comet is the Rust-based equivalent from the Arrow project: a native Parquet scan, native operators and a native columnar shuffle, for Spark 3.4 through 4.0.

Solves: the same problem as Gluten with a smaller runtime, Rust memory safety and tight Arrow integration.
Costs: ℹ️ operator coverage is still growing. Check the compatibility guide for your plan, and run explain(): Comet-executed operators appear with a Comet prefix, so you can see exactly where it fell back.

spark.plugins=org.apache.spark.CometPlugin
spark.comet.exec.shuffle.enabled=true
spark.memory.offHeap.enabled=true
spark.memory.offHeap.size=8g

Photon

Databricks' proprietary C++ engine. On Databricks it is a checkbox; off Databricks it does not exist.

Solves: the same CPU-bound SQL problem, with the best integration of the four (Delta, SQL warehouses, streaming support) and no configuration.
Costs: platform lock-in and a premium DBU rate. ℹ️ UDFs and RDDs still fall back to the JVM, so a Photon cluster running a row-at-a-time Python UDF is paying the premium for nothing. Photon on YARN is not a configuration; the playground calls this a platform conflict and tells you the job runs nowhere.

Docs: Photon.

RAPIDS Accelerator (GPU)

NVIDIA's RAPIDS Accelerator for Apache Spark runs scans, joins, aggregates and sorts on GPUs, optionally with a UCX-based GPU-to-GPU shuffle.

Solves: very wide, very numeric ETL and joins where the GPU's memory bandwidth dwarfs the CPU's. On the right workload it is the fastest thing here by a wide margin.
⚠️ Costs: GPU memory is small (tens of GB) and unforgiving; wide strings, UDFs and unsupported expressions spill or fall back, and the fallbacks are expensive because the data has to come back off the GPU. GPU nodes cost several times more per hour, so the speed-up has to be large to pay for itself. Benchmark your plan before committing a budget. Pair it with stage-level scheduling (Layer 5) so only the stage that benefits runs on GPU executors.

spark.plugins=com.nvidia.spark.SQLPlugin
spark.rapids.sql.enabled=true
spark.executor.resource.gpu.amount=1
spark.task.resource.gpu.amount=0.25

Choosing an engine

Workload	First choice	Why
Anything with Python UDFs or RDDs	JVM	native engines fall back anyway and pay a conversion each way
CPU-bound SQL/DataFrame ETL on open-source Spark	Gluten or Comet, with off-heap sized	2–3× for the cost of a plugin and some memory config
Same, on Databricks	Photon	zero configuration, best Delta integration
Wide numeric ETL, GPU nodes already available	RAPIDS, scoped with stage-level scheduling	memory bandwidth
Streaming	JVM (or Photon on Databricks)	the open-source native engines have thin streaming support

A rule that holds across all of them: run explain() and count the fallbacks. A native engine that executes 95% of the plan is a win; one that executes 40% of it, with conversions at every boundary, is slower than stock Spark.

Layer 3b: Planner, joins and AQE

This is the layer the previous article spent most of its time in, so here is the compressed version with every knob attached. Catalyst produces a physical plan from statistics it can see at planning time; Adaptive Query Execution re-plans at every shuffle boundary using the sizes it actually observed. Most of the parts below either feed the planner better information or tell AQE what it is allowed to change.

Adaptive Query Execution

On by default since Spark 3.2. At each shuffle boundary, AQE looks at the real map output sizes and may coalesce partitions, switch join strategies or split skewed partitions. In explain() you see AdaptiveSparkPlan isFinalPlan=false before the first action and isFinalPlan=true after.

spark.sql.adaptive.enabled=true

Three sub-features sit under it. They do nothing without it.

Coalesce shuffle partitions. The default spark.sql.shuffle.partitions is 200, whatever the data size. AQE merges small post-shuffle partitions up to an advisory size, so a 1 GB aggregate ends up in 16 reducers instead of 200 tiny ones, and the write produces 16 files instead of 200.

spark.sql.adaptive.coalescePartitions.enabled=true
spark.sql.adaptive.advisoryPartitionSizeInBytes=64m

Skew join handling. If one post-shuffle partition is more than five times the median and more than 256 MB, AQE splits it into several reader tasks on the big side and duplicates the matching partition on the small side. The three-hour straggler becomes twelve two-minute tasks. This is the single most valuable automatic fix in modern Spark.

spark.sql.adaptive.skewJoin.enabled=true
spark.sql.adaptive.skewJoin.skewedPartitionFactor=5
spark.sql.adaptive.skewJoin.skewedPartitionThresholdInBytes=256m

Runtime broadcast conversion. The planner chose a sort-merge join because the dimension's file size looked big; after the shuffle, the dimension's actual bytes are 4 MB. AQE converts to a broadcast hash join on the fly, and the local shuffle reader lets the big side read its own map outputs instead of a real exchange.

spark.sql.adaptive.autoBroadcastJoinThreshold=10m
spark.sql.adaptive.localShuffleReader.enabled=true

Docs: Adaptive Query Execution.

How Spark picks a join

Before the knobs, the decision tree, because every knob below moves you to a different leaf:

                 is one side ≤ autoBroadcastJoinThreshold (10 MB)
                 or hinted BROADCAST?
                        │
            yes ────────┴──────── no
             │                     │
   BroadcastHashJoin       equi-join and preferSortMergeJoin=false
   · build side collected  and one side small enough per partition?
     on the DRIVER, sent             │
     to every executor      yes ─────┴───── no
   · zero Exchange for       │                │
     the big side     ShuffledHashJoin   SortMergeJoin
                      · shuffle both     · shuffle both, sort both
                      · hash the small   · spills gracefully
                        side per task    · the default for big-to-big
                                              │
                                   both sides bucketed on the key,
                                   same bucket count?
                                              │
                                     yes ─────┴───── no
                                      │                │
                              SortMergeJoin         SortMergeJoin
                              with NO Exchange      with 2 Exchanges

Broadcast hint on small tables

Tell the planner the dimension is small instead of letting it guess from directory size. The directory size estimate is what fooled the planner in the opening story: a 50 MB table that compressed to 12 MB on disk still looked bigger than the 10 MB threshold.

trips.join(F.broadcast(zones), "zone_id")
# or in SQL:  SELECT /*+ BROADCAST(z) */ ...

Solves: a sort-merge join with two exchanges becomes a broadcast join with zero exchanges on the fact side.
Costs: the build side is collected on the driver and sent to every executor. Hint a table that turns out to be 3 GB and you have found the fastest way to kill a driver.

`autoBroadcastJoinThreshold = 100 MB`

Raise the automatic threshold from 10 MB to 100 MB, so moderately sized dimensions broadcast without hints.

Solves: a star schema with a handful of 20–80 MB dimensions, none of which anyone remembered to hint.
ℹ️ Costs: 100 MB of Parquet can be a gigabyte of Java objects once decompressed and hashed on the driver. Raise spark.driver.memory at the same time, every time.

spark.sql.autoBroadcastJoinThreshold=100m
spark.driver.memory=8g

🧨 `autoBroadcastJoinThreshold = 2 GB`

"The executors have plenty of RAM." They do; the driver does not, and the driver is where every broadcast relation is built first. Broadcast builds are capped at 8 GB and 512 million rows, and spark.sql.broadcastTimeout defaults to 300 seconds. A 2 GB threshold works on the quiet day the setting was tested and dies on the first busy one with Could not execute broadcast in 300 secs or a driver OOM, taking every job in the application with it.

spark.sql.autoBroadcastJoinThreshold=2g     # don't

Dynamic partition pruning

On by default since 3.0. In a star-schema join where the dimension is filtered (calendar.quarter = 'Q1'), DPP runs the dimension side first, collects the matching keys, and uses them to prune the fact table's partitions before scanning. In explain() the fact scan shows PartitionFilters: [dynamicpruningexpression(...)].

Solves: the "the filter is on the calendar table, so the 5 TB fact scan reads two years" problem. Requires the fact table to be partitioned by the join key.

spark.sql.optimizer.dynamicPartitionPruning.enabled=true

Cost-based optimizer and statistics

Catalyst's default size estimates come from file sizes. ANALYZE TABLE stores row counts, distinct counts and min/max per column in the catalog; with CBO on, the planner reorders multi-way joins and estimates join output sizes properly.

Solves: wrong join order in three-way-plus joins, and dimensions that broadcast only after a filter because the planner now knows how selective the filter is.
Costs: statistics go stale, and ANALYZE on a 2 TB table is itself a job. Table formats (Delta, Iceberg) carry per-file statistics that remove most of the need.

spark.sql.cbo.enabled=true
spark.sql.cbo.joinReorder.enabled=true

spark.sql("ANALYZE TABLE trips COMPUTE STATISTICS FOR ALL COLUMNS")

Prefer shuffled hash join

By default Spark prefers sort-merge over shuffled hash because SMJ spills gracefully. Turning the preference off skips the sort when one side is much smaller.

ℹ️ The build side of a shuffled hash join must fit in memory per task. Keep it for clearly asymmetric joins where you have measured the gain.

spark.sql.join.preferSortMergeJoin=false

Key salting for hot keys

The manual skew fix, from before AQE existed and still needed when the skew is in an aggregation rather than a join, or when the hot key is a NULL. Append a random suffix to the hot key on the big side, explode the small side into every suffix, join on both.

big   = big.withColumn("salt", (F.rand() * 16).cast("int"))
small = small.withColumn("salt", F.explode(F.array([F.lit(i) for i in range(16)])))
big.join(small, ["key", "salt"])

Costs: code complexity, the small side grows 16×, and the next hot key will not announce itself. Prefer AQE skew handling when the shape allows it.

Bucketed joins

Both tables written with bucketBy(N, key) into a metastore table. Rows with the same key hash land in the same bucket number on both sides, so the join needs no Exchange at all. With Iceberg, the same idea is called a storage-partitioned join (Spark 3.3+).

df.write.bucketBy(256, "driver_id").sortBy("driver_id").saveAsTable("trips_b")

Solves: the recurring 5 TB-to-5 TB join on the same key, every night. The shuffle is paid once at write time instead of every read.
Costs: both sides must agree on bucket count and column, the write is a shuffle, and plain Parquet directories cannot carry bucketing metadata (hence the Hive metastore or Iceberg prerequisite).

Tune `spark.sql.shuffle.partitions`

The default 200 reducers is right for roughly 25 GB of shuffle at 128 MB each. For 1.2 TB of shuffle it means 6 GB per reducer and spills everywhere; for 200 MB it means 200 one-megabyte files. Size it to the data, or let AQE coalescing handle the small end.

spark.sql.shuffle.partitions=1200      # ≈ shuffle bytes / 128 MB

🧨 `spark.sql.shuffle.partitions = 1`

"I only want one output file." Now every aggregation and every join in the application runs on a single task, on a single core, with a single task's memory. Use coalesce(1) on the final write if you must have one file, and leave the shuffle alone.

Pushdown-friendly predicates

Not a config: a habit. The planner can only prune partitions and push filters into Parquet when the predicate is a plain comparison on the partition column or a stats-bearing column.

# ✓ prunes partitions: PartitionFilters: [dt >= 2026-03-01, dt <= 2026-03-31]
trips.filter(F.col("dt").between("2026-03-01", "2026-03-31"))

# ✗ scans everything: the filter is on a derived expression over a different column
trips.filter(F.month("pickup_ts") == 3)

Check explain() for PartitionFilters: [] and PushedFilters: []. Empty brackets on a partitioned table are the most expensive two characters in Spark.

Choosing

Symptom	Lever
200 tiny output files	AQE coalesce, or `coalesce()` before the write
One task runs for hours, the rest finish in minutes	AQE skew join; salting if it is an aggregation
`SortMergeJoin` against a table you know is small	broadcast hint; or CBO stats so the planner knows too
Star join scans the whole fact table	DPP + partition the fact by the join key
Same big join every night	bucket both tables
`PartitionFilters: []`	rewrite the predicate on the partition column
Driver OOM during a join	your broadcast threshold is too high

Layer 4a: Shuffle

A shuffle is the moment Spark's "every task is independent" model breaks: to group by key, every reducer needs rows from every mapper. Where the intermediate files live decides whether losing an executor costs you a task or a stage, and whether you can release executors at all.

        mappers write                       reducers read
   ┌──────────────┐                     ┌──────────────┐
   │ executor A   │  shuffle files      │ executor C   │
   │ task 1 ──────┼──► [p0 p1 p2 … ]    │ reducer p0 ◄─┼── fetches p0 from A and B
   └──────────────┘        │            └──────────────┘
   ┌──────────────┐        │  WHERE do these live?
   │ executor B   │        │   (a) executor's local disk  → die with the executor
   │ task 2 ──────┼──► [p0 p1 p2 … ]    (b) node's shuffle service → survive the executor
   └──────────────┘                      (c) remote shuffle cluster → survive the NODE

External shuffle service (YARN, Standalone)

A long-running process per node that serves shuffle files on behalf of executors that have exited.

Solves: executors can be removed mid-job (dynamic allocation) without losing the map outputs they produced. Also survives an executor crash: the stage does not need to be recomputed.
Needs: YARN (as a NodeManager auxiliary service) or Standalone (in the worker). ℹ️ Kubernetes has no equivalent.
Does not solve: losing the node. The files are still on that node's disk.

spark.shuffle.service.enabled=true

Shuffle tracking (Kubernetes)

The Kubernetes workaround for dynamic allocation without a shuffle service: Spark tracks which executors hold shuffle files that are still needed, and refuses to release those.

Solves: scale-up on Kubernetes, and scale-down for executors whose shuffle data has been consumed.
ℹ️ Costs: scale-down is slow and partial. An executor that produced map output for a long stage stays alive until that stage's reducers have all read it. With a 120 s timeout you still pay for idle pods.

spark.dynamicAllocation.shuffleTracking.enabled=true
spark.dynamicAllocation.shuffleTracking.timeout=120s

Apache Celeborn (remote shuffle)

Apache Celeborn runs a separate cluster of shuffle servers. Mappers push their output to Celeborn workers, which merge it per reducer; executors keep nothing on local disk.

Solves: three problems at once. Executors become stateless, so dynamic allocation can release any of them instantly and spot nodes can vanish without triggering recomputation. Reducers do a few large sequential reads instead of thousands of small random ones. And Kubernetes gets a proper shuffle service.
Costs: another cluster to run, size and monitor, and an extra network hop for every shuffle byte. It pays for itself from the first spot interruption on.

spark.shuffle.manager=org.apache.spark.shuffle.celeborn.SparkShuffleManager
spark.celeborn.master.endpoints=celeborn-master:9097

Apache Uniffle (remote shuffle)

Apache Uniffle, from Tencent, is the other Apache remote shuffle service with the same architecture (coordinator + shuffle servers) and the same benefits. Pick one; they do not coexist in one application.

spark.shuffle.manager=org.apache.spark.shuffle.RssShuffleManager
spark.rss.coordinator.quorum=rss-coord:19999

Push-based shuffle (Magnet)

LinkedIn's contribution to Spark 3.2: mappers push blocks to the external shuffle service, which merges them per reducer ahead of time. Reducers read one merged file instead of N small ones.

Solves: the small-random-read problem on very large shuffles, without a separate shuffle cluster.
Needs: YARN with the external shuffle service. Not available on Kubernetes or Databricks.

spark.shuffle.push.enabled=true
spark.shuffle.push.server.mergedShuffleFileManagerImpl=org.apache.spark.network.shuffle.RemoteBlockPushResolver

zstd shuffle compression

The shuffle codec defaults to lz4. zstd produces 20–30% smaller shuffle files and spills for slightly more CPU. On a network- or disk-bound shuffle that is a net win.

spark.io.compression.codec=zstd

⚠️ Disabling shuffle compression

"Compression costs CPU." It does: about 1% of the task time with lz4. Turning it off costs 3–5× the network and disk traffic. Shuffle stages are almost never CPU-bound; leave it on.

spark.shuffle.compress=false            # don't
spark.shuffle.spill.compress=false      # really don't

Fast local disks

spark.local.dir is where shuffle files and spills go. On cloud VMs the default is often the boot disk: a network-attached volume with a few thousand IOPS shared by everything. Point it at local NVMe and list several disks so Spark stripes across them.

spark.local.dir=/mnt/nvme0,/mnt/nvme1

ℹ️ On Kubernetes and serverless platforms this is a pod spec or platform setting, not a Spark conf.

Dataproc Enhanced Flexibility Mode

Dataproc-specific: shuffle data for jobs on preemptible secondary workers is written to the primary (on-demand) workers, so a preempted secondary loses nothing. The same idea as a remote shuffle service, built into the platform.

Docs: Dataproc Enhanced Flexibility Mode.

Bigger fetch buffers

spark.reducer.maxSizeInFlight (default 48 MB) bounds how much a reducer requests at once. On high-latency networks, doubling it reduces round trips. Pair with more fetch retries for flaky networks.

spark.reducer.maxSizeInFlight=96m
spark.shuffle.io.maxRetries=10

Choosing

Cluster	For dynamic allocation	For spot nodes
YARN	external shuffle service (+ Magnet for huge shuffles)	Celeborn / Uniffle, or EFM on Dataproc
Standalone	external shuffle service	Celeborn / Uniffle
Kubernetes	shuffle tracking (slow scale-down) or Celeborn / Uniffle	Celeborn / Uniffle + decommissioning
Serverless	built in	built in

And everywhere: zstd on, compression never off, local disks that are actually local.

Layer 4b: Memory, caching and serialization

Executor memory is not one number. It is a heap split by the unified memory manager, plus memory outside the heap that Python workers, Arrow buffers and native engines use, plus the container limit that YARN or Kubernetes enforces. Most memory incidents are a mismatch between those three.

  container limit (what YARN / K8s kills you for exceeding)
  ┌─────────────────────────────────────────────────────────────────────┐
  │ spark.executor.memory (JVM heap)                 │ memoryOverhead   │
  │ ┌──────────────────────────────────┬───────────┐ │ (off-JVM:        │
  │ │ unified region (60% of heap-300MB)│ user +    │ │  Python workers, │
  │ │ ┌────────────────┬─────────────┐ │ reserved  │ │  Arrow buffers,  │
  │ │ │ execution      │ storage     │ │           │ │  native engine,  │
  │ │ │ (shuffle, sort,│ (cache,     │ │           │ │  thread stacks)  │
  │ │ │  join, agg)    │  broadcast) │ │           │ │                  │
  │ │ │  ◄── borrows either way ──►  │ │           │ │ + offHeap.size   │
  │ │ └────────────────┴─────────────┘ │           │ │   if enabled     │
  │ └──────────────────────────────────┴───────────┘ │                  │
  └─────────────────────────────────────────────────────────────────────┘

Right-sized executors (4–5 cores)

The hardware provisioning guide's advice has not changed in a decade: 4–5 cores and 16–32 GB per executor. Enough parallelism per JVM to amortise overhead, small enough that GC pauses stay short and a lost executor loses little work. Above 5 cores per executor, HDFS and S3 client throughput stops scaling anyway.

spark.executor.cores=4
spark.executor.memory=20g
spark.executor.instances=10

⚠️ Fat executors (32 cores, 200 GB)

One executor per node sounds efficient. In practice: multi-second GC pauses on a 200 GB heap, 32 tasks lost when one executor dies, and object-store clients saturating at a fraction of the cores. Spark Connect and native engines soften this a little; they do not remove it.

Memory overhead for Python and native code

Python workers, Arrow buffers and Velox/DataFusion/GPU allocations all live outside the JVM heap. memoryOverhead (default 10% of executor memory, minimum 384 MB) is the only room they have. The classic failure is Container killed by YARN for exceeding memory limits. 20.5 GB of 20 GB physical memory used while the heap graph shows 40% free.

spark.executor.memoryOverhead=4g
spark.executor.pyspark.memory=4g        # caps each Python worker; counts toward the container

Off-heap memory

Lets Tungsten allocate execution memory with sun.misc.Unsafe outside the heap, so sort and hash buffers do not churn the garbage collector. Required by Gluten and Comet. On a plain JVM job it is a modest GC win on large executors.

spark.memory.offHeap.enabled=true
spark.memory.offHeap.size=8g             # counts toward the container limit too

`cache()` a reused DataFrame (`MEMORY_AND_DISK`)

If an intermediate is read more than once, materialise it. cache() uses MEMORY_AND_DISK: blocks that do not fit in storage memory spill to local disk instead of being recomputed.

march = trips.filter(...).cache()
march.count()                            # materialises
march.groupBy("zone_id").count().show()  # reads the cache

Costs: storage memory competes with execution memory; a big cache can push shuffles into spilling. Cache what is reused, not what is big. And remember that cached data is per-application: a nightly batch job caching its only pass gains nothing.

⚠️ `persist(MEMORY_ONLY)` on a huge dataset

2 TB into 160 GB of RAM. MEMORY_ONLY never spills; when storage memory is short it evicts whole partitions and recomputes them from lineage the next time they are read. The job gets slower than with no cache at all, and the Storage tab shows "27% cached" forever. Use MEMORY_AND_DISK, or do not cache.

`MEMORY_ONLY_SER` and Kryo

Serialized cache blocks are 2–5× smaller than deserialized Java objects, at the cost of CPU on every read. With Kryo registered for your classes, smaller still. This matters for RDDs and Datasets of case classes; DataFrames are already stored as compact UnsafeRows, so the gain there is small.

spark.serializer=org.apache.spark.serializer.KryoSerializer
spark.kryo.registrationRequired=false

Arrow for `toPandas()` and `createDataFrame()`

Without Arrow, toPandas() pickles rows one at a time through the driver. With it, the JVM ships columnar batches that pandas ingests directly. Minutes become seconds. There is no reason to leave this off in Spark 3.x.

spark.sql.execution.arrow.pyspark.enabled=true
spark.sql.execution.arrow.maxRecordsPerBatch=10000

`spark.driver.maxResultSize` sized on purpose

A guard rail. The default 1 GB limit makes a careless collect() fail fast with a clear error instead of filling the driver heap and taking the application down. Raise it only as far as the driver can actually hold.

spark.driver.maxResultSize=2g

🧨 `collect()` / `toPandas()` the whole table

"I just want it in pandas." Every row lands in driver memory, after being serialized through it. trips.toPandas() on 3.2 billion rows is a driver OOM with extra steps. Aggregate first, sample first, or write to storage and read the output with pandas.

pdf = trips.toPandas()          # 3.2 billion rows → one JVM. Don't.
pdf = trips.filter(...).groupBy(...).agg(...).toPandas()   # a few thousand rows. Fine.

Bigger driver (16 GB)

The driver holds the file index (16,000 FileStatus objects for trips; millions for a small-files disaster), every broadcast build, every collect() result and the plans themselves. 1 GB is the default; 8–16 GB is normal for a production job touching large tables.

spark.driver.memory=16g
spark.driver.cores=4

`checkpoint()` long lineages

Iterative jobs (graph algorithms, 200-round feature pipelines) build lineages hundreds of stages deep. If a partition is lost, Spark replays the lineage to rebuild it; two hundred stages later you wish it had not. checkpoint() writes the DataFrame to reliable storage and truncates the lineage there. Structured Streaming checkpoints for the same reason, automatically.

spark.sparkContext.setCheckpointDir("s3a://ridehub/chk/")
df = df.checkpoint()

G1GC with region tuning

G1 is the default collector on JDK 11+ and is right for almost everyone. Tune it only when the Spark UI's GC Time column exceeds roughly 10% of task time; the first knob is starting concurrent collection earlier.

spark.executor.extraJavaOptions=-XX:+UseG1GC -XX:InitiatingHeapOccupancyPercent=35

Choosing

Symptom	Lever
`Container killed by YARN for exceeding memory limits`, heap mostly free	`memoryOverhead` (Python / native / Arrow)
Long GC pauses, executors "lost" with no error	smaller executors, off-heap, G1 tuning
Driver OOM	`maxResultSize` guard, no whole-table `collect()`, bigger driver, lower broadcast threshold
Cache shows partial, job slower than uncached	`MEMORY_ONLY` → `MEMORY_AND_DISK`, or stop caching
`toPandas()` takes minutes on small data	Arrow
Iterative job's recovery replays everything	`checkpoint()`

Docs: Tuning Spark: memory management, Hardware provisioning, RDD persistence.

Layer 5: Scheduling and resilience

Layers 1–4 decide how fast the job runs when nothing goes wrong. This layer decides what happens at 03:40 when a spot node gets its two-minute notice, when one task is ten times slower than its siblings, and when S3 returns its fifth 503 of the night.

Dynamic allocation

Add executors when tasks queue, release them when idle. The difference between paying for 50 executors all night and paying for 50 during the join and 4 during the write.

Needs somewhere safe for shuffle files, or releasing an executor would destroy map output that a later stage needs. On YARN or Standalone: the external shuffle service. On Kubernetes: shuffle tracking or a remote shuffle service. On serverless: built in. ℹ️ Spark refuses to start with dynamic allocation enabled and none of these present; the playground models it as an inactive component with an error.

spark.dynamicAllocation.enabled=true
spark.dynamicAllocation.minExecutors=2
spark.dynamicAllocation.maxExecutors=50
spark.dynamicAllocation.executorIdleTimeout=60s

Docs: Dynamic resource allocation.

Speculative execution

When 90% of a stage's tasks are done and a straggler is running 3× longer than the median, launch a copy on another executor and take whichever finishes first.

Solves: slow nodes (a bad disk, a noisy neighbour). It does not solve skew: a copy of the task that holds 30% of the keys is just as slow.
ℹ️ Costs: duplicated side effects. Two copies of a task writing to a non-idempotent sink (a JDBC insert, a foreach that calls an API) means duplicate rows. Safe with file commit protocols and table formats; dangerous with everything else.

spark.speculation=true
spark.speculation.multiplier=3
spark.speculation.quantile=0.9

FAIR scheduler pools

Inside one application, jobs run FIFO by default: the first notebook cell to submit takes every slot until it finishes. FAIR mode shares slots between pools by weight, so a dashboard query in the dashboards pool gets its 20% even while a 30-minute backfill runs in batch.

spark.scheduler.mode=FAIR
spark.scheduler.allocation.file=fairscheduler.xml

spark.sparkContext.setLocalProperty("spark.scheduler.pool", "dashboards")

This is scheduling within an application. Between applications, that is the cluster manager's queue system (YARN queues, Kubernetes namespaces).

Docs: Scheduling within an application.

`spark.locality.wait = 0` on object storage

Spark waits up to 3 seconds per locality level (process, node, rack) for a slot near the data before scheduling a task elsewhere. On HDFS that wait finds a node-local replica and saves a network read. On S3 or GCS there are no local replicas; every byte is a network read; the wait is pure delay. Three seconds times a few locality levels times a few hundred scheduling rounds is a noticeable chunk of a short job.

spark.locality.wait=0s        # object storage only; keep the default on HDFS

Stage-level scheduling

Spark 3.1+ lets a job request a different executor profile for one stage. The feature pipeline runs on cheap 4-core executors until the model-scoring stage, which runs on 8-core, 64 GB executors with a GPU, and then the job goes back to cheap ones.

from pyspark.resource import ResourceProfileBuilder, ExecutorResourceRequests
rp = (ResourceProfileBuilder()
      .require(ExecutorResourceRequests().cores(8).memory("64g").resource("gpu", 1))
      .build())
scored = features.rdd.withResources(rp).mapPartitions(score)

Needs: YARN or Kubernetes with dynamic allocation (the new executors have to come from somewhere). Scoped to the RDD API until recent versions; check your Spark version for DataFrame support.

Docs: Stage level scheduling overview.

Generous task retries

spark.task.maxFailures (default 4) is how many times one task may fail before the stage, and the job, fails. On cloud storage and spot nodes, transient failures are normal: an S3 503 SlowDown, a preempted node mid-task, a shuffle fetch from an executor that just died. Eight attempts with the default back-off costs seconds when it matters and nothing when it does not.

spark.task.maxFailures=8
spark.stage.maxConsecutiveAttempts=8

🧨 `spark.task.maxFailures = 1`

"Fail fast." One throttled S3 request, three hours into the job, and the job is gone. This is not fail-fast; it is fail-often. If the goal is to stop retrying a genuinely broken task, the default 4 already does that in under a minute.

Graceful decommissioning

Spark 3.1+. When a node is about to be taken away (spot interruption notice, autoscaler scale-in, Kubernetes drain), Spark can stop scheduling new tasks on it and migrate its shuffle blocks and cached RDD blocks to other executors before it dies. Without this, a preempted node means recomputing every stage that produced data on it.

spark.decommission.enabled=true
spark.storage.decommission.enabled=true
spark.storage.decommission.shuffleBlocks.enabled=true
spark.storage.decommission.rddBlocks.enabled=true

Works best when there is time (the 2-minute spot notice is enough for most shuffle outputs) and somewhere to put the blocks. With a remote shuffle service there is nothing to migrate, which is even better.

Docs: Decommissioning.

Executor exclusion on failures

Formerly "blacklisting". A node with a failing disk makes every task that lands on it fail; after a threshold, Spark stops scheduling there for a while. Cheap insurance.

spark.excludeOnFailure.enabled=true

Spot / preemptible workers

60–90% cheaper compute that can be reclaimed with two minutes' notice (AWS, GCP) or less (Azure). The single biggest cost lever for batch, and the single biggest source of 04:00 pages when used naked.

⚠️ Costs: losing a node loses its shuffle files and cached blocks. Every stage that produced data there must be recomputed. Three interruptions in a 2-hour job can double its runtime, and the job may never finish on a bad day.
Make it safe with: a remote shuffle service (nothing on the node to lose) or decommissioning (move it before the node dies) or EFM on Dataproc, plus generous retries, plus the driver on an on-demand node. The playground gives the combination spot + Celeborn + dynamic allocation a synergy bonus for exactly this reason.

Event logs and the History Server

The Spark UI disappears when the application ends. Event logging writes every event to storage, and the History Server replays them into the same UI, days later. It is the only way to debug yesterday's failure, and the only way to compare tonight's plan with last week's.

spark.eventLog.enabled=true
spark.eventLog.dir=s3a://ridehub/spark-events/
spark.history.fs.logDirectory=s3a://ridehub/spark-events/

Docs: Monitoring and instrumentation.

Prometheus metrics sink

Executor memory, GC time, task counts and shuffle bytes as time series, with alerts. Spark 3.0+ exposes a Prometheus servlet on the driver and executors natively.

spark.ui.prometheus.enabled=true
spark.metrics.conf.*.sink.prometheusServlet.class=org.apache.spark.metrics.sink.PrometheusServlet

Choosing

Goal	Levers
Pay only for what the stage needs	dynamic allocation + a shuffle home
Survive spot interruptions	remote shuffle or decommissioning, retries ≥ 8, on-demand driver
One slow node should not define the job	speculation (idempotent sinks only), executor exclusion
Dashboards and backfills on one session	FAIR pools
GPU for one stage only	stage-level scheduling
Short jobs on S3 feel sluggish	`locality.wait = 0`
Debug last night	event logs + History Server; Prometheus for trends

Layer 6a: File and table formats

Everything above this line decides how fast Spark can process a byte. This layer decides how many bytes there are to process, and whether Spark can find out which files it does not need to open before opening them.

                  can skip columns?   can skip row groups?   one file = many tasks?   file list from
  Parquet / ORC         yes                 yes                    yes               directory listing
  Avro                  no                  no                     yes               directory listing
  CSV / JSON            no                  no                     yes               directory listing
  gzip CSV / JSON       no                  no                     NO                directory listing
  Delta / Iceberg /     yes                 yes + per-FILE stats   yes               transaction log /
  Hudi (on Parquet)                         (skip whole files)                       manifests

Parquet

Columnar, splittable, compressed per column, with a footer holding min/max statistics per row group. The default output format and the right one for anything that will be read more than once.

Solves: column pruning (read 6 columns of 40 and you read ~15% of the bytes), predicate pushdown via row-group statistics, parallel reads of one file.
Spark's vectorized Parquet reader is the fastest path in the JVM engine; the native engines all have their own.

df.write.parquet("s3a://ridehub/trips/")

Docs: Parquet files.

ORC

The other columnar format, from the Hive world: stripes instead of row groups, built-in indexes and optional bloom filters per column. Functionally equivalent to Parquet for Spark; pick whichever your ecosystem already speaks.

spark.sql.orc.impl=native
spark.sql.orc.filterPushdown=true

Docs: ORC files.

Avro

Row-oriented, schema embedded, splittable. The lingua franca of Kafka and schema registries.

Solves: schema evolution and record-at-a-time producers.
ℹ️ Costs: a SELECT one_column reads every column of every row. Fine as a landing format for a streaming pipeline; convert to Parquet before anything analytical touches it.

Docs: Avro files.

CSV (uncompressed)

Text. Every read parses every byte, types are guesses unless you provide a schema, and quoting rules are a daily surprise.

Splittable (Spark can start a task at any newline), so parallelism is fine. Throughput is not: parsing dominates.
ℹ️ Fine as a landing format. Convert on arrival.

⚠️ gzip-compressed CSV or JSON

gzip has no block index, so a file cannot be split: one file is one task on one core, however large. A 50 GB events.csv.gz is a single task that runs for an hour while 399 cores sit idle. Use bzip2 or seekable zstd if you must compress text, split into many ~128 MB files, or convert to Parquet on landing.

JSON lines

Self-describing and verbose. The Jackson parser is the bottleneck; schema inference is a full extra pass over the data. Same advice as CSV, with more emphasis on always passing a schema.

Docs: JSON files, CSV files.

Delta Lake

Parquet files plus a transaction log (_delta_log/). The log lists exactly which files make up the current version, with per-file min/max statistics.

Solves: ACID writes (no half-written tables after a failed job), time travel (VERSION AS OF), MERGE INTO for upserts and GDPR deletes, data skipping from per-file statistics, OPTIMIZE and ZORDER, and no directory listing: the file list comes from the log, which is why a Delta table with two million files still plans quickly.
Streaming source and sink with exactly-once semantics, which makes it the natural landing table for Structured Streaming.

spark.sql.extensions=io.delta.sql.DeltaSparkSessionExtension
spark.sql.catalog.spark_catalog=org.apache.spark.sql.delta.catalog.DeltaCatalog

df.write.format("delta").partitionBy("dt").save("s3a://ridehub/trips_delta/")

Docs: Delta Lake.

Apache Iceberg

An open table format designed for engine neutrality (Spark, Trino, Flink, Snowflake and BigQuery all read it natively).

Solves: the same ACID/time-travel/upsert problems as Delta, plus hidden partitioning (partition by days(pickup_ts) without a dt column that users have to remember to filter on), manifest-level pruning (the planner skips whole manifests of files from their statistics, without listing or opening anything), and storage-partitioned joins (3.3+).
A REST catalog or Hive metastore holds the table pointer.

spark.sql.extensions=org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions
spark.sql.catalog.lake=org.apache.iceberg.spark.SparkCatalog
spark.sql.catalog.lake.type=rest

df.writeTo("lake.ridehub.trips").partitionedBy(F.days("pickup_ts")).create()

Docs: Iceberg with Spark.

Apache Hudi

The upsert-first table format, from Uber. Copy-on-write tables rewrite files on update; merge-on-read tables append deltas and compact later. Incremental queries ("give me everything that changed since commit X") are a first-class feature.

Solves: CDC-style ingestion with record-level updates and incremental downstream pulls.

df.write.format("hudi").option("hoodie.datasource.write.recordkey.field", "trip_id").save(path)

Docs: Apache Hudi.

Hive metastore table (Parquet)

Not a format, but a different read path: plain Parquet files registered as a partitioned table in a metastore. Partition locations come from the catalog, not from walking S3, and bucketing metadata lives here too.

Solves: listing cost on partitioned tables, and it is the prerequisite for bucketBy.
Costs: MSCK REPAIR TABLE or ALTER TABLE ADD PARTITION after every write that creates a new partition, and a metastore to run.

df.write.partitionBy("dt").saveAsTable("ridehub.trips")
spark.sql("MSCK REPAIR TABLE ridehub.trips")

Choosing

Need	Format
Landing zone from Kafka	Avro (or JSON if you must), converted downstream
Analytical tables, single engine, no updates	Parquet, partitioned by date
Analytical tables with updates, deletes, time travel	Delta (Databricks-centric) or Iceberg (multi-engine)
CDC ingestion with incremental consumers	Hudi
Millions of files and you cannot change the writer	a table format, purely for the manifest-based file list
Anything compressed with gzip	re-compress, or split into many files

Layer 6b: Data layout and the read path

Same bytes, same format, radically different job: this layer is about where rows sit inside and across files so the planner can skip them, and about the handful of read-side settings that decide whether the first stage is a scan or a surprise.

`partitionBy(date)` on write

Hive-style directories, one per value: dt=2026-03-01/part-0000.parquet. A filter on dt never opens the other 334 days.

df.write.partitionBy("dt").parquet(path)

   trips/
   ├── dt=2026-03-01/  ◄─┐
   ├── dt=2026-03-02/    ├── filter dt BETWEEN '2026-03-01' AND '2026-03-31'
   │   …                 │   opens these 31 directories (1,400 files)
   ├── dt=2026-03-31/  ◄─┘
   ├── dt=2026-04-01/      and never lists or opens the other 334 (14,600 files)
   │   …

Solves: the single biggest I/O win available, for the single most common filter (time).
Pairs with pushdown-friendly predicates (Layer 3b); the playground gives the pair a synergy bonus because each is nearly useless without the other.

🧨 `partitionBy(user_id)` and other high-cardinality keys

"Queries filter by user." 40 million users means 40 million directories with one tiny file each. Listing them takes longer than the old full scan did, the driver's file index runs out of memory, and every write creates thousands of files. Partition on low-cardinality columns only (date, region, tenant). For point lookups on high-cardinality keys, use clustering or bucketing below.

`bucketBy(key)` on write

Pre-hash rows into N buckets by the join or aggregation key. Later jobs that join or group by that key skip the shuffle entirely. Metastore tables only.

df.write.bucketBy(256, "driver_id").sortBy("driver_id").saveAsTable("ridehub.trips_b")

`OPTIMIZE … ZORDER BY` (Delta)

Z-ordering rewrites files so that rows close in a multi-dimensional space (say, driver_id × zone_id) land in the same file. Per-file min/max statistics then become selective for either column, and a point lookup opens a handful of files out of thousands.

spark.sql("OPTIMIZE ridehub.trips ZORDER BY (driver_id, zone_id)")

Solves: the high-cardinality lookup that partitioning cannot. Costs: a rewrite of the table; schedule it.

Liquid clustering (Databricks)

Databricks' replacement for partitioning plus Z-ORDER: declare clustering columns once, and OPTIMIZE incrementally clusters new data without rewriting old data or picking partition boundaries up front.

CREATE TABLE trips CLUSTER BY (dt, driver_id) AS SELECT ...

Docs: Liquid clustering.

`sortWithinPartitions` before write

The cheapest clustering there is. Sorting each output partition by a column means each file (and each row group inside it) covers a narrow range of that column, so Parquet's min/max statistics finally skip data on it. The previous article's fare > 300 filter went from "every row group might match" to "3% of row groups might match" with one extra method call.

df.sortWithinPartitions("fare").write.partitionBy("dt").parquet(path)

Compaction to 128 MB–1 GB files

Streaming sinks, hourly jobs and partitionBy(user_id) all produce small files. Every file costs a listing entry, an S3 GET for the footer, a task scheduling decision and a few milliseconds of fixed overhead. Two million 400 KB files is 800 GB of data with the overhead of 2,000,000 opens. A compaction job rewrites them into a few thousand ~256 MB files.

# Delta / Databricks
spark.sql("OPTIMIZE ridehub.trips")
# Iceberg
spark.sql("CALL lake.system.rewrite_data_files(table => 'ridehub.trips')")
# plain Parquet
spark.read.parquet(src).repartition(200).write.mode("overwrite").parquet(dst)

`repartition` / `coalesce` before write

Output file count equals the number of partitions in the final stage, which after a shuffle is spark.sql.shuffle.partitions: 200 files, however small the result. Decide on purpose.

result.coalesce(8).write.parquet(out)                             # fewer files, no shuffle
result.repartition(64, "dt").write.partitionBy("dt").parquet(out) # one well-sized file per dt

coalesce(n) merges partitions without a shuffle but can leave them uneven; repartition(n, col) shuffles and balances. For partitionBy writes, repartition by the partition column avoids every task writing a sliver into every directory.

Delta optimized writes and auto compaction

Databricks' automatic version of the two parts above: an adaptive shuffle before the write to produce ~128 MB files, then a compaction of small files after each commit.

spark.databricks.delta.optimizeWrite.enabled=true
spark.databricks.delta.autoCompact.enabled=true

Explicit schema on read

For CSV and JSON, pass a schema. No inference pass, deterministic types, and no "the column was all nulls on Tuesday so it became a string" incident.

schema = "trip_id string, driver_id long, zone_id int, pickup_ts timestamp, fare double"
spark.read.schema(schema).json("s3a://ridehub/raw/")

⚠️ `inferSchema = True`

For CSV and JSON this is a full extra pass over every byte before the real job starts, and the inferred types can change between runs. Parquet and ORC carry their schema in the footer, so the option does nothing there.

`maxPartitionBytes = 512 MB`

spark.sql.files.maxPartitionBytes (default 128 MB) is the target size of a scan task; openCostInBytes (default 4 MB) is the fixed cost charged per file when bin-packing. Raising the target halves the number of scan tasks, which helps when files are many and the work per byte is small; lowering it helps when a few huge files leave cores idle.

spark.sql.files.maxPartitionBytes=512m
spark.sql.files.openCostInBytes=8m

Cloud-native output committer

Spark's default commit protocol writes to a temporary directory and renames on commit. On HDFS a rename is a metadata operation; on S3 it is a copy and delete of every byte, and it is not atomic. The S3A magic committer (and EMR's S3-optimized committer, and GCS's equivalent) use multipart uploads that are only completed at commit time: no copy, no window of half-visible output.

spark.hadoop.fs.s3a.committer.name=magic
spark.sql.sources.commitProtocolClass=org.apache.spark.internal.io.cloud.PathOutputCommitProtocol
spark.sql.parquet.output.committer.class=org.apache.spark.internal.io.cloud.BindingParquetOutputCommitter

ℹ️ Table formats solve this one level up: Delta, Iceberg and Hudi make a write visible by committing to their log, so the files underneath never need renaming.

Docs: Committing work to S3 with the S3A committers.

`mergeSchema` on read

When Parquet files in one directory have evolved schemas, mergeSchema=true reads every footer and reconciles them. Correct, and on 16,000 files it is 16,000 footer reads on the driver before the job starts. Table formats track schema evolution in their log and avoid this.

🧨 `ignoreCorruptFiles = true`

"The job keeps failing on one bad file." Now it silently skips any file it cannot read, including the one that failed because S3 throttled the request and the one that was half-written by a job that died. Missing rows, no error, and nobody finds out until the finance numbers do not add up. Quarantine bad files explicitly instead.

zstd Parquet compression

Snappy is the default Parquet codec: fast, moderate ratio. zstd gives noticeably smaller files (often 20–40%) at similar read speed on modern CPUs. Less S3 storage, less network, slightly more CPU on write.

spark.sql.parquet.compression.codec=zstd

Reuse one DataFrame per source

spark.read.parquet(path) builds an InMemoryFileIndex by listing the directory tree. Calling it again lists again. Read once, filter many times, and the listing (which on 16,000 S3 objects is several seconds of API calls) is paid once per session.

trips = spark.read.parquet(path)     # lists once
march = trips.filter(...)
april = trips.filter(...)            # no second listing

Parquet column index and bloom filters

Parquet 1.11+ writes page-level min/max (the column index), so pushdown can skip pages inside a row group, not just whole row groups. Bloom filters, enabled per column at write time, answer "is this trip_id definitely not in this row group?" for point lookups on high-cardinality keys where min/max statistics are useless.

spark.hadoop.parquet.bloom.filter.enabled#trip_id=true
spark.hadoop.parquet.bloom.filter.expected.ndv#trip_id=50000000

Choosing

Symptom	Lever
Date filter still scans the whole table	`partitionBy(dt)` + a plain predicate on `dt`
Point lookups on a high-cardinality key	Z-ORDER / liquid clustering / bloom filters, not partitioning
Listing takes minutes	compaction, table format, reuse the DataFrame
200 tiny output files	`coalesce` / `repartition` before write, optimized writes, AQE coalesce
Range filter on a non-partition column reads everything	`sortWithinPartitions` on that column at write time
Job spends minutes before stage 0	`inferSchema` on text, `mergeSchema`, a cold listing: pass a schema, use a table format
Half-written output visible after a failure	cloud committer or a table format

Combinations that are worth more than their parts

A few of these parts only make sense together, and the playground gives them explicit synergy bonuses. They are worth listing because each is a small design pattern:

Combination	Why it is more than the sum
AQE + coalesce + skew join + runtime broadcast	Self-tuning shuffles. The planner's three biggest mistakes (too many reducers, one hot reducer, wrong join strategy) all fixed after the fact.
`partitionBy(dt)` + pushdown-friendly predicates	Partition pruning on every read. Either one alone does nothing.
pandas UDF + Arrow + `memoryOverhead`	Production-grade PySpark. Vectorized logic, fast transfer, and room for the worker to live.
Celeborn + dynamic allocation + spot	Elastic, spot-tolerant shuffle. Stateless executors can be added, removed or reclaimed at any moment.
Spot + graceful decommissioning	Nodes drain instead of crash. The two-minute notice becomes enough time.
Kubernetes + Celeborn + dynamic allocation	Stateless autoscaling on Kubernetes. The missing shuffle service, supplied.
Structured Streaming + RocksDB + watermark + idempotent `foreachBatch`	A streaming pipeline that runs for months. Bounded state, off-heap, exactly-once.
Gluten / Comet + off-heap	A native engine with the memory it needs. Without it, you get the fallbacks and none of the speed.
Iceberg + bucketed joins	Storage-partitioned joins. The big-to-big join with no Exchange.
Thrift server + `cache()` + FAIR pools	A shared BI endpoint that stays responsive. Warm data, fair slots.
Delta + Z-ORDER + compaction	Point lookups on a lakehouse table. Few files, each covering a narrow key range.
Event logs + Prometheus	An observable cluster. Post-mortems and trends.

Six starter profiles

The playground ships these as one-click presets. Each is a defensible starting point for its workload, not a tuned endpoint; the point is to see which layers a given problem actually touches.

🌙 Nightly ETL starter

YARN · DataFrame API · built-in functions · JVM engine · AQE (coalesce, skew) · broadcast hint · pushdown-friendly predicates · external shuffle service · zstd shuffle · 4-core executors · dynamic allocation · History Server · Parquet · partitionBy(dt) · coalesce before write

The RideHub job after Maya fixed it. Nothing exotic: every component is in open-source Spark, and the whole configuration fits in a dozen lines.

spark.master=yarn
spark.submit.deployMode=cluster
spark.sql.adaptive.enabled=true
spark.sql.adaptive.coalescePartitions.enabled=true
spark.sql.adaptive.skewJoin.enabled=true
spark.shuffle.service.enabled=true
spark.io.compression.codec=zstd
spark.executor.cores=4
spark.executor.memory=20g
spark.dynamicAllocation.enabled=true
spark.dynamicAllocation.maxExecutors=50
spark.eventLog.enabled=true

📊 Interactive SQL endpoint

Kubernetes · DataFrame API · Thrift server · JVM · AQE (runtime broadcast) · CBO statistics · Celeborn · cache() the hot week · FAIR pools · locality.wait = 0 · Delta · Z-ORDER · compaction · one DataFrame per source

Dashboards at nine o'clock. The levers are all about latency: a warm long-running session, fair sharing, few well-clustered files, and no three-second locality waits on object storage.

🐍 PySpark ML features

Kubernetes · DataFrame API · pandas UDF · mapInPandas · JVM · AQE (coalesce) · shuffle tracking · memoryOverhead · Arrow · 4-core executors · dynamic allocation · stage-level scheduling · Parquet · partitionBy(dt) · pushdown-friendly predicates

Python at scale without the 40× penalty. Note what is absent: no row-at-a-time UDF, and no native engine (it would fall back on the UDF anyway).

🌊 Streaming Kafka → Delta

Kubernetes · DataFrame API · Structured Streaming · RocksDB · watermark · idempotent foreachBatch · JVM · AQE · 4-core executors · memoryOverhead · generous retries · History Server · Delta · optimized writes

Runs for months. Bounded state off the heap, a sink that tolerates replay, retries that tolerate Kafka hiccups, and a table format that compacts its own small files.

🎟️ Cost-optimised spot batch

Kubernetes · DataFrame API · built-in functions · JVM · AQE (coalesce, skew) · Celeborn · zstd shuffle · 4-core executors · dynamic allocation · spot workers · graceful decommissioning · generous retries · Parquet · partitionBy(dt) · pushdown-friendly predicates · zstd Parquet · coalesce before write

The nightly ETL at 30% of the price. Every resilience lever is on because every node is disposable: shuffle lives off the executors, nodes drain when reclaimed, and eight retries absorb the churn.

🚀 Native engine ETL (Gluten)

YARN · DataFrame API · built-in functions · Gluten + Velox · AQE (coalesce) · broadcast hint · pushdown-friendly predicates · external shuffle service · off-heap memory · 4-core executors · dynamic allocation · Parquet · partitionBy(dt) · compaction

The CPU-bound ETL, roughly twice as fast. Built-in functions only (no fallbacks), off-heap sized for Velox, and compacted inputs so the native scan reads big files.

Cheat sheet: symptom to lever

You see	It usually means	Reach for
`PartitionFilters: []` on a partitioned table	predicate is not on the partition column, or wraps it in a function	rewrite the predicate; DPP for star joins
16,000 scan tasks for one month of data	no pruning, or tiny files	pruning above; compaction; `maxPartitionBytes`
`SortMergeJoin` with two `Exchange`s against a small table	planner sized the dimension from files	broadcast hint; CBO stats; AQE runtime broadcast
`Could not execute broadcast in 300 secs` / driver OOM in a join	broadcast threshold too high, or a hint on a big table	lower the threshold; bigger driver; let it SMJ
199 tasks in 1 minute, 1 task in 3 hours	hot key	AQE skew join; salting for aggregations
`BatchEvalPython` in the plan, 40× slower	row-at-a-time Python UDF	built-in functions; pandas UDF; Arrow UDF
`Container killed by YARN for exceeding memory limits`, heap free	off-JVM memory (Python / Arrow / native)	`memoryOverhead`; `pyspark.memory`; off-heap
200 output files of 2 MB	default shuffle partitions	AQE coalesce; `coalesce()`; optimized writes
One task per `.csv.gz`, hours long	unsplittable compression	re-compress; split; convert to Parquet
Job spends 5 minutes before stage 0	listing, `inferSchema`, `mergeSchema`	schema; table format; reuse DataFrame
`FetchFailedException` after a node vanished	shuffle files died with the node	ESS; remote shuffle; decommissioning
Dynamic allocation never scales down on Kubernetes	shuffle tracking holding executors	Celeborn / Uniffle
Dashboards freeze while a backfill runs	FIFO within one session	FAIR pools
Streaming executor OOM after days	unbounded state on heap	watermark; RocksDB
Duplicate rows after a retry	non-idempotent sink with speculation or restarts	`foreachBatch` keyed by `batchId`; table format
Native engine slower than JVM	fallbacks on every operator	count fallbacks in `explain()`; remove UDFs; size off-heap
Job silently produced fewer rows	`ignoreCorruptFiles`	turn it off; quarantine explicitly

Play it: the Distributed Compute Playground

Reading a parts catalogue is one thing. Bolting the wrong parts together and watching the task count explode is how it sticks.

So I turned this whole article into a game. 🎮

Distributed Compute Playground is a drag-and-drop sandbox where you assemble a Spark job from exactly the nine layers above. Every component in this article (all 104 of them) is a chip. Drop it in and the playground tells you, instantly, what Spark would do with the job you just built:

the explain()-style physical plan, with PartitionFilters, the join strategy and AQEShuffleRead filled in from your choices
the stage DAG with task counts and bytes per stage
how many output files, how many Spark jobs, and roughly how long it takes on your cluster

Drop in month(pickup_ts) = 3 and watch PartitionFilters: [] come back with 16,000 tasks. Swap it for a range on dt and watch it fall to 1,400. Add a broadcast hint and watch the SortMergeJoin turn into a BroadcastHashJoin. Raise the threshold to 2 GB and watch the memory-safety score fall off a cliff.

What's inside

🧱 Build it: 104 real Spark components across 9 colour-coded layers. Prerequisites and conflicts are enforced: dynamic allocation without a shuffle home is an error, Photon on YARN "runs nowhere", a Gluten plan with a Python UDF gets a warning about fallbacks.
🔬 Read the plan it would produce: a live job simulator for 10 workloads, from the nightly ETL to two million tiny files, a hot-key join, a 5 TB-to-5 TB join and a Kafka stream.
📊 Get scored: the eleven dimensions from this article, a grade from S to D, and loud warnings when you do something reckless (yes, spark.task.maxFailures = 1, I'm looking at you 🙃).
🕵️ Crack 25 missions: on-call tickets written like cloud certification exam questions. Each is a ~300-word story about a company in trouble, with a hidden question and goals that check the plan, not just the score. Mission 1 is FIN-2231, the ticket from the previous article — except the company is now Capsule Corp and the engineer is Sakura, because every company and person in the playground borrows a name from the Naruto or Dragon Ball Z universe (no real company ever gets quoted). Then there is the three-hour straggler, the job that only fails on Fridays, the GDPR delete by Friday, and the Photon bill.
🌗 Light and dark themes, zero dependencies, zero network calls. Nothing leaves your browser.

Up and running in 10 seconds

git clone https://github.com/AnikethSD/distributed-computing-playground.git
cd distributed-computing-playground
open index.html        # macOS; use xdg-open on Linux, or just double-click it

No npm install. No Docker. No cluster. Just a browser.

One small favour ⭐

If this article or the game saved you even one evening of staring at a Spark UI, please drop a ⭐ on the repo.

It takes two seconds and costs nothing. It's the single biggest signal that tells me to keep building (the .write article and a second engine family are next 👀), and it helps other engineers find the project.

Got a production incident that would make a brutal mission? A straggler story? A plan that fooled you? Issues and PRs are very welcome. There's a mission authoring guide to get you started, and if databases are more your thing, its sibling, the Database Playground, is waiting too.

The Database Playground: Everything You Can Plug In and Tune in PostgreSQL and MySQL (Including AI Workloads)

Aniketh Deshpande — Tue, 29 Sep 2026 05:50:15 +0000

TL;DR: PostgreSQL and MySQL are not black boxes. Both let you swap storage engines, index types, durability
guarantees, where data physically lives, and even add vector search and ML, often per table and sometimes
per transaction. This article maps those options by the storage or retrieval problem each one solves,
with links to the official docs for every fact.

🤖 Here for AI workloads? Jump straight to Part 3: AI and ML workloads
for vector types, HNSW/IVFFlat/DiskANN tuning, hybrid search, in-database ML, and a full RAG query in SQL.

🎮 Rather play than read? I turned this entire article into a free, open-source browser game.
Drag real PostgreSQL and MySQL components into a database stack and watch it tell you exactly what you just built.
👉 Play the Database Playground on GitHub, and if it clicks, a ⭐ means the world.
(More about it below)

How to read this article
Part 1: PostgreSQL
Part 2: MySQL (plus MariaDB and Percona)
Part 3: AI and ML workloads 🤖
Cheat sheet: problem to engine
Play it: the Database Playground game 🎮
Further reading

How to read this article

Every database, however it's marketed, has the same layers. Each layer has something you can plug in (swap the component) or tune (turn a knob):

            ┌───────────────────────────────┐
 clients →  │ Connection pooler             │  PgBouncer, thread pool
            ├───────────────────────────────┤
            │ Parser / planner / executor   │  hooks, JIT, parallelism, hash joins
            ├───────────────────────────────┤
            │ Access methods                │  ← storage engines + index types
            │  (tables and indexes)         │     heap, columnar, InnoDB, MyRocks, HNSW...
            ├───────────────────────────────┤
            │ Buffer cache                  │  shared_buffers, innodb_buffer_pool_size
            ├───────────────────────────────┤
            │ Write-ahead log / redo log    │  durability vs speed knobs
            ├───────────────────────────────┤
            │ Files, disks, object storage  │  tablespaces, SSD tuning, S3, Iceberg
            └───────────────────────────────┘

The two main parts follow those layers from the top down to the disk. Part 3 then covers AI workloads across both databases.

Conventions

Versions matter. "PG15+" means PostgreSQL 15 or later. MySQL facts use the 8.4 LTS manual unless stated otherwise. The VECTOR type needs MySQL 9.x; 9.7 (April 2026) is the newest LTS.
Restart marks a setting that needs a server restart. Everything else can be changed with a reload, or per session.
To see your current values, run SELECT name, setting, unit, context FROM pg_settings; in PostgreSQL or SHOW VARIABLES LIKE 'innodb%'; in MySQL.

⚠️ Every knob here is a trade-off, not a free speed-up. Change one thing at a time, benchmark with your own
workload, and read the linked doc before touching anything marked unsafe.

Part 1: PostgreSQL

1.1 How PostgreSQL plugs things in

PostgreSQL's superpower is its extension system. A lot of what feels like "core" features ships as extensions that use documented plug points:

Plug point	What it lets an extension do	Since
`CREATE EXTENSION`	Package types, functions, operators and index types	9.1
Table access methods	Replace how rows are stored (the storage engine)	12
Index access methods + `CREATE ACCESS METHOD`	Add new index types (for example, pgvector's HNSW)	9.6
Custom scan providers	Take over query execution (pg_duckdb, Citus and TimescaleDB use this)	9.5
Foreign data wrappers	Query external data as if it were a table	9.1
Background workers	Run jobs inside the server (schedulers, partition maintenance)	9.3
Custom WAL resource managers	Let extension storage be crash-safe and replicated	15
Archive modules (`archive_library`)	Replace shell-based WAL archiving with a loadable library	15
Logical decoding output plugins	Stream row changes out (`pgoutput`, `wal2json`)	9.4

Extensions that hook into the server at startup must be listed in shared_preload_libraries, which needs a restart. List what's installed with \dx, and what's available with SELECT * FROM pg_available_extensions;.

1.2 Storage engines (table access methods)

PostgreSQL has one built-in storage engine, heap. Others plug in through the table access method API:

CREATE TABLE orders (...) USING heap;              -- default
CREATE TABLE events (...) USING columnar;          -- from an extension
ALTER  TABLE events SET ACCESS METHOD heap;        -- PG15+, rewrites the table
SET default_table_access_method = 'columnar';      -- default for new tables

ALTER TABLE ... SET ACCESS METHOD arrived in PG15. It rewrites the whole table under an exclusive lock, so plan it like a migration. The default comes from default_table_access_method.

Engine	What it is	Best for	Watch out for
`heap` (core)	Row store: 8 KB slotted pages, MVCC, TOAST for big values	General OLTP. The safe default	Bloat from dead row versions, which needs VACUUM
Citus columnar (`USING columnar`)	Compressed column store in the `citus_columnar` extension	Append-only analytics, event history	Designed as append-only: `UPDATE`/`DELETE` are unsupported, and several index types aren't available
OrioleDB (`USING orioledb`)	Undo-log MVCC, row-level WAL, no buffer-mapping table	Write-heavy OLTP without VACUUM bloat	Beta, and needs a patched PostgreSQL. The project says it's not for production yet
pg_duckdb	Embeds DuckDB's vectorised engine; mainly a query accelerator	Analytics over heap tables and data-lake files	A different execution engine, so SQL semantics can differ in edge cases
pg_lake (`USING iceberg`)	Transactional Apache Iceberg tables on object storage	Lakehouse tables writable from Postgres	Needs the companion `pgduck_server` process
TimescaleDB hypertables	Automatic time partitioning, plus a row-to-column store ("hypercore")	Time series, metrics, IoT	Columnstore features are under the Timescale License, not Apache 2

💡 In my own tests (see my Database Engines 101 article's lab), the same 5 M rows took 868 MB in heap and 127 MB in
columnar, and an analytics query read 35x fewer pages. Columnar also flatly refused UPDATE.

1.3 Index types (index access methods)

This is where most performance is won or lost. PostgreSQL ships six index types (overview), and extensions add more:

Index	Syntax	Use it for	Key knobs
B-tree (default)	`USING btree`	Equality, ranges, sorting, uniqueness	`fillfactor`, `deduplicate_items`
Hash	`USING hash`	Equality only, on long keys	Crash-safe only since PG10
GIN	`USING gin`	JSONB, arrays, full-text, trigrams	`fastupdate`, `gin_pending_list_limit`
GiST	`USING gist`	Geometry, ranges, nearest-neighbour	Operator-class specific
SP-GiST	`USING spgist`	Quadtrees, k-d trees, radix trees (IPs, phone prefixes)	Operator-class specific
BRIN	`USING brin`	Huge tables naturally ordered by a column (time)	`pages_per_range`, `autosummarize`
bloom (contrib)	`USING bloom`	Equality on any combination of many columns	`length`, `col1..colN`
RUM (ext)	`USING rum`	Full-text search ranked inside the index	Slower to build than GIN
HNSW / IVFFlat (pgvector)	`USING hnsw`	Vector similarity search	See Part 3

B-tree features worth knowing

Deduplication (PG13): duplicate keys are stored once, so indexes on low-cardinality columns shrink a lot. It's on by default. Indexes carried over by pg_upgrade need a REINDEX to benefit. Docs
Covering indexes with INCLUDE (PG11) enable index-only scans without making the extra columns part of the key: CREATE INDEX ON orders (customer_id) INCLUDE (total); Docs
Skip scan (PG18): a multicolumn index on (region, created_at) can now serve WHERE created_at > ... even with no condition on region. PG18 release notes
uuidv7() (PG18): time-ordered UUIDs land at the end of the B-tree instead of at random pages. That means fewer page splits than random gen_random_uuid() values. PG18 release notes

Index features that apply to every type

Partial indexes index only the rows you query: CREATE INDEX ON orders (created_at) WHERE status = 'open'; Docs
Expression indexes index a computed value: CREATE INDEX ON users (lower(email)); Docs
CREATE INDEX CONCURRENTLY builds an index without blocking writes. It can't run inside a transaction, and if it fails it leaves an INVALID index you must drop. Docs
Operator classes change what an index can answer:
- text_pattern_ops makes LIKE 'abc%' indexable under non-C collations. Docs
- jsonb_path_ops gives a smaller, faster GIN index that supports only containment (@>) and jsonpath queries. Docs

GIN pending list. With fastupdate on (the default), new entries go to a pending list that is merged later. This makes writes cheaper, but a reader that triggers the merge pays the cost. For steady read latency, shrink gin_pending_list_limit (default 4 MB) or turn fastupdate off. Docs

BRIN stores just the min and max for each block range (default pages_per_range = 128), so it is kilobytes where a B-tree would be gigabytes. PG14 added minmax_multi operator classes, which tolerate outliers, and bloom operator classes for unordered equality lookups. Docs

CREATE INDEX ON logs USING brin (ts timestamptz_minmax_multi_ops) WITH (pages_per_range = 32);

PG18 can also build GIN indexes in parallel. B-tree (PG11) and BRIN (PG17) builds were already parallel. PG18 release notes

1.4 Row and page layout knobs

These are set per table or per column, and they change how bytes are laid out on disk.

fillfactor and HOT updates

Default 100 for tables. Setting it to 80–90 leaves free space on each page.
An update can then write the new row version on the same page, and if no indexed column changed, no index is touched. That's a heap-only tuple (HOT) update.
ALTER TABLE accounts SET (fillfactor = 85);
HOT docs · storage parameters

TOAST (how big values are stored) (docs)

Values larger than about 2 KB are compressed and/or moved out of line into a TOAST table.
toast_tuple_target (default 2032 bytes) controls when that kicks in.
Per-column storage modes:
- PLAIN: inline only, no compression.
- MAIN: compress, and move out of line only as a last resort.
- EXTERNAL: out of line, no compression. Fast substring() on large text.
- EXTENDED: compress, then move out of line. The default for most types.

ALTER TABLE docs ALTER COLUMN body SET STORAGE EXTERNAL;

Compression algorithm (PG14)

Choose pglz (the default) or lz4 per column, or server-wide with default_toast_compression.
lz4 is much faster to compress and decompress. It requires a server built with lz4 support.
Changing it only affects newly written values.
Docs

ALTER TABLE docs ALTER COLUMN body SET COMPRESSION lz4;

Virtual generated columns (PG18): computed when read, so they take no disk space. Before PG18, generated columns always took space (STORED). PG18 release notes

1.5 WAL and durability knobs

The write-ahead log is what makes commits survive crashes. It's also the biggest lever for write speed. All of these are on the WAL settings page unless linked otherwise.

Knob	Default	What it trades	Safe?
`synchronous_commit`	`on`	`off` returns before the WAL flush. A crash can lose the last few hundred ms of commits, but never corrupts data	✅ Per transaction
`wal_compression`	`off`	Compresses full-page images with `pglz`, `lz4` or `zstd` (PG15+). Costs CPU, saves WAL volume and replication bandwidth	✅
`checkpoint_timeout` / `max_wal_size`	`5min` / `1GB`	Fewer checkpoints mean fewer full-page writes, at the cost of longer crash recovery	✅
`wal_level`	`replica`	`logical` is needed for CDC and logical replication. `minimal` suits standalone bulk loads	✅ Restart
`commit_delay`	`0`	Groups commits together. Helps only at high concurrency on slow fsync	✅
`full_page_writes`	`on`	Protects against torn pages. Turn off only if storage guarantees atomic 8 KB writes	⚠️
`fsync`	`on`	`off` risks unrecoverable corruption on a crash	❌ Throwaway loads only

Per-transaction durability is the most underused trick here:

BEGIN;
SET LOCAL synchronous_commit = off;   -- only this transaction
INSERT INTO page_views ...;           -- losing this on a crash is acceptable
COMMIT;

(Asynchronous commit docs)

Unlogged tables skip WAL entirely. That makes them much faster to write, but they are emptied after a crash and not replicated. They suit staging and cache tables. Switch between the two modes with ALTER TABLE t SET LOGGED | UNLOGGED. Docs

Backups, archiving and replication

archive_library (PG15) archives WAL through a loadable module instead of running a shell command for every file. Docs
Incremental backups (PG17): set summarize_wal = on, then run pg_basebackup --incremental and merge with pg_combinebackup.
Quorum synchronous replication: synchronous_standby_names = 'ANY 1 (replica_a, replica_b)' waits for any one of the replicas, so one slow standby doesn't stall commits. Docs

The official Non-Durable Settings page lists every knob that trades safety for speed, in one place.

1.6 Memory and cache

All on the resource consumption page, except where linked otherwise.

Knob	Default	Guidance
`shared_buffers`	128 MB	Start around 25% of RAM on a dedicated server. The OS page cache holds the rest. Restart
`effective_cache_size`	4 GB	A planner hint only; it allocates nothing. Set to ~50–75% of RAM
`work_mem`	4 MB	Applies per sort or hash operation, per worker, so one query can use it many times over. Raise it per session for analytics
`hash_mem_multiplier`	2.0 (PG15+)	Lets hash joins and aggregates use `work_mem × this` before spilling to disk
`maintenance_work_mem`	64 MB	Used by VACUUM, `CREATE INDEX` and adding foreign keys. 1–2 GB speeds up index builds (it matters a lot for vector indexes)
`huge_pages`	`try`	Use `on` with preallocated Linux huge pages when shared_buffers is ≥ 8 GB. Restart

Keeping hot data hot: pg_prewarm can load a table into cache on demand. Its autoprewarm worker saves the list of cached blocks and reloads it after a restart, so you don't start with a cold cache. pg_buffercache shows what's in the cache right now.

1.7 Disks, SSDs and asynchronous I/O

Tell the planner you're on SSD.

random_page_cost defaults to 4.0, which assumes spinning disks.
On NVMe or cloud SSDs, 1.1–1.5 makes the planner choose index scans when it should.
It can also be set per tablespace (see 1.8).

Asynchronous I/O (PG18). This is one of the biggest storage changes in years.

The new io_method setting lets backends queue several reads at once for sequential scans, bitmap heap scans and VACUUM. Restart.
Options: sync (the old behaviour), worker (background I/O workers) and io_uring (Linux, needs a build with liburing).
The new pg_aios view shows I/O in flight.
PG18 release notes

io_method = io_uring            # or 'worker' (default)
effective_io_concurrency = 256  # PG18 default is 16
io_combine_limit = 256kB        # PG17+: largest combined read

PG18 also raised the defaults of effective_io_concurrency and maintenance_io_concurrency to 16. PG18 release notes

Data checksums are now on by default (PG18 initdb), so silent disk corruption is detected when pages are read. Use --no-data-checksums to opt out. pg_upgrade needs matching settings in the old and new clusters. PG18 release notes · pg_checksums

1.8 Hot and cold data: tablespaces and partitioning

Tablespaces map tables to disks. Put the hot tables on NVMe and the archive on cheap disks, and tell the planner each disk's real cost:

CREATE TABLESPACE fast LOCATION '/nvme/pg'
  WITH (random_page_cost = 1.1, effective_io_concurrency = 256);
CREATE TABLESPACE slow LOCATION '/hdd/pg'
  WITH (random_page_cost = 4.0);

ALTER TABLE orders_2019 SET TABLESPACE slow;   -- locks and copies the table

(CREATE TABLESPACE · managing tablespaces)

Declarative partitioning makes hot and cold data a matter of metadata (docs):

RANGE, LIST and HASH partitions. Partition pruning skips irrelevant partitions both at plan time and at run time.
Each partition can live in its own tablespace.
DETACH PARTITION ... CONCURRENTLY (PG14) removes an old partition without blocking queries. You can then archive it, move it, or drop it.

CREATE TABLE events (id bigint, ts timestamptz, payload jsonb) PARTITION BY RANGE (ts);
CREATE TABLE events_2026_09 PARTITION OF events
  FOR VALUES FROM ('2026-09-01') TO ('2026-10-01') TABLESPACE fast;
CREATE TABLE events_2025 PARTITION OF events
  FOR VALUES FROM ('2025-01-01') TO ('2026-01-01') TABLESPACE slow;

Automating it

pg_partman creates future partitions and applies retention rules: drop old partitions, or detach them into an archive schema. It runs as a background worker or via CALL partman.run_maintenance_proc().
TimescaleDB compresses old chunks into its columnstore automatically, e.g. CALL add_columnstore_policy('metrics', after => INTERVAL '7 days'); (docs). On its managed cloud, it can also tier old chunks to S3. (The company is now called TigerData; the extension is still TimescaleDB.)

1.9 Lakehouse, object storage and external data

This is where PostgreSQL is changing fastest. Postgres is turning into a front door to the data lake.

Tool	What it does
postgres_fdw	Query another Postgres. Pushes down filters, joins, aggregates and sorts. `async_capable` (PG14) scans remote partitions in parallel
file_fdw	Read server-side CSV or text files as tables
mysql_fdw	Read and write MySQL tables from Postgres
pg_duckdb	Read and write Parquet, CSV, JSON, Iceberg and Delta on S3, GCS, Azure and R2, and join them with regular tables. `SET duckdb.force_execution = true;` also runs regular queries on DuckDB's engine
pg_lake	Transactional Iceberg tables (`USING iceberg`) with Postgres as the catalog. Query and `COPY` Parquet, CSV and JSON on S3. Open-sourced by Snowflake from its Crunchy Data acquisition
pg_mooncake	Keeps a columnstore mirror of a Postgres table in Iceberg format, synced via logical replication
Neon	A modified Postgres with separate storage and compute: WAL goes to safekeepers, and pageservers keep pages on object storage. This makes instant branching and scale-to-zero possible. Databricks acquired Neon in 2025

-- pg_duckdb: join a Parquet file on S3 with a local table
SELECT c.name, count(*)
FROM read_parquet('s3://my-bucket/reviews/*.parquet') r
JOIN customers c ON c.id = r['customer_id']
GROUP BY c.name;

💡 Pattern: keep hot data in heap, send cold data to the lake. Keep recent rows in regular partitions.
Periodically copy old partitions to Parquet or Iceberg on object storage (pg_duckdb or pg_lake), then detach
and drop them locally. You can still query all of it.

1.10 Data types that change how data is stored and found

Choosing the right type changes both storage size and which indexes can help.

Type	Why it matters	Index with
`jsonb` vs `json`	`jsonb` is a parsed binary format that's fast to query and indexable. `json` keeps the exact input text and is reparsed every time	GIN (`jsonb_ops` or `jsonb_path_ops`)
Arrays	Many values in one column; containment and overlap queries	GIN
`hstore`	Flat key/value pairs. Older than jsonb, still compact	GIN / GiST
Range and multirange types (multirange PG14)	Time periods and numeric ranges, plus exclusion constraints ("no overlapping bookings")	GiST
PostGIS	Geometry and geography, spatial joins, distance queries	GiST, SP-GiST, BRIN
`pg_trgm`	Makes `LIKE '%foo%'`, `ILIKE`, regex and fuzzy `similarity()` indexable	GIN / GiST (`gin_trgm_ops`)
`tsvector`	Built-in full-text search with stemming and ranking	GIN
`citext`	Case-insensitive text (emails, usernames)	B-tree
`ltree`	Tree paths (`europe.germany.berlin`) for category hierarchies	GiST

-- "no two bookings of the same room may overlap", enforced by the database
CREATE EXTENSION btree_gist;
CREATE TABLE bookings (
  room int,
  during tstzrange,
  EXCLUDE USING gist (room WITH =, during WITH &&)
);

1.11 VACUUM and MVCC knobs

PostgreSQL keeps old row versions for MVCC, and autovacuum cleans them up. Autovacuum that falls behind is the most common cause of table bloat. All settings are on the vacuuming settings page.

Trigger formula: a table is vacuumed when dead rows exceed autovacuum_vacuum_threshold (50) + autovacuum_vacuum_scale_factor (0.2) × table rows. On a 1-billion-row table, that's 200 million dead rows. Lower the scale factor per table:

  ALTER TABLE big_events SET (autovacuum_vacuum_scale_factor = 0.01);

Insert-only tables (PG13+) now get vacuumed too, via autovacuum_vacuum_insert_threshold. That keeps the visibility map current, which index-only scans depend on.
PG18 adds autovacuum_vacuum_max_threshold, a cap on the trigger for huge tables. It also adds autovacuum_worker_slots, so you can raise autovacuum_max_workers without a restart. PG18 release notes
Speed: on SSD, raise autovacuum_vacuum_cost_limit (it effectively defaults to 200) so vacuum keeps up.
Reclaiming space: VACUUM FULL rewrites the table but blocks all access while it runs. pg_repack does the same online, with only brief locks.

1.12 Query execution and connections

Parallel query (docs)

max_parallel_workers_per_gather (default 2) controls how many CPU cores one query can use.
Raise it for analytics. Keep it low for high-concurrency OLTP.

JIT compilation (docs)

On by default, and kicks in above jit_above_cost (100000).
Compiling can take longer than a medium-sized query itself, so many OLTP systems set jit = off.

Statistics (docs)

For badly estimated skewed columns, raise statistics per column: ALTER TABLE t ALTER COLUMN c SET STATISTICS 1000;
CREATE STATISTICS teaches the planner about correlated columns such as city and zip code.

Connection poolers. Every PostgreSQL connection is a separate process, so thousands of clients need a pooler in front:

PgBouncer: pool_mode = transaction is the usual choice.
PgCat: pooling plus load balancing and sharding.
Supavisor: cloud-native and multi-tenant.

1.13 Observability plug-ins

You can't tune what you can't see:

pg_stat_statements: the top queries by total time. The first extension to install.
auto_explain: logs the execution plan of any query slower than N ms.
pg_stat_io (PG16): I/O broken down by backend type and context. Docs
track_io_timing = on: adds I/O time to EXPLAIN (ANALYZE, BUFFERS) and the stats views.

1.14 PostgreSQL starter profiles

These are starting points for a 64 GB RAM, 16 vCPU, NVMe server on PG18. Always benchmark. PGTune and the official tuning wiki are good sanity checks.

# (a) OLTP on NVMe
shared_buffers = 16GB
effective_cache_size = 48GB
work_mem = 16MB
maintenance_work_mem = 2GB
huge_pages = on
wal_compression = lz4
checkpoint_timeout = 15min
max_wal_size = 16GB
random_page_cost = 1.1
effective_io_concurrency = 256
io_method = io_uring              # PG18 + liburing build; else 'worker'
autovacuum_vacuum_cost_limit = 2000
autovacuum_vacuum_scale_factor = 0.05
jit = off
shared_preload_libraries = 'pg_stat_statements,auto_explain,pg_prewarm'

# (b) Bulk load / ETL (trades some durability, never corrupts)
synchronous_commit = off
wal_level = minimal               # no replicas / PITR on this box
max_wal_senders = 0
archive_mode = off
max_wal_size = 64GB
checkpoint_timeout = 30min
maintenance_work_mem = 4GB
max_parallel_maintenance_workers = 8
# + UNLOGGED staging tables + COPY, then ALTER TABLE ... SET LOGGED

# (c) Analytics / reporting
work_mem = 256MB                  # per operation, per worker!
hash_mem_multiplier = 4.0
max_parallel_workers_per_gather = 8
max_parallel_workers = 16
max_worker_processes = 16
random_page_cost = 1.1
effective_io_concurrency = 256
default_statistics_target = 500
enable_partitionwise_join = on
enable_partitionwise_aggregate = on

Part 2: MySQL (plus MariaDB and Percona)

2.1 How MySQL plugs things in

MySQL made "pluggable storage engines" famous. The server layer (parser, optimizer, replication) talks to storage engines through a handler API. On top of that sits a general plugin system, plus the newer component system:

SHOW ENGINES;                                          -- storage engines
SHOW PLUGINS;                                          -- everything plugged in
INSTALL PLUGIN clone SONAME 'mysql_clone.so';          -- classic plugin
INSTALL COMPONENT 'file://component_validate_password'; -- newer component model

(Plugin loading · plugin types · components)

💡 The big difference from PostgreSQL: in MySQL a storage engine is the whole package. It brings its own
index format, locking, crash recovery and caching. PostgreSQL splits table engines and index types into separate
layers that you can mix freely.

2.2 Storage engines

CREATE TABLE t (...) ENGINE=InnoDB;
ALTER TABLE t ENGINE=MyISAM;          -- full table copy

The default comes from default_storage_engine. disabled_storage_engines blocks the ones you never want used.

Engines in stock MySQL (overview)

Engine	What it is	Use it for	Watch out for
InnoDB (default)	ACID, row locks, MVCC, foreign keys, rows clustered by primary key	Almost everything	Primary key choice matters (see 2.3)
MyISAM	No transactions, table-level locks	Legacy read-only data	Not crash-safe
MEMORY	RAM only, hash indexes by default	Temporary lookup tables	Lost on restart. Size capped by `max_heap_table_size`
ARCHIVE	zlib-compressed rows. `INSERT` and `SELECT` only	Cold audit and log data	No `UPDATE`/`DELETE`; almost no indexes
CSV	Plain CSV files on disk	Swapping files with other tools	No indexes
BLACKHOLE	Throws data away but still writes it to the binlog	Replication relays and filters	Stores nothing, by design
FEDERATED	A proxy to a table on another MySQL server	Occasional remote reads	Disabled by default; slow joins
NDB	Distributed, shared-nothing, in-memory	Telecom-style high availability	Ships as the separate NDB Cluster product

Extra engines in MariaDB and Percona Server

Engine	Where	What it's for
MyRocks (RocksDB LSM)	Percona, MariaDB	Write-heavy workloads and data much bigger than RAM. Much better compression and less flash wear than InnoDB
ColumnStore	MariaDB	Columnar analytics (OLAP), with optional object storage
S3	MariaDB 10.5+	Read-only archive tables stored on S3. Move a table there with `ALTER TABLE old_orders ENGINE=S3;`
Spider	MariaDB	Shards one logical table across many servers
CONNECT	MariaDB	Treats CSV, JSON, XML, ODBC and remote tables as MariaDB tables
Aria	MariaDB	Crash-safe replacement for MyISAM; used for internal temp tables

⚠️ MyRocks trade-offs: no foreign keys, no FULLTEXT or SPATIAL indexes, and it requires binlog_format=ROW.
Limitations.
TokuDB, the other write-optimised engine, has been removed from both
Percona and
MariaDB (removed in 10.6).

2.3 InnoDB layout: clustered keys, page size and row formats

The clustered primary key (docs)

InnoDB stores rows inside the primary-key B-tree, and every secondary index stores the primary key as its row pointer. So:
- Keep primary keys short. A 36-byte UUID string gets copied into every secondary index.
- Keep inserts sequential. Random UUIDs split pages all over the tree. UUID_TO_BIN(UUID(), 1) reorders the time bits so new values sort roughly in insert order.
- With no primary key, InnoDB invents a hidden 6-byte one. sql_generate_invisible_primary_key (8.0.30) makes it visible and explicit.

Page size

innodb_page_size: 4 KB to 64 KB, default 16 KB.
It can only be set when the data directory is initialised. Smaller pages suit point lookups on SSD; larger pages suit scans.

Row formats (docs)

DYNAMIC is the default. It stores long BLOB, TEXT and VARCHAR values fully off-page, keeping only a 20-byte pointer in the row.
COMPRESSED adds zlib compression (see 2.8).

Instant DDL

ALTER TABLE t ADD COLUMN c INT, ALGORITHM=INSTANT; changes only metadata. Since 8.0.29 it works for adding a column at any position, and for dropping columns.
Online DDL docs

2.4 Indexes

Index	Syntax	Use it for
B-tree	default	Everything; the clustered primary key plus secondary indexes
Hash	`USING HASH`	MEMORY and NDB only. InnoDB silently uses a B-tree instead
Adaptive hash index	automatic	InnoDB builds an in-memory hash over hot pages. Off by default in 8.4
FULLTEXT	`FULLTEXT(body)`	Natural-language search. Add `WITH PARSER ngram` for Chinese, Japanese and Korean
SPATIAL (R-tree)	`SPATIAL INDEX(g)`	Geo queries. The column needs `NOT NULL` and an `SRID`
Functional (8.0.13)	`INDEX ((LOWER(email)))`	Indexing expressions
Multi-valued (8.0.17)	`INDEX ((CAST(doc->'$.tags' AS CHAR(32) ARRAY)))`	Arrays inside JSON (`MEMBER OF`, `JSON_OVERLAPS`)
Descending (8.0)	`INDEX (a ASC, b DESC)`	Mixed-direction `ORDER BY`
Invisible (8.0)	`ALTER TABLE t ALTER INDEX i INVISIBLE`	Test dropping an index safely. It's still maintained, but the optimizer ignores it

Tools the optimizer uses instead of, or alongside, indexes

Histograms (8.0): ANALYZE TABLE t UPDATE HISTOGRAM ON col WITH 64 BUCKETS; gives the optimizer data distributions without the write cost of an index. 8.4 adds AUTO UPDATE.
Hash join (8.0.18): replaced block nested-loop for joins without usable indexes. Memory is bounded by join_buffer_size.

2.5 Durability: redo log, doublewrite and binlog

Variables are on the InnoDB parameters and binary log options pages.

Knob	Default	What it trades	Safe?
`innodb_flush_log_at_trx_commit`	`1`	`2` writes at commit but flushes about once per second, so an OS crash can lose ~1 s. `0` means a mysqld crash can lose ~1 s	⚠️
`sync_binlog`	`1`	`0` leaves binlog flushing to the OS	⚠️
`innodb_redo_log_capacity` (8.0.30)	100 MB	Bigger means fewer forced flushes but longer recovery. Resizable at runtime	✅
`innodb_doublewrite`	`ON`	Protects against torn pages. `DETECT_ONLY` (8.0.30) detects them but can't repair them	⚠️
`binlog_transaction_compression` (8.0.20)	`OFF`	zstd-compresses the binlog, and it stays compressed on replicas	✅
`binlog_row_image`	`FULL`	`MINIMAL` makes the binlog much smaller, but breaks CDC tools like Debezium	⚠️
`binlog_group_commit_sync_delay`	`0` µs	Waits so more commits share one fsync	✅

The nuclear option for bulk loads: ALTER INSTANCE DISABLE INNODB REDO_LOG (8.0.21) turns off redo logging for the whole instance.

ALTER INSTANCE DISABLE INNODB REDO_LOG;   -- ONLY on a fresh instance you can rebuild
LOAD DATA INFILE '/data/orders.csv' INTO TABLE orders FIELDS TERMINATED BY ',';
ALTER INSTANCE ENABLE INNODB REDO_LOG;

❌ A crash while redo logging is disabled can leave the entire instance unrecoverable. It's meant only for
loading data into a new instance.

2.6 Memory and the buffer pool

innodb_buffer_pool_size: the single most important knob. The default is 128 MB; aim for about 70–75% of RAM on a dedicated server. It can be resized online.
innodb_dedicated_server: sizes the buffer pool and redo log from the machine's RAM and CPU count. It's a good default on dedicated hosts.
Buffer pool warm-up: innodb_buffer_pool_dump_at_shutdown and innodb_buffer_pool_load_at_startup (both on by default) save the hottest pages at shutdown and reload them at startup. That's MySQL's equivalent of pg_prewarm. Docs

MySQL 8.4 changed many InnoDB defaults. If you upgraded from 8.0 and performance shifted, this is why. (What's new in 8.4)

Variable	8.0 default	8.4 default
`innodb_adaptive_hash_index`	ON	OFF
`innodb_change_buffering`	all	none
`innodb_flush_method` (Linux)	fsync	O_DIRECT
`innodb_io_capacity`	200	10000
`innodb_log_buffer_size`	16 MB	64 MB
`innodb_numa_interleave`	OFF	ON

2.7 Disks and SSDs

innodb_flush_method = O_DIRECT (the 8.4 default on Linux) bypasses the OS page cache, so data isn't cached twice.
innodb_io_capacity / innodb_io_capacity_max tell InnoDB how many IOPS it may use for background flushing. Raise them on fast NVMe.
innodb_flush_neighbors = 0 (the default since 8.0) stops InnoDB flushing neighbouring pages together. That trick only helps spinning disks.
innodb_read_io_threads / innodb_write_io_threads: background I/O threads. Restart.
innodb_tmpdir puts the temporary sort files from online ALTER TABLE on a bigger or faster disk.

2.8 Compression

Method	How	Trade-off
Table compression	`ROW_FORMAT=COMPRESSED KEY_BLOCK_SIZE=8`	zlib. The buffer pool holds compressed and uncompressed copies, which costs CPU and RAM
Transparent page compression	`COMPRESSION='lz4'` or `'zlib'`	Compresses each page, then punches holes in the file. Needs a filesystem with hole-punching (ext4, XFS)
ARCHIVE engine	`ENGINE=ARCHIVE`	Heavy compression, but insert and select only
MyRocks	`rocksdb_default_cf_options="compression=kLZ4Compression;bottommost_compression=kZSTD"`	Compression per LSM level. Usually the best ratio of the lot
Binlog	`binlog_transaction_compression=ON`	Smaller binlogs and less replication bandwidth

2.9 Hot and cold data placement

Put a table on a different disk

With innodb_file_per_table (the default), each table is its own .ibd file, and DATA DIRECTORY can place it on another disk.
The directory must be listed in innodb_directories.

CREATE TABLE logs_archive (...) DATA DIRECTORY = '/mnt/hdd/mysql';

General tablespaces group several tables into one file on a chosen disk (docs):

CREATE TABLESPACE ts_cold ADD DATAFILE '/mnt/hdd/ts_cold.ibd' ENGINE=InnoDB;
ALTER TABLE logs_2019 TABLESPACE ts_cold;

Partitioning (docs)

Types: RANGE, LIST, HASH and KEY, plus the COLUMNS variants.
Partition pruning is visible in the partitions column of EXPLAIN.
In 8.0+, only InnoDB and NDB support partitioning.

The classic archiving pattern uses EXCHANGE PARTITION, which swaps a partition with a table instantly:

CREATE TABLE orders_2019 LIKE orders;
ALTER TABLE orders_2019 REMOVE PARTITIONING;
ALTER TABLE orders EXCHANGE PARTITION p2019 WITH TABLE orders_2019;  -- instant swap
ALTER TABLE orders DROP PARTITION p2019;
ALTER TABLE orders_2019 ENGINE=ARCHIVE;     -- or ENGINE=S3 on MariaDB → cold storage on S3

2.10 Lakehouse and analytics bridges

MySQL HeatWave: Oracle's in-memory, columnar, scale-out accelerator that sits next to InnoDB.
- HeatWave Lakehouse queries CSV, Parquet, Avro and JSON files in object storage and joins them with InnoDB tables.
- It's a managed cloud service (Oracle Cloud, AWS and others), not part of MySQL Community.
MariaDB ColumnStore: open-source columnar analytics inside MariaDB.
MariaDB S3 engine: cold, read-only tables on object storage.
CDC to the lake: stream the binlog through Debezium and Kafka into Iceberg, Delta or a warehouse. This needs binlog_format=ROW and binlog_row_image=FULL, and GTIDs are recommended.

2.11 Data types that change storage and retrieval

JSON is stored in an optimised binary format.
- Partial in-place updates (8.0) apply to JSON_SET, JSON_REPLACE and JSON_REMOVE.
- Pair them with binlog_row_value_options=PARTIAL_JSON so only the diff is written to the binlog.
- Index JSON with generated columns, functional indexes or multi-valued indexes.
Spatial types support SRIDs, with geographic calculations since 8.0.
ENUM stores a string as a 1–2 byte integer.
BLOB/TEXT columns are stored off-page and need prefix indexes. They're one more reason to avoid SELECT *.
VECTOR (9.0+) is covered in Part 3.

2.12 Plugins worth knowing

Plugin / component	What it adds
Clone plugin	`CLONE INSTANCE FROM ...` makes a full physical copy for provisioning replicas
Thread pool	Handles thousands of connections efficiently. Enterprise-only in MySQL; free in Percona and MariaDB (`thread_handling=pool-of-threads`)
Group Replication	Multi-node, consensus-based high availability
Query Rewrite	Rewrites bad queries server-side without changing application code
Keyring components	Encryption at rest (`ENCRYPTION='Y'`). The old keyring plugins are gone in 8.4
validate_password	Password policy, now implemented as a component

Online schema changes without downtime:

gh-ost: triggerless; reads the binlog.
pt-online-schema-change: uses triggers.
MySQLTuner gives quick configuration recommendations.

2.13 MySQL starter profiles

# (a) OLTP on NVMe (64 GB RAM, 16 vCPU)
[mysqld]
innodb_buffer_pool_size        = 48G
innodb_redo_log_capacity       = 8G
innodb_flush_log_at_trx_commit = 1
sync_binlog                    = 1
innodb_flush_method            = O_DIRECT
innodb_io_capacity             = 10000
innodb_io_capacity_max         = 20000
innodb_flush_neighbors         = 0
innodb_buffer_pool_dump_at_shutdown = ON
innodb_buffer_pool_load_at_startup  = ON
binlog_format                  = ROW
binlog_transaction_compression = ON

# (b) Bulk load into a disposable instance (revert afterwards!)
[mysqld]
disable-log-bin
innodb_flush_log_at_trx_commit = 2
sync_binlog                    = 0
innodb_doublewrite             = OFF     # restart; unsafe on crash
innodb_redo_log_capacity       = 16G
innodb_log_buffer_size         = 256M
# + ALTER INSTANCE DISABLE INNODB REDO_LOG for the load itself
# + load in primary-key order, add secondary indexes afterwards

# (c) Write-heavy with MyRocks (Percona Server)
[mysqld]
plugin-load-add             = rocksdb=ha_rocksdb.so
default-storage-engine      = ROCKSDB
binlog_format               = ROW
transaction_isolation       = READ-COMMITTED
rocksdb_block_cache_size    = 32G
innodb_buffer_pool_size     = 1G
rocksdb_max_background_jobs = 8
rocksdb_default_cf_options  = "compression=kLZ4Compression;bottommost_compression=kZSTD"

Part 3: AI and ML workloads

🤖🤖🤖 This is the section most people don't know exists. You don't necessarily need a separate vector
database. Both PostgreSQL and MySQL-family databases can store embeddings, run approximate nearest-neighbour
search, combine it with keyword search, and even call models, inside the database, next to the rows you
already have. Here's what to plug in and which knobs matter.

3.1 The five AI workload shapes

Shape	What the database must do	Key components
Store embeddings	Hold 384–3,072-dim float vectors per row, cheaply	Vector types, half precision, TOAST/off-page storage
Similarity search (ANN)	"Top-k closest vectors" in milliseconds	HNSW, IVFFlat, DiskANN, ScaNN indexes
Filtered search	"Closest vectors where tenant = 42"	Partial indexes, partitions, iterative scans, label filters
Hybrid search	Combine semantic and keyword relevance	Full-text/BM25 plus rank fusion
In-database ML	Generate embeddings, predict, train	Cloud ML extensions, PL/Python, HeatWave AutoML

3.2 PostgreSQL + pgvector: the core

pgvector is the standard. It's available on practically every managed Postgres service, and at the time of writing the latest release is 0.8.6 (changelog).

CREATE EXTENSION vector;
CREATE TABLE items (id bigserial PRIMARY KEY, embedding vector(1536));
SELECT * FROM items ORDER BY embedding <=> '[...]' LIMIT 5;   -- cosine distance

Types, and how many dimensions each type can index (README):

Type	Bytes per dim	Max dims in an HNSW/IVFFlat index	Use when
`vector`	4 (float32)	2,000	Default
`halfvec` (0.7.0)	2 (float16)	4,000	Halves storage and index size, usually with little recall loss. Needed to index 3,072-dim models
`bit`	1/8	64,000	Binary quantization for very fast first-pass candidates
`sparsevec` (0.7.0)	per non-zero	1,000 non-zero	Sparse models (SPLADE, BM25-style vectors)

Distance operators:

<-> L2
<#> negative inner product
<=> cosine
<+> L1
<~> Hamming
<%> Jaccard

The operator in your query must match the index's operator class (vector_cosine_ops for <=>, for example), or the index won't be used.

HNSW vs IVFFlat: the two index types, and their knobs

	HNSW	IVFFlat
How	Multi-layer proximity graph	k-means clusters ("lists")
Build	Slower, more memory. Can be built on an empty table	Faster. Build after loading data
Speed/recall	Better	Good if tuned
Build knobs	`m` (default 16), `ef_construction` (default 64)	`lists` (start with rows/1000 up to 1 M rows, then √rows)
Query knob	`hnsw.ef_search` (default 40)	`ivfflat.probes` (default 1; start at √lists)

CREATE INDEX ON items USING hnsw (embedding vector_cosine_ops) WITH (m = 16, ef_construction = 64);

BEGIN;
SET LOCAL hnsw.ef_search = 100;    -- higher = better recall, slower
SELECT id FROM items ORDER BY embedding <=> $1 LIMIT 10;
COMMIT;

Build-time knobs (these matter enormously)

maintenance_work_mem: HNSW builds are much faster when the graph fits in memory. If you see NOTICE: hnsw graph no longer fits into maintenance_work_mem, the build has fallen back to a slow path. Raise it, but not so far that the server runs out of memory. README
max_parallel_maintenance_workers: parallel index builds.
Load with COPY, then create the index. Watch progress in pg_stat_progress_create_index.

SET maintenance_work_mem = '8GB';
SET max_parallel_maintenance_workers = 7;   -- plus the leader process
CREATE INDEX CONCURRENTLY ON items USING hnsw (embedding vector_cosine_ops);

⚠️ Keep pgvector patched. Version 0.8.2 fixed a buffer overflow in parallel HNSW builds, and 0.8.3 fixed
possible index corruption during HNSW vacuuming (changelog).
Check yours with SELECT extversion FROM pg_extension WHERE extname = 'vector';

Filtered search: the #1 production gotcha

Approximate indexes find the top-k first and apply your WHERE afterwards. With a selective filter, you can get back fewer rows than your LIMIT. You have four options:

Iterative index scans (0.8.0+): keep scanning the index until enough rows pass the filter.

   SET hnsw.iterative_scan = relaxed_order;   -- or strict_order; default off
   SET hnsw.max_scan_tuples = 20000;          -- safety cap

Partial indexes per hot filter value: CREATE INDEX ... USING hnsw (embedding vector_l2_ops) WHERE (category_id = 123);
Partition by tenant, so each tenant gets its own small index.
pgvectorscale label filtering (below), which filters inside the graph search.

(README: filtering · iterative scans in 0.8.0)

Shrinking vectors: half precision and binary quantization plus re-ranking

-- index at half precision: half the index size
CREATE INDEX ON items USING hnsw ((embedding::halfvec(1536)) halfvec_cosine_ops);

-- binary quantization: tiny index; re-rank the candidates at full precision
CREATE INDEX ON items USING hnsw ((binary_quantize(embedding)::bit(1536)) bit_hamming_ops);

SELECT * FROM (
  SELECT * FROM items
  ORDER BY binary_quantize(embedding)::bit(1536) <~> binary_quantize($1)
  LIMIT 40                                   -- over-fetch cheap candidates
) candidates
ORDER BY embedding <=> $1                    -- re-rank with full precision
LIMIT 10;

3.3 More PostgreSQL vector engines

Extension	Index	Why you'd pick it	Status
pgvectorscale (TigerData)	`USING diskann` (StreamingDiskANN)	Disk-friendly graph index with Statistical Binary Quantization. Filters on `smallint[]` labels inside the search	Active; uses pgvector's types
VectorChord (TensorChord)	`USING vchordrq`	IVF plus RaBitQ quantization and re-ranking; fast builds at large scale	Active; successor to the deprecated pgvecto.rs
AlloyDB ScaNN (Google)	`USING scann`	Google's ScaNN algorithm inside AlloyDB	Cloud-only (AlloyDB)
pg_diskann (Azure)	`USING diskann`	Microsoft's DiskANN on Azure Postgres	Cloud-only (Azure)

-- pgvectorscale: filter by labels inside the ANN search
CREATE EXTENSION vectorscale CASCADE;
CREATE INDEX ON documents USING diskann (embedding vector_cosine_ops, labels);
SELECT * FROM documents
WHERE labels && ARRAY[1, 3]::smallint[]
ORDER BY embedding <=> $1 LIMIT 10;

3.4 Hybrid search: semantic plus keyword

Pure vector search misses exact matches such as product codes, names and error messages. Hybrid search runs both kinds of search and fuses the rankings:

Built-in full-text search: a tsvector column with a GIN index, queried with websearch_to_tsquery and ranked with ts_rank_cd. Docs
pg_trgm: typo-tolerant fuzzy matching.
ParadeDB pg_search: true BM25 ranking (built on Tantivy) through a bm25 index. Licensed AGPL.
Reciprocal Rank Fusion (RRF): score = Σ 1 / (60 + rank) across the result lists. pgvector's authors publish a reference example.

3.5 End-to-end RAG retrieval in one SQL query (PostgreSQL)

This combines half-precision storage, an HNSW index, a tenant filter kept full by iterative scans, keyword search and RRF:

CREATE EXTENSION IF NOT EXISTS vector;

CREATE TABLE chunks (
  id          bigserial PRIMARY KEY,
  document_id bigint NOT NULL,
  tenant_id   int    NOT NULL,
  content     text   NOT NULL,
  content_tsv tsvector GENERATED ALWAYS AS (to_tsvector('english', content)) STORED,
  embedding   halfvec(1536) NOT NULL          -- 2 bytes per dimension
);

-- after bulk-loading with COPY:
SET maintenance_work_mem = '8GB';
CREATE INDEX ON chunks USING hnsw (embedding halfvec_cosine_ops);
CREATE INDEX ON chunks USING gin (content_tsv);
CREATE INDEX ON chunks (tenant_id);

-- $1 = query embedding, $2 = tenant id, $3 = user question
BEGIN;
SET LOCAL hnsw.ef_search = 100;
SET LOCAL hnsw.iterative_scan = relaxed_order;
WITH semantic AS MATERIALIZED (
  SELECT id, embedding <=> $1::halfvec(1536) AS distance
  FROM chunks WHERE tenant_id = $2
  ORDER BY distance LIMIT 40
),
semantic_ranked AS (
  SELECT id, RANK() OVER (ORDER BY distance) AS rank FROM semantic
),
keyword AS (
  SELECT c.id, RANK() OVER (ORDER BY ts_rank_cd(c.content_tsv, q) DESC) AS rank
  FROM chunks c, websearch_to_tsquery('english', $3) q
  WHERE c.tenant_id = $2 AND c.content_tsv @@ q
  ORDER BY ts_rank_cd(c.content_tsv, q) DESC LIMIT 40
)
SELECT c.id, c.content,
       COALESCE(1.0 / (60 + s.rank), 0) + COALESCE(1.0 / (60 + k.rank), 0) AS rrf_score
FROM semantic_ranked s
FULL OUTER JOIN keyword k ON s.id = k.id
JOIN chunks c ON c.id = COALESCE(s.id, k.id)
ORDER BY rrf_score DESC
LIMIT 8;          -- these 8 chunks go into the LLM prompt
COMMIT;

3.6 In-database ML and model calls (PostgreSQL)

Option	What it does	Status
Google `google_ml_integration`	`embedding('text-embedding-005', text)` and `ml_predict_row()` from SQL	AlloyDB and Cloud SQL
Azure `azure_ai`	`azure_openai.create_embeddings()`, Azure ML and language services	Azure Postgres
AWS `aws_ml`	Calls SageMaker, Comprehend and Bedrock from SQL	Aurora PostgreSQL
PL/Python (`plpython3u`)	Run scikit-learn and similar libraries inside the server	Self-managed only. It's untrusted, so superuser only
pg_duckdb	Fast feature engineering and aggregations over tables and Parquet	Active
Apache AGE	Graph queries (openCypher) inside Postgres, useful for GraphRAG	Active

⚠️ Check a project's status before adopting it. Timescale's pgai (automatic
embedding "vectorizers" and LLM calls from SQL) is now an archived repository.
PostgresML and Apache MADlib were
once the go-to options for in-database ML. Check their recent activity before depending on them. For
auto-embedding today, a common pattern is a trigger that writes to a queue, plus an external worker that calls
the embedding API. Don't call an LLM synchronously inside a trigger.

3.7 MySQL-family AI features

MySQL 9.x: VECTOR type (added in 9.0; up to 16,383 dimensions) (docs)

CREATE TABLE docs (id INT PRIMARY KEY, v VECTOR(768));
INSERT INTO docs VALUES (1, STRING_TO_VECTOR('[0.1, 0.2, ...]'));
SELECT VECTOR_DIM(v), VECTOR_TO_STRING(v) FROM docs;

⚠️ Know what Community MySQL does and doesn't give you. It has the VECTOR type and its conversion
functions. According to the vector functions docs,
the DISTANCE() function is available only in Oracle's HeatWave service (and the Enterprise "MySQL AI"
offering), and Community MySQL has no vector index. For open-source vector search in the MySQL family, look
at MariaDB.

MySQL HeatWave (managed service) (docs)

GenAI: sys.ML_GENERATE (LLM calls), sys.ML_EMBED_ROW (embeddings), sys.ML_RAG (retrieval-augmented generation over a built-in vector store).
AutoML: sys.ML_TRAIN, sys.ML_PREDICT_ROW and sys.ML_EXPLAIN. You can train and serve models on your tables with SQL alone.

CALL sys.ML_TRAIN('shop.churn', 'churned', JSON_OBJECT('task', 'classification'), @model);
CALL sys.ML_MODEL_LOAD(@model, NULL);
SELECT sys.ML_PREDICT_ROW(JSON_OBJECT('tenure', 3, 'plan', 'basic'), @model, NULL);

MariaDB Vector (open source, 11.7+; 11.8 is the first LTS with it) (docs)

Type VECTOR(N), with a VECTOR INDEX based on a modified HNSW ("mHNSW") that works with InnoDB.
Distance functions: VEC_DISTANCE_EUCLIDEAN and VEC_DISTANCE_COSINE.
Knobs: mhnsw_default_m (graph connectivity), mhnsw_ef_search (recall vs speed), mhnsw_default_distance, and mhnsw_max_cache_size (memory for the index).

CREATE TABLE chunks (
  id        BIGINT UNSIGNED AUTO_INCREMENT PRIMARY KEY,
  content   TEXT NOT NULL,
  embedding VECTOR(4) NOT NULL,
  VECTOR INDEX (embedding) M=16 DISTANCE=cosine
);

INSERT INTO chunks (content, embedding) VALUES
  ('Tuning InnoDB buffer pool', VEC_FromText('[0.12,0.45,0.33,0.91]')),
  ('Vector search in MariaDB',  VEC_FromText('[0.10,0.40,0.35,0.88]'));

SET SESSION mhnsw_ef_search = 40;   -- higher = better recall
SELECT id, content
FROM chunks
ORDER BY VEC_DISTANCE_COSINE(embedding, VEC_FromText('[0.11,0.42,0.34,0.90]'))
LIMIT 5;

The distance function in ORDER BY must match the index's DISTANCE setting, or the index isn't used.

Managed MySQL with AI features

Google Cloud SQL for MySQL: vector columns, ANN vector indexes and approx_distance() searches. Docs
Amazon Aurora MySQL: calls SageMaker, Comprehend (for example aws_comprehend_detect_sentiment()) and Bedrock from SQL. Docs
PlanetScale: MySQL-compatible vector columns and indexes. Docs
MindsDB: an external SQL layer that connects to both MySQL and Postgres to add models and knowledge bases.

3.8 AI workload tuning checklist

#	Do this	Why
1	Size `maintenance_work_mem` so the HNSW graph fits, and use parallel builds	Builds that spill are many times slower
2	Keep the vector index in RAM (shared_buffers plus OS cache, or `mhnsw_max_cache_size` in MariaDB)	Graph search is random access; disk reads kill latency
3	Use `halfvec` (or quantization) for storage and indexes	Half the memory, and 3,072-dim models become indexable
4	Over-fetch, then re-rank	Cheap approximate candidates, then exact or cross-encoder precision
5	Plan for filters: partial indexes, partitions, iterative scans or label filtering	Otherwise filtered queries return too few rows
6	Tune recall per query with `SET LOCAL hnsw.ef_search` / `ivfflat.probes`	Trade recall for latency where it matters
7	Measure recall against exact search (`SET enable_indexscan = off`)	Otherwise you're tuning blind
8	Load with `COPY`, build indexes afterwards; generate embeddings asynchronously	Faster loads, and no LLM calls inside transactions
9	Keep embeddings in a separate `chunks` table	The main table stays small, and re-embedding with a new model doesn't rewrite business rows
10	Add hybrid search (FTS or BM25 plus RRF)	Catches exact terms that embeddings miss

Cheat sheet: problem to engine

Problem	PostgreSQL	MySQL / MariaDB
General OLTP	`heap` + B-tree	InnoDB
Write-heavy, larger than RAM	Tune WAL and checkpoints; OrioleDB (beta)	MyRocks
Append-only analytics	Citus `columnar`, TimescaleDB columnstore, pg_duckdb	MariaDB ColumnStore, HeatWave (cloud)
Faster writes, some data loss acceptable	`SET LOCAL synchronous_commit = off`	`innodb_flush_log_at_trx_commit = 2`
Staging or scratch tables	`UNLOGGED` tables	`sql_log_bin=0`, MEMORY/TempTable
One-off bulk load	`wal_level = minimal`, `COPY`, build indexes afterwards	`ALTER INSTANCE DISABLE INNODB REDO_LOG`
Hot data on NVMe, cold on HDD	Tablespaces + partitions	`DATA DIRECTORY`, general tablespaces
Cold data on S3	pg_lake / pg_duckdb (Parquet, Iceberg)	MariaDB S3 engine, HeatWave Lakehouse
Rolling archive	`DETACH PARTITION CONCURRENTLY` + pg_partman	`EXCHANGE PARTITION` → ARCHIVE
Huge time-ordered table	BRIN index	Partition by range
JSON documents	`jsonb` + GIN (`jsonb_path_ops`)	JSON + multi-valued / functional indexes
Fuzzy text search	`pg_trgm`	FULLTEXT (+ ngram parser)
Geo	PostGIS + GiST	SPATIAL (R-tree) + SRID
Vector search	pgvector HNSW / pgvectorscale / VectorChord	MariaDB Vector, HeatWave, Cloud SQL
Hybrid search	FTS or pg_search + RRF	FULLTEXT + vector + RRF in SQL
In-DB ML	Cloud ML extensions, PL/Python	HeatWave AutoML, Aurora ML
Remote data	FDWs (postgres_fdw, mysql_fdw)	FEDERATED, MariaDB CONNECT/Spider
Thousands of connections	PgBouncer	Thread pool (Percona/MariaDB)

Play it: the Database Playground game

Reading about knobs is one thing. Breaking a database on purpose is how it actually sticks.

So I turned this whole article into a game. 🎮

Database Playground is a drag-and-drop sandbox where you assemble a database engine layer by layer, from the connection pooler down to the disk. It then tells you, instantly, what you've built: what it's great at, what it's terrible at, and what kind of data belongs in it.

What's inside

🧱 Build it: drag 100+ real components (56 PostgreSQL, 47 MySQL / MariaDB / Percona) into colour-coded layers: table engines, index types, WAL and redo, memory, storage, vector types.
📊 See it analysed live: capability scores, best-fit workloads, sample data that suits your build, the generated postgresql.conf / my.cnf and SQL, and loud warnings when you do something reckless (yes, fsync = off, I'm looking at you 🙃).
🕵️ Crack 24 missions: story-driven case studies written like cloud certification exam questions. Each is a ~300-word story about a company in trouble, with a hidden question. Work out the real problem, build the fix, and earn up to ⭐⭐⭐.
🤖 Go deep on AI: a RAG bot that can't find error code E-4012, 500 million embeddings on a 256 GB box, clinical RAG on MariaDB, and the classic "ask for 10 results, get 2" filtered vector search trap.
🌗 Light and dark themes, zero dependencies, zero network calls. Your data never leaves your browser.

Up and running in 10 seconds

git clone https://github.com/AnikethSD/database-playground.git
cd database-playground
open index.html        # macOS; use xdg-open on Linux, or just double-click it

No npm install. No Docker. No sign-up. Just a browser.

One small favour ⭐

If this article or the game saved you even one evening of doc-diving, please drop a ⭐ on the repo.

It takes two seconds and costs nothing. It's also the single biggest signal that tells me to keep building (more AI missions are next 👀), and it helps other developers discover the project.

Got a production war story that would make a brutal mission? Spotted a score that feels wrong? Issues and PRs are very welcome. There's a mission authoring guide to get you started.

⭐ Star Database Playground on GitHub →

Database Engines 101: How They Work, the Main Types, and How to Swap Them in PostgreSQL

Aniketh Deshpande — Mon, 28 Sep 2026 12:02:03 +0000

🧪 This article has a hands-on lab. You'll run PostgreSQL in Docker, store the same 5 million rows in
two different storage engines, crash the database on purpose, and measure what changes.
If you already know the theory, skip straight to the lab.

TL;DR: A database engine is the part of a database that turns "save this row" into bytes on disk and back again.
There's no single best engine. Row stores suit transactions, LSM trees suit heavy writes,
column stores suit analytics and in-memory engines suit speed.
PostgreSQL lets you choose per table with CREATE TABLE ... USING <engine>.

Part 1: How an engine works
Part 2: Types of engines
Part 3: The lab
Cheat sheet: pick an engine by use case

Part 1: How an engine works

When you run INSERT INTO users VALUES (1, 'alice', 1000), two very different pieces of software are involved:

   SQL text
      │
      ▼
 ┌──────────────────────────┐
 │  Query layer             │   parse → plan → execute
 │  "what do you want?"     │
 └────────────┬─────────────┘
              │  "store this row", "give me row X"
              ▼
 ┌──────────────────────────┐
 │  Storage engine          │   pages, rows, logs, versions
 │  "how do the bytes live?"│
 └────────────┬─────────────┘
              ▼
          disk / S3

This article is about the bottom box. The easiest way to see why it looks the way it does is to build the obvious version and watch it fail.

The naive engine: one row per line in a text file

with open("users.csv", "w") as f:
    for i in range(200_000):
        f.write(f"{i},user_{i:06d},{i * 7 % 100000}\n")

On my machine it failed three ways almost immediately:

Read cost depends on which row you want. Finding row N means counting N newlines. Row 0 took 0.25 ms and row 199,999 took 17.6 ms, 70x slower, on a file under 5 MB.
A row that grows overwrites its neighbour. The only safe fix is to rewrite everything after it. I changed 40 bytes and had to write 4.6 MB, a 121,667x write amplification.
A crash creates a row that never existed. Cut power halfway through replacing 42,alice,1000 with 42,bob,2000 and you get 42,bobce,1000. It parses fine, so nothing warns you.

Real engines solve these with four ideas.

Idea 1: Pages

Split every file into fixed-size blocks called pages. PostgreSQL uses 8 KB. Finding block 5,000 becomes plain arithmetic (5000 × 8192), so every read costs the same. The page becomes the unit of disk I/O, caching and crash recovery.

Idea 2: Slotted pages

Inside each page, a small slot array at the front points to rows stored at the back. The two regions grow toward each other:

byte 0     ┌────────────────────────────────┐
           │ Page header (24 bytes)         │
           ├────────────────────────────────┤
           │ Slot 1 │ Slot 2 │ Slot 3 │ ... │  ← slots grow DOWN ↓
           ├────────────────────────────────┤
           │          free space            │
           ├────────────────────────────────┤
           │ ... │ Row 3 │ Row 2 │ Row 1    │  ← rows grow UP ↑
byte 8192  └────────────────────────────────┘

A row's address is (page, slot), not a byte offset. So the engine can move rows around inside a page to reclaim space, and every index entry pointing at (5, 3) still works.

Idea 3: Checksums and the write-ahead log

Checksums detect damage: each page stores a checksum of its own bytes, so a half-written page fails verification instead of returning 42,bobce,1000.

The write-ahead log (WAL) repairs damage. Before touching a page, the engine appends a description of the change to a sequential log and flushes it to disk:

1. append "page 5, slot 3: balance 1000 → 2000" to the WAL  → fsync
2. change page 5 in memory
3. write page 5 to disk ... eventually

If the database crashes before step 3, recovery replays the log. You'll watch this happen in the lab.

Idea 4: MVCC

Instead of overwriting a row, write a new version and stamp each version with the transaction that created it (xmin) and the one that replaced it (xmax). Each transaction reads from a snapshot, so a long report keeps seeing the old balance while an update commits alongside it. Readers never block writers. The cost is that old versions pile up and must be cleaned out, which is what PostgreSQL's VACUUM does.

Part 2: Types of engines

Those four ideas describe PostgreSQL's default engine. Other engines make different trade-offs, depending on what they're optimised for.

Row-oriented store (B-tree + pages)

Whole rows are kept together in pages and updated in place, with B-tree indexes to find them. The classic design.

Great at: transactions, point lookups, frequent updates (OLTP).
Weak at: scanning a few columns across billions of rows, because it reads whole rows anyway.
Examples: PostgreSQL heap, MySQL InnoDB, SQLite, Oracle, SQL Server.

Log-structured merge tree (LSM)

Never update in place. Writes go to an in-memory buffer (plus a log for safety). When the buffer fills, it's flushed to disk as an immutable sorted file, and background compaction merges those files over time.

Great at: very heavy write and ingest rates, since every disk write is sequential.
Weak at: reads may check several files (Bloom filters help), and compaction uses background CPU and I/O.
Examples: RocksDB, LevelDB, Cassandra, ScyllaDB, MyRocks, CockroachDB's Pebble.

Column-oriented store

Store each column separately and compress it. A query that needs 2 of 12 columns reads only those 2.

Row store (one page)             Column store
┌──────────────────────────┐     id:      1   2   3   4  ...
│ 1 │ IN │ 12.50 │ chrome  │     country: IN  US  IN  DE ...  ← compresses
│ 2 │ US │  8.00 │ safari  │     amount:  12.5 8.0 3.2 ...      very well
│ 3 │ IN │  3.20 │ chrome  │     browser: ...
└──────────────────────────┘

Great at: analytics such as aggregates, dashboards and scans over huge tables (OLAP). Much smaller on disk.
Weak at: updating or deleting single rows, and fetching whole rows one at a time.
Examples: ClickHouse, DuckDB, Snowflake, BigQuery, Parquet files, Citus/Hydra columnar for PostgreSQL.

In-memory store

Keep everything in RAM. Durability comes from snapshots and append-only logs, or you give it up.

Great at: microsecond latency for caches, sessions, leaderboards and queues.
Weak at: RAM is expensive, and durability is weaker unless you configure it.
Examples: Redis, Memcached, MySQL MEMORY, VoltDB.

Pluggable engines

Some databases let you choose the engine per table. MySQL made this famous:

CREATE TABLE orders (...) ENGINE=InnoDB;
CREATE TABLE cache  (...) ENGINE=MEMORY;

PostgreSQL has had the same thing since version 12, through the Table Access Method API:

CREATE TABLE orders (...) USING heap;       -- the default row store
CREATE TABLE events (...) USING columnar;   -- added by an extension

It goes further than tables:

What you can swap	Syntax	Built in	Added by extensions
Table engine	`CREATE TABLE ... USING x`	`heap`	`columnar` (Citus, Hydra), `orioledb`
Index engine	`CREATE INDEX ... USING x`	`btree`, `hash`, `gin`, `gist`, `spgist`, `brin`	`hnsw`, `ivfflat` (pgvector)
Durability	`CREATE UNLOGGED TABLE`	skip the WAL entirely	—

Now let's plug some in.

Part 3: The lab

You need: Docker, and about 15 minutes. The table load takes about a minute.

We'll use the citusdata/citus image: stock PostgreSQL 17 with the columnar table engine already installed. We're only using its columnar engine, not its distributed features.

Step 0: Start PostgreSQL

docker run -d --name pg-engines -e POSTGRES_PASSWORD=pw citusdata/citus:13.0
docker exec -it pg-engines psql -U postgres

Inside psql, turn on timing:

\timing on

Step 1: See which engines are installed

PostgreSQL lists every engine in the pg_am catalog table ("am" = access method):

SELECT amname AS engine,
       CASE amtype WHEN 't' THEN 'table' ELSE 'index' END AS kind
FROM pg_am ORDER BY kind DESC, amname;

  engine  | kind
----------+-------
 columnar | table     ← added by the extension
 heap     | table     ← PostgreSQL's default
 brin     | index
 btree    | index
 gin      | index
 gist     | index
 hash     | index
 spgist   | index

Step 2: Same table, two engines

A realistic 12-column web-events table. The only difference between the two is the USING clause:

CREATE TABLE events_row (
  id bigint, ts timestamptz, user_id int, country text, amount numeric(10,2),
  device text, browser text, page_url text, referrer text,
  session_id uuid, status smallint, latency_ms int
) USING heap;

CREATE TABLE events_col (LIKE events_row) USING columnar;

Load 5 million rows of fake events into both:

INSERT INTO events_row
SELECT g,
       '2026-01-01'::timestamptz + g * interval '1 second',
       ((g::bigint * 7919) % 100000)::int,
       (ARRAY['IN','US','DE','BR','JP'])[1 + g % 5],
       (g % 10000) / 100.0,
       (ARRAY['mobile','desktop','tablet'])[1 + g % 3],
       (ARRAY['chrome','firefox','safari','edge'])[1 + g % 4],
       '/products/' || (g % 5000) || '/details?ref=campaign_' || (g % 97),
       'https://www.example-' || (g % 300) || '.com/search?q=item' || (g % 1000),
       md5(g::text)::uuid,
       (ARRAY[200,200,200,404,500])[1 + g % 5],
       (g % 900) + 20
FROM generate_series(1, 5000000) g;

INSERT INTO events_col SELECT * FROM events_row;
VACUUM ANALYZE events_row;
ANALYZE events_col;

Now compare their size on disk:

SELECT c.relname AS "table", a.amname AS engine,
       pg_size_pretty(pg_total_relation_size(c.oid)) AS size
FROM pg_class c JOIN pg_am a ON a.oid = c.relam
WHERE c.relname LIKE 'events_%';

   table    |  engine  |  size
------------+----------+--------
 events_col | columnar | 127 MB
 events_row | heap     | 868 MB

Same data, 6.8x smaller. Storing each column together means similar values sit next to each other, and they compress well.

Step 3: Run an analytics query

Average order amount per country. BUFFERS shows how many 8 KB pages each engine had to read:

EXPLAIN (ANALYZE, BUFFERS, COSTS OFF, TIMING OFF)
SELECT country, round(avg(amount), 2) FROM events_row GROUP BY country;

EXPLAIN (ANALYZE, BUFFERS, COSTS OFF, TIMING OFF)
SELECT country, round(avg(amount), 2) FROM events_col GROUP BY country;

Trimmed output:

-- heap
 ->  Parallel Seq Scan on events_row
       Buffers: shared read=111093          ← 868 MB of pages
 Execution Time: 852 ms

-- columnar
 ->  Custom Scan (ColumnarScan) on events_col
       Columnar Projected Columns: country, amount
       Buffers: shared hit=3164 read=16     ← 25 MB of pages
 Execution Time: 1409 ms

The columnar engine read 35x fewer pages. Projected Columns: country, amount shows why: it skipped the other 10 columns entirely.

But look at the times. On my laptop, heap was still faster. The whole dataset fits in RAM, so reading pages is cheap. PostgreSQL also split the heap scan across 3 CPU cores, while this columnar engine scans on one.

💡 That's the real lesson: an engine is a bet on your bottleneck. Columnar wins when I/O is the limit, for
example tables bigger than RAM, network-attached cloud disks, or storage you pay for by the byte scanned.
When everything is already in memory, the difference shrinks, and a well-parallelised row store can win.

Step 4: Update one row

UPDATE events_row SET status = 500 WHERE id = 42;
UPDATE events_col SET status = 500 WHERE id = 42;

UPDATE 1
ERROR:  UPDATE and CTID scans not supported for ColumnarScan

The columnar engine refuses. Compressed column chunks can't be cheaply edited in place, so this engine is append-only. It suits event logs and history tables, not a users table.

Step 5: Turn durability down with `UNLOGGED`

The WAL from Part 1 has a cost. PostgreSQL lets you skip it per table:

CREATE TABLE          clicks_safe (LIKE events_row);
CREATE UNLOGGED TABLE clicks_fast (LIKE events_row);

SELECT pg_current_wal_lsn() AS before \gset
INSERT INTO clicks_safe SELECT * FROM events_row LIMIT 1000000;
SELECT pg_current_wal_lsn() AS middle \gset
INSERT INTO clicks_fast SELECT * FROM events_row LIMIT 1000000;
SELECT pg_current_wal_lsn() AS after \gset

SELECT pg_size_pretty(pg_wal_lsn_diff(:'middle', :'before')) AS wal_written_safe,
       pg_size_pretty(pg_wal_lsn_diff(:'after',  :'middle')) AS wal_written_fast;

 wal_written_safe | wal_written_fast
------------------+------------------
 199 MB           | 0 bytes

The unlogged insert wrote zero bytes of WAL and ran about 1.5x faster (1.5 s vs 2.3 s). Both tables hold 1,000,000 rows.

Step 6: Crash it on purpose

Quit psql (\q), then kill PostgreSQL the hard way. SIGKILL gives it no chance to clean up, like pulling the power cord:

docker kill -s KILL pg-engines
docker start pg-engines
docker logs pg-engines 2>&1 | grep -E "not properly shut down|redo"

LOG:  database system was not properly shut down; automatic recovery in progress
LOG:  redo starts at 0/57A9DD88
LOG:  redo done at 0/57A9DE58

That's the WAL replay from Part 1. Now reconnect and count:

SELECT (SELECT count(*) FROM clicks_safe) AS safe_rows,
       (SELECT count(*) FROM clicks_fast) AS fast_rows;

 safe_rows | fast_rows
-----------+-----------
   1000000 |         0

The unlogged table is empty. With no WAL to replay, PostgreSQL can't know whether its pages are consistent, so after a crash it truncates them. It's a good fit for staging tables and caches you can rebuild. It's a bad fit for anything you can't afford to lose.

Step 7: Swap index engines

Indexes are pluggable too. Put a B-tree and a BRIN index on the same timestamp column:

CREATE INDEX events_ts_btree ON events_row USING btree (ts);
CREATE INDEX events_ts_brin  ON events_row USING brin  (ts);

SELECT indexrelid::regclass AS index,
       pg_size_pretty(pg_relation_size(indexrelid)) AS size
FROM pg_index WHERE indexrelid::regclass::text LIKE 'events_ts_%';

      index      |  size
-----------------+--------
 events_ts_btree | 107 MB
 events_ts_brin  | 40 kB

The BRIN index is about 2,700x smaller. A B-tree stores one entry per row. BRIN ("Block Range INdex") stores only the min and max timestamp for each group of 128 pages. Since events arrive in time order, those ranges barely overlap, so the index can still skip most of the table.

Both indexes answer "give me Feb 1st" correctly (86,400 rows). BRIN is lossy: it returns whole page ranges, and PostgreSQL rechecks the rows, discarding 11,028 extras. If your data isn't physically ordered by that column, BRIN becomes useless and you're back to the B-tree.

Step 8: Change a table's engine later

You don't have to choose the engine forever. Since PostgreSQL 15, one command moves an existing table to a different engine:

SELECT pg_size_pretty(pg_total_relation_size('clicks_safe'));   -- 174 MB

ALTER TABLE clicks_safe SET ACCESS METHOD columnar;
SELECT pg_size_pretty(pg_total_relation_size('clicks_safe'));   -- 25 MB

ALTER TABLE clicks_safe SET ACCESS METHOD heap;                 -- and back again

This is the "plug and play" part. The table name, the columns and your queries stay the same. Only the engine underneath changes. The catch is that it rewrites the whole table and locks it while doing so, so plan it like a migration.

You can also change the default for new tables:

SET default_table_access_method = columnar;

Clean up

docker rm -f pg-engines

Cheat sheet: pick an engine by use case

Use case	Pick	Why
Users, orders, payments	`heap` + `btree`	Frequent updates, point lookups, full durability
Event logs, metrics history, audit trails	`columnar`	Append-only, scanned by a few columns, ~7x smaller
ETL staging, rebuildable caches	`UNLOGGED` heap	No WAL, faster writes. Emptied after a crash
Huge time-ordered tables, range filters	`brin` index	Kilobytes instead of 100+ MB
Full-text search, JSONB, arrays	`gin` index	Indexes the values inside a column
Maps and geometry (PostGIS)	`gist` index	Handles overlaps and nearest-neighbour queries
AI embeddings, similarity search	`hnsw` (pgvector)	Approximate nearest-neighbour search
Massive write-heavy key-value data	LSM engine (RocksDB, Cassandra)	Sequential writes and compaction
Sub-millisecond cache or session store	In-memory (Redis)	RAM speed; durability is optional

Want to go deeper?

I'm learning this by building one: LakePG is a from-scratch Python storage engine that writes byte-for-byte PostgreSQL-compatible pages, with slotted pages, checksums, tuples and MVCC. Its tests compare the pages it writes against a real PostgreSQL 17 server. It's a learning reimplementation of ideas from PostgreSQL, Neon and Databricks Lakebase, not an original design and not production software.

Your Python venv Is (Mostly) a Symlink — Here's Why That Matters

Aniketh Deshpande — Mon, 28 Sep 2026 05:17:36 +0000

TL;DR — On Linux and macOS, python3 -m venv .venv does not create a copy of Python.
The python inside your venv is a symlink that points back to the system interpreter.
The standard library isn't copied either. Only site-packages (your installed packages) really belongs to the venv.
So if the system Python gets upgraded, moved or deleted, your venv can break outright,
or keep running against a different Python without telling you.

The claim
Part 1 — See it with your own eyes (GCP VM / EC2 lab)
Part 2 — What does this mean? Pros, cons, and real-world impact
Part 3 — Building a venv that doesn't betray you
Cheat sheet

The claim

Most of us picture a virtual environment like this:

   What we THINK a venv is
   ───────────────────────

   .venv/
   ├── 🐍 a full private copy of Python
   ├── 📚 a full private copy of the standard library
   └── 📦 my packages

Here's what you actually get:

   What a venv ACTUALLY is (Linux/macOS default)
   ─────────────────────────────────────────────

   .venv/
   ├── bin/
   │   ├── python   ──symlink──►  python3
   │   ├── python3  ──symlink──►  /usr/bin/python3   ◄── the SYSTEM interpreter
   │   ├── pip                    (small script, absolute shebang)
   │   └── activate               (shell script)
   ├── lib/python3.12/site-packages/   ◄── the ONLY truly private part
   └── pyvenv.cfg                      ◄── "home = /usr/bin"  (a pointer back)

   Standard library (os, json, asyncio, ssl, ...)?  → still read from /usr/lib/python3.12

A venv is really an isolated site-packages folder, plus a pointer back to an interpreter that lives somewhere else.
I'll prove it.

Part 1 — See it with your own eyes

You can run this on any Linux box. A throwaway cloud VM is ideal because it's clean and you can wreck it without worrying.

Step 0: Get a VM

Option A — Google Cloud (GCE)

gcloud compute instances create venv-lab \
  --zone=us-central1-a \
  --machine-type=e2-micro \
  --image-family=ubuntu-2404-lts-amd64 \
  --image-project=ubuntu-os-cloud

gcloud compute ssh venv-lab --zone=us-central1-a

Option B — AWS (EC2)

aws ec2 run-instances \
  --image-id resolve:ssm:/aws/service/canonical/ubuntu/server/24.04/stable/current/amd64/hvm/ebs-gp3/ami-id \
  --instance-type t3.micro \
  --key-name <your-key-pair> \
  --security-group-ids <sg-allowing-ssh> \
  --tag-specifications 'ResourceType=instance,Tags=[{Key=Name,Value=venv-lab}]'

ssh -i <your-key>.pem ubuntu@<public-ip>

Once you're in, install the venv module (Ubuntu ships it as a separate package):

sudo apt update && sudo apt install -y python3-venv
python3 --version          # Python 3.12.x on Ubuntu 24.04

💡 The output below comes from Ubuntu 24.04 / Python 3.12. Other distros and versions show the same pattern, with different version numbers.

Experiment 1: Look inside `bin/`

cd ~
python3 -m venv demo
ls -la demo/bin/ | grep python

lrwxrwxrwx 1 ubuntu ubuntu   7 ... python -> python3
lrwxrwxrwx 1 ubuntu ubuntu  16 ... python3 -> /usr/bin/python3
lrwxrwxrwx 1 ubuntu ubuntu   7 ... python3.12 -> python3

The l at the start of each permission string and the -> arrows mean these are symlinks. None of them is a real binary.

Experiment 2: Follow the chain

readlink -f demo/bin/python
readlink -f /usr/bin/python3

/usr/bin/python3.12
/usr/bin/python3.12

They end at the same file. You can also check that the venv entry is just a link and not a hard copy:

stat -c '%i  %s bytes  %n' /usr/bin/python3.12 demo/bin/python3

16123164  6939856 bytes  /usr/bin/python3.12   ← the real binary (~7 MB)
18122480       16 bytes  demo/bin/python3      ← a 16-byte symlink

   Symlink resolution chain
   ────────────────────────

   demo/bin/python
        │  (symlink)
        ▼
   demo/bin/python3
        │  (symlink)
        ▼
   /usr/bin/python3
        │  (symlink, owned by the distro)
        ▼
   /usr/bin/python3.12   ◄── the one and only real interpreter binary

Experiment 3: Read `pyvenv.cfg`, the venv's "birth certificate"

cat demo/pyvenv.cfg

home = /usr/bin
include-system-site-packages = false
version = 3.12.3
executable = /usr/bin/python3.12
command = /usr/bin/python3 -m venv /home/ubuntu/demo

home = /usr/bin is the key line. At startup, Python sees this file and uses it to find its real home (and the standard library that comes with it).

Experiment 4: Ask Python where things come from

demo/bin/python - <<'EOF'
import sys, os, json
print("sys.executable :", sys.executable)
print("realpath       :", os.path.realpath(sys.executable))
print("sys.prefix     :", sys.prefix)        # the venv
print("sys.base_prefix:", sys.base_prefix)   # the REAL Python install
print("json stdlib    :", json.__file__)
EOF

sys.executable : /home/ubuntu/demo/bin/python
realpath       : /usr/bin/python3.12
sys.prefix     : /home/ubuntu/demo
sys.base_prefix: /usr
json stdlib    : /usr/lib/python3.12/json/__init__.py   ← NOT inside the venv!

There are two prefixes. sys.prefix is your venv, where packages get installed. sys.base_prefix is the system install, where the interpreter and the stdlib come from.

Experiment 5: Check the size

du -sh demo
ls demo/lib/python3.12/site-packages

12M   demo
pip  pip-24.0.dist-info

Almost all of those 12 MB are pip. The interpreter and the stdlib (tens of MB) aren't there, because they're borrowed.

Experiment 6: Move the venv and watch it break

head -1 demo/bin/pip
mv demo demo-moved
demo-moved/bin/pip --version

#!/home/ubuntu/demo/bin/python3
bash: demo-moved/bin/pip: /home/ubuntu/demo/bin/python3: bad interpreter: No such file or directory

Every console script (pip, pytest, black, and so on) has the absolute path of the venv baked into its shebang line. Venvs aren't relocatable.

mv demo-moved demo   # put it back

Experiment 7: Delete the base Python (the scary one)

We don't want to wreck the VM's real /usr/bin/python3, since apt and cloud-init depend on it. So we'll build a fake "system Python" in /opt, create venvs from it, then "uninstall" it.

# 1. Build a standalone copy of the interpreter + stdlib at /opt/fakepy
sudo mkdir -p /opt/fakepy/bin /opt/fakepy/lib
sudo cp /usr/bin/python3.12 /opt/fakepy/bin/
sudo cp -r /usr/lib/python3.12 /opt/fakepy/lib/
/opt/fakepy/bin/python3.12 -c "import sys; print(sys.prefix)"   # → /opt/fakepy

# 2. Create two venvs from it: default (symlinks) and --copies
/opt/fakepy/bin/python3.12 -m venv --without-pip ~/linked
/opt/fakepy/bin/python3.12 -m venv --without-pip --copies ~/copied
ls -l ~/linked/bin/python3.12     # → /opt/fakepy/bin/python3.12
ls -l ~/copied/bin/python3.12     # → a real 7 MB file

# 3. Simulate "the base Python was upgraded / uninstalled"
sudo mv /opt/fakepy /opt/fakepy.gone

Now run both:

~/linked/bin/python -c 'print("hello")'

bash: /home/ubuntu/linked/bin/python: No such file or directory

💥 The symlink is dangling, so the venv is dead.

~/copied/bin/python -c 'import sys, os; print(sys.base_prefix, os.__file__)'

/usr /usr/lib/python3.12/os.py

😱 The --copies venv did start, but pyvenv.cfg pointed to a home that no longer exists. Python fell back to its compiled-in prefix (/usr) and silently switched to a different standard library. Here the two happened to be identical, so nothing went wrong. On a real machine they could be different patch or minor versions, and you'd get strange ImportErrors or subtle bugs that nobody can explain.

Clean up:

sudo mv /opt/fakepy.gone /opt/fakepy   # or: sudo rm -rf /opt/fakepy.gone
rm -rf ~/linked ~/copied

And when you're finished with the lab, delete the VM:

gcloud compute instances delete venv-lab --zone=us-central1-a   # GCP
aws ec2 terminate-instances --instance-ids <instance-id>          # AWS

Part 2 — What does this mean?

How a venv actually boots

This comes from PEP 405, the spec that defines venv:

  $ .venv/bin/python app.py
            │
            ▼
  ┌──────────────────────────────────────┐
  │ Kernel follows symlinks              │
  │ .venv/bin/python → /usr/bin/python3.12│  ← executes the SYSTEM binary
  └──────────────────────────────────────┘
            │  (but argv[0] / sys.executable still says .venv/bin/python)
            ▼
  ┌──────────────────────────────────────┐
  │ Python looks for pyvenv.cfg next to  │
  │ (or one level above) sys.executable  │
  └──────────────────────────────────────┘
            │ found!
            ▼
  ┌──────────────────────────────────────┐        ┌──────────────────────────────┐
  │ sys.prefix      = .venv              │ ─────► │ .venv/lib/python3.12/        │
  │                                      │        │        site-packages/  📦     │
  │ sys.base_prefix = derived from       │        └──────────────────────────────┘
  │                   "home = /usr/bin"  │        ┌──────────────────────────────┐
  │                                      │ ─────► │ /usr/lib/python3.12/  📚      │
  └──────────────────────────────────────┘        │ (os, json, ssl, asyncio ...) │
                                                  └──────────────────────────────┘

So a venv is three things:

Piece	Where it lives	Owned by
Interpreter binary	`/usr/bin/python3.12` (symlinked)	🧑‍💼 The OS / package manager
Standard library	`/usr/lib/python3.12/` (referenced)	🧑‍💼 The OS / package manager
Third-party packages	`.venv/lib/python3.12/site-packages/`	🙋 You

⚠️ Being precise: "the venv is a symlink" is a handy shorthand, but the venv directory is real, and so is site-packages.
What's symlinked is the interpreter. The stdlib is borrowed through pyvenv.cfg and never linked or copied.
On Windows, venvs use a small copied python.exe launcher instead of a symlink, but it still reads pyvenv.cfg and depends on the base install in the same way.

Why did Python's designers do it this way?

On purpose. Venvs are meant to be cheap, fast and disposable, not self-contained deployments.

✅ Pros

Pro	Why it matters
Tiny	~12 MB (mostly pip) instead of ~40–100 MB per env. Having 30 projects doesn't mean 30 copies of CPython.
Fast to create	Creating a few symlinks takes milliseconds. Great for CI, tox and nox matrices.
Security patches for free	When `apt upgrade` ships a patched `python3.12` (say, an `ssl` or `zipfile` CVE fix), every venv picks it up with no rebuild.
One interpreter to trust	Only one binary sits on disk to audit, scan and patch.
Package isolation still works	Two projects can pin different versions of `requests` without conflict, which is the reason venvs exist.

❌ Cons

Con	What it looks like
Fragile to base changes	Base Python removed or upgraded → `No such file or directory` / `bad interpreter`.
Silent minor-version drift	Venv made with `python3 -m venv` links to `/usr/bin/python3`, not `python3.12`. If the distro repoints `python3` to 3.13, your venv starts running 3.13 and looks for `lib/python3.13/site-packages`, which doesn't exist. Result: `ModuleNotFoundError` for everything you installed.
Not relocatable	Absolute paths live in shebangs, `pyvenv.cfg` and `activate`. You can't `mv`, `cp -r`, rsync to another path, or bake it in one Docker stage and copy it to another path.
Not portable	You can't copy `.venv` to another machine unless the exact same Python exists at the exact same path.
Compiled wheels are tied to the base	C extensions (`numpy`, `psycopg`, `cryptography`) were built against a specific ABI (`cp312`). Swapping the base breaks them.
`--copies` is only half a fix	It copies the binary but still borrows the stdlib, and falls back silently if `home` disappears (see Experiment 7).

How it bites in real life

                   ┌──────────────────────────────┐
                   │   Your venv  (.venv/)        │
                   │   python3 → /usr/bin/python3 │
                   └───────────────┬──────────────┘
                                   │ depends on
                                   ▼
   ┌───────────────────────────────────────────────────────────────┐
   │            Base interpreter you DON'T control                 │
   └───────────────────────────────────────────────────────────────┘
      ▲               ▲                 ▲                   ▲
      │               │                 │                   │
 do-release-upgrade   brew upgrade   pyenv uninstall    Docker: copy venv
 22.04 → 24.04        python@3.12    3.11.4             into distroless /
 (3.10 → 3.12)        + brew cleanup                    alpine image
      │               │                 │                   │
      ▼               ▼                 ▼                   ▼
 ModuleNotFound   dangling link     dangling link      "no such file"
 (drift)          → venv dead       → venv dead        at container start

Some scenarios you may have hit already:

Ubuntu release upgrade. 22.04 ships Python 3.10 and 24.04 ships 3.12. After do-release-upgrade, venvs built with python3 -m venv start running 3.12 and all your packages are gone. Venvs built with python3.10 -m venv are dead links.
macOS + Homebrew. brew upgrade installs a new patch version, and brew cleanup deletes the old Cellar directory your venv pointed to. This is the classic "my venv broke overnight".
Cron jobs / systemd services on a VM. You set ExecStart=/opt/app/.venv/bin/python ... in a unit file. Months later, unattended upgrades or an OS image refresh change the base Python, and the service dies at 3 a.m.
Docker multi-stage builds. You build /opt/venv in python:3.12 and copy it into gcr.io/distroless/base. /usr/local/bin/python3 doesn't exist there, so the venv's symlink points nowhere.
Renaming a project folder. mv my-app my-app-v2 breaks pip, pytest and every other console script.
CI caching. You cache .venv between runs. The runner image updates its Python, the cache restores a venv linked to a Python that's gone, and the build fails in confusing ways.

Part 3 — Building a venv that doesn't betray you

The root problem isn't the symlink itself. It's that the venv depends on an interpreter you don't control. So the fix is to control the base interpreter, and to treat venvs as disposable.

                    Which venv strategy do I need?
                    ──────────────────────────────

                  ┌──────────────────────────────┐
                  │ Is this for local dev / CI ? │
                  └───────┬───────────────┬──────┘
                       yes│               │no (deploying / shipping)
                          ▼               ▼
         ┌──────────────────────┐   ┌───────────────────────────────┐
         │ Rule 1 + Rule 2      │   │ Must it run where Python may   │
         │ pinned base + lock   │   │ be missing/different?          │
         │ file, recreate often │   └──────┬─────────────────┬──────┘
         └──────────────────────┘       yes│                 │no
                                           ▼                 ▼
                        ┌──────────────────────────┐  ┌─────────────────────┐
                        │ Rule 4: ship the         │  │ Rule 3: Docker with │
                        │ interpreter too          │  │ venv at the SAME    │
                        │ (standalone Python /     │  │ path, SAME base img │
                        │  container / PyInstaller)│  └─────────────────────┘
                        └──────────────────────────┘

Rule 1 — Never build venvs on the distro's `python3`

The system Python exists for the OS (apt, cloud-init, gcloud/aws tooling), and the OS will upgrade it whenever it needs to. Install your own interpreter at a pinned, versioned path that only changes when you change it.

Option A: `uv` (fast and simple)

curl -LsSf https://astral.sh/uv/install.sh | sh
uv python install 3.12            # downloads a standalone CPython build
uv venv --python 3.12 .venv       # venv linked to *uv's* Python, not /usr/bin
uv pip install -r requirements.txt

Option B: `pyenv` (compiles from source)

curl -fsSL https://pyenv.run | bash      # then follow the shell setup it prints
pyenv install 3.12.7
~/.pyenv/versions/3.12.7/bin/python -m venv .venv

Either way, the venv is still a symlink, but now it points at an interpreter that belongs to you:

   BEFORE (fragile)                         AFTER (stable)
   ────────────────                         ──────────────
   .venv/bin/python3                        .venv/bin/python3
        │                                        │
        ▼                                        ▼
   /usr/bin/python3  ◄── apt owns this      ~/.pyenv/versions/3.12.7/bin/python3.12
        │                (changes whenever       ◄── YOU own this
        ▼                 the OS wants)              (changes only when you
   /usr/bin/python3.12                                pyenv uninstall it)

Option C: No extra tools? At least pin the minor version

python3.12 -m venv .venv      # ✅ links to python3.12 — no silent 3.12→3.13 drift
python3    -m venv .venv      # ❌ links to python3   — drifts when the distro repoints it

Check it:

ls -l .venv/bin/ | grep python
# python3.12 -> /usr/bin/python3.12   ✅ pinned

Rule 2 — Treat the venv as a cache, not a pet

Assume any venv can die, and make rebuilding it a one-liner:

# Lock exact versions
pip freeze > requirements.lock          # or: uv pip compile / poetry lock / pip-tools

# Rebuild from scratch any time
rm -rf .venv
python3.12 -m venv .venv
.venv/bin/pip install -r requirements.lock

Also:

Never commit .venv to git (add it to .gitignore).
Never mv a venv. Recreate it at the new path instead.
If you upgraded the base Python in place (same minor version), refresh the links with:

  python3.12 -m venv --upgrade .venv

Rule 3 — In Docker, keep the venv's world identical across stages

Multi-stage builds work fine as long as the final image has the same interpreter at the same path:

# ---------- build stage ----------
FROM python:3.12-slim AS build
RUN python -m venv /opt/venv
ENV PATH="/opt/venv/bin:$PATH"
COPY requirements.lock .
RUN pip install --no-cache-dir -r requirements.lock

# ---------- runtime stage ----------
FROM python:3.12-slim                 # ✅ SAME base → /usr/local/bin/python3.12 exists
COPY --from=build /opt/venv /opt/venv # ✅ SAME path → shebangs still valid
ENV PATH="/opt/venv/bin:$PATH"
COPY . /app
WORKDIR /app
CMD ["python", "main.py"]

   build stage (python:3.12-slim)          runtime stage (python:3.12-slim)
   ┌──────────────────────────────┐        ┌──────────────────────────────┐
   │ /usr/local/bin/python3.12 ◄─┐│        │ /usr/local/bin/python3.12 ◄─┐│
   │ /opt/venv/bin/python ───────┘│ COPY ► │ /opt/venv/bin/python ───────┘│
   │ /opt/venv/lib/.../site-pkgs  │        │ /opt/venv/lib/.../site-pkgs  │
   └──────────────────────────────┘        └──────────────────────────────┘
                 ✅ link target exists on both sides

❌ Don't copy /opt/venv into alpine (musl vs glibc), distroless (no Python at that path), or python:3.13-slim (wrong minor version).

Rule 4 — If Python might be missing at the destination, ship the interpreter

When the target machine may not have a matching Python at all (air-gapped servers, customer machines, minimal images), a venv is the wrong tool. Ship the interpreter along with it:

Approach	What you get
Container image	Interpreter, stdlib and venv frozen together. The most common answer.
python-build-standalone	A relocatable CPython tarball. Unpack it in `/opt/myapp/python`, build the venv from that Python, and ship the whole `/opt/myapp` directory.
`conda-pack`	Packs a full conda env (interpreter included) into a tarball you can relocate.
PyInstaller / Nuitka / shiv	Bundles your app and interpreter into a single executable or zipapp.

What about `--copies`?

python3.12 -m venv --copies .venv

   --symlinks (default on Linux/macOS)      --copies
   ───────────────────────────────────      ─────────────────────────────────
   bin/python3.12 → /usr/bin/python3.12     bin/python3.12   (real 7 MB copy)
   stdlib: borrowed from /usr/lib           stdlib: STILL borrowed from /usr/lib
   base removed → hard fail 💥              base removed → silent fallback 😶
   base patched → auto-upgrade ✅           base patched → stale binary +
                                                           new stdlib ⚠️ mismatch

--copies is useful when your filesystem can't do symlinks (some network mounts, Vagrant/VirtualBox shared folders, certain Windows setups). It is not a way to make a venv self-contained, because the stdlib still lives in the base install. In some ways it's worse: a symlinked venv fails loudly, while a copied one can keep running with a mismatched stdlib.

Bonus: a venv health check

Drop this in your dotfiles or CI to spot broken or drifted venvs before they cause trouble:

#!/usr/bin/env bash
# venv-doctor.sh — usage: ./venv-doctor.sh path/to/.venv [more venvs...]
for v in "$@"; do
  py="$v/bin/python"
  if [[ ! -e "$py" ]]; then
    echo "💀 $v: interpreter link is dangling → $(readlink -m "$py")"
    continue
  fi
  "$py" - "$v" <<'EOF'
import sys, os, re
venv = sys.argv[1]
cfg = open(os.path.join(venv, "pyvenv.cfg")).read()
home = re.search(r"^home\s*=\s*(.+)$", cfg, re.M).group(1).strip()
want = re.search(r"^version(?:_info)?\s*=\s*(\d+\.\d+)", cfg, re.M).group(1)
have = f"{sys.version_info.major}.{sys.version_info.minor}"
problems = []
if want != have:
    problems.append(f"version drift: created with {want}, now running {have}")
if not os.path.isdir(home):
    problems.append(f"home '{home}' no longer exists (stdlib fallback: {sys.base_prefix})")
print(("⚠️  " if problems else "✅ ") + venv + (": " + "; ".join(problems) if problems else f": OK (Python {have} → {os.path.realpath(sys.executable)})"))
EOF
done

$ ./venv-doctor.sh ~/proj-a/.venv ~/proj-b/.venv ~/old-thing/.venv
✅ /home/ubuntu/proj-a/.venv: OK (Python 3.12 → /home/ubuntu/.pyenv/versions/3.12.7/bin/python3.12)
⚠️  /home/ubuntu/proj-b/.venv: version drift: created with 3.10, now running 3.12
💀 /home/ubuntu/old-thing/.venv: interpreter link is dangling → /opt/fakepy/bin/python3.12

Cheat sheet

Question	Command
Is my venv's Python a symlink?	`ls -l .venv/bin/python*`
What does it really run?	`readlink -f .venv/bin/python`
Where does it think home is?	`cat .venv/pyvenv.cfg`
Venv vs base prefix	`.venv/bin/python -c 'import sys;print(sys.prefix, sys.base_prefix)'`
Where does the stdlib come from?	`.venv/bin/python -c 'import os;print(os.__file__)'`
Create a pinned venv	`python3.12 -m venv .venv`
Create a venv on a Python you own	`uv venv --python 3.12` / `~/.pyenv/versions/3.12.7/bin/python -m venv .venv`
Refresh links after base patch upgrade	`python3.12 -m venv --upgrade .venv`
Rebuild from scratch	`rm -rf .venv && python3.12 -m venv .venv && .venv/bin/pip install -r requirements.lock`

Wrapping up

   A venv is a lightweight overlay:

        ┌───────────────────────────────┐
        │  YOUR packages (site-packages)│  ◄── isolated, yours
        ├───────────────────────────────┤
        │  stdlib                       │  ◄── borrowed
        │  interpreter binary           │  ◄── symlinked
        └───────────────────────────────┘
                 base Python install

The symlink design makes venvs cheap, fast and auto-patched. That was a deliberate choice.
It also means a venv is only as stable as the interpreter it points to.
So: own your base interpreter (uv / pyenv / a pinned python3.X), lock your dependencies, treat venvs as disposable, and when you need portability, ship the interpreter, not just the venv.

Next time a venv "randomly" breaks after an upgrade, run ls -l .venv/bin/python first. There's a good chance the link is pointing at a Python that no longer exists.

Found this useful? Drop a ❤️ or tell me in the comments about your worst "my venv broke overnight" story.

Predicate Pushdown - Understanding Practically With An Example

Aniketh Deshpande — Wed, 17 Apr 2024 19:42:28 +0000

What is predicate pushdown?

The immediate theoretical answer that we get on searching is

Predicate pushdown is a query optimisation technique used in database technologies

Okay, I got to know that it is an optimisation technique. But I still did not understand...

How is the optimisation happening? 🤨
What is predicate? 🤔
What exactly is the meaning of pushed down here? 🤷‍♂️

I'm sure since you are also reading this article, even you have these questions in mind!

Now, lets explore this interesting topic practically in PySpark using explain()
(Similar phenomenon could be observed in relational databases as well)

1] Reading a csv file containing employee information.

emp = spark.read.format('csv')\
                .option('header', True)\
                .load('data/employee.csv')
emp.show()

emp.explain(mode="formatted")

Here, step (1) is related to read csv file

2] Lets do a group by and get number of employees on each department

df = emp.groupBy('deptID').count()
df.show()

df.explain(mode="formatted")

Here, step (2) is related to group by

3] Lets filter data for only dept number 10

dept_10 = df.filter(df.deptID == 10)
dept_10.show()

dept_10.explain(mode="formatted")

Here, we can see that step (2) is filtering and step (3) is grouping.

Now here is the catch,
Ideally if we go by the sequence of operations, grouping should be done first and then filtering.

However, the optimiser does filtering first and then grouping, because grouping is an expensive operation and it is optimal to filter first and then group data.

In the physical plan we see that filtering (predicate) is pushed down with respect to grouping. That is why it is called push down !

Write Through

Aniketh Deshpande — Fri, 22 Mar 2024 13:06:39 +0000

Write through cache is a simple to implement caching mechanism.

Here the newly arrived data is written into cache and as well as persisted into the disk or a database. Atomicity is maintained.

There are two ways to implement it:
1] The application writes data to cache and database simultaneously

2] The application writes data to cache and then the cache writes the data into the database.

Advantages:

Simple to implement.
Faster response times.
Data integrity because of atomic nature of write operation.
Lower latency for subsequent reads.

Disadvantages:

Cache pollution: Since every time the data is filled into cache, it can get filled with less frequently read data and more cache eviction which could introduce some latency.
Not suitable for write intensive scenarios as the write operations are slower compared to other methods because data needs to be written in cache as well as persistent storage everytime.

Database Caching Strategies

Aniketh Deshpande — Fri, 22 Mar 2024 12:43:08 +0000

We often face high latencies while fetching data from Database and are unable to meet SLA. Caching is one of the solutions to implement after DB query and table optimisations.

A Cache is used to store the data so that it can be delivered faster to the client as compared to persistent systems like database or disk.

Popular caching tools are:

There are multiple ways in which data can be written into and read from the cache. Let us explore the most prominently used methods or policies.

Cache Aside

Cache Aside or Lazy Loading is one of the cache write policies or strategies.

In cache aside method, the application is responsible for storing data into cache.
When the client sends request to the application, it looks for data in the cache.
If the data is found in the cache it is called as cache-hit. The data is fetched from cache and returned to client.
However, if the data is not found in the cache, also called as cache-miss, the application queries for data in the database, writes the data into cache and sends the response to the client.
Since we store data in cache only when it is necessary, this strategy is also called as Lazy-Loading.

Advantages:

Cost effective because, only the frequently accessed data is stored into cache.

Disadvantages:

The response time can be slow when there is cache miss, because, it involves many i/o operations to fetch data from DB and store it in cache.

In case the system can tolerate an initial delay and the same data is to be fetched repeatedly, then this mechanism works best.

Write through

Write through cache is a simple to implement caching mechanism.

Here the newly arrived data is written into cache and as well as persisted into the disk or a database. Atomicity is maintained.

There are two ways to implement it:
1] The application writes data to cache and database simultaneously

2] The application writes data to cache and then the cache writes the data into the database.

Advantages:

Faster response times.
Data integrity because of atomic nature of write operation.
Lower latency for subsequent reads.

Disadvantages:

Cache pollution: Since every time the data is filled into cache, it can get filled with less frequently read data and more cache eviction which could introduce some latency.
Not suitable for write intensive scenarios as the write operations are slower compared to other methods because data needs to be written in cache as well as persistent storage everytime.

In case the system cannot tolerate an initial delay and the writes are infrequent, then this mechanism works best.

Also if the number of records to be cached is also fixed, then this approach can provide the best of the results.

Thank you :)

Cache Aside

Aniketh Deshpande — Fri, 22 Mar 2024 12:41:36 +0000

Cache Aside or Lazy Loading is one of the cache write policies or strategies.

In cache aside method, the application is responsible for storing data into cache.
When the client sends request to the application, it looks for data in the cache.
If the data is found in the cache it is called as cache-hit. The data is fetched from cache and returned to client.
However, if the data is not found in the cache, also called as cache-miss, the application queries for data in the database, writes the data into cache and sends the response to the client.
Since we store data in cache only when it is necessary, this strategy is also called as Lazy-Loading.

Advantages:

The implementation is simple.

Disadvantages:

The response time can be slow when there is cache miss, because, it involves many io operations to fetch data from DB and store it in cache.

Shallow Copy Vs Deep Copy

Aniketh Deshpande — Sat, 29 Oct 2022 12:57:16 +0000

In our day to day development tasks, we come across the need to copy objects and perform various operations.

Python provides two important functions in the copy library - copy and deepcopy.

Let us understand the difference between the two and their respective use cases.

Shallow Copy - copy.copy(obj)

It makes a copy of the obj at the surface level.
It copies all the contents of the obj.

However, the point to be noted here is that, in case the contents of x are mutable, then, y has reference to the contents of x. Any modification to the contents of x would be reflected in y as well.

In this example we see that modifying the contents of x[1], also modified the contents of y[1].

Similarly, we see that modifying the contents of y, also modified the contents of x.

Deep Copy - copy.deepcopy(obj)

Deepcopy copies the object recursively. Recursively means, it copies the contents of the object and not merely its reference.
So we should use deep copy in case we need an independent copy of the contents of the obj.

In deepcopy, any changes to the contents of x do not effect y, unlike shallow copy.

Copying objects properly is a very important basic python concepts. This knowledge helps in building error free code wrt copying objects.

Thank you
Aniketh Deshpande

Dead Letter Queue

Aniketh Deshpande — Sun, 23 Oct 2022 11:50:55 +0000

This is going to be a very short introduction to dead letter queues.

What is a dead letter queue??

Dead letter queues are messages queues specifically deployed to holding messages that could not be delivered to their intended queues.
Messages sometimes fail to get delivered to their intended queues as they might be unavailable or the queue is full.
Popular message queue tools that support or do not support DLQ:
- AWS SQS (Simple Queue Service) supports DLQ.
- RabbitMQ (free and open source) also supports DLQ.
- Redis does not support DLQ.

Thanks for reading :)
DLQ is a very important component of a scalable and resilient software architecture. This article only provides an introduction to the concept and helps readers with useful links.

Following are some of the useful resources that give more indepth information.

I request the readers of this post to kindly add more helpful URLs in the comment section, or add your experiences using a DLQ in a real world project. I believe that would help beginners get even better idea about how to use this in their projects.

Thank You
Aniketh Deshpande

Change Data Capture - PostgreSQL

Aniketh Deshpande — Sat, 22 Oct 2022 14:58:18 +0000

Change Data Capture is the concept of recording the changes in the database table fields.

It is very helpful in use cases where we want to track creation, updation, deletion of records in the table.
We might want to use this information to make changes in other databases or notify customers or notify other services.
Example:
1. Save a copy of this data in a warehouse post transform.
2. Trigger notification service to notify users about this change.
3. Cache the data.

CDC In Postgres

Using Notify/Listen
- NOTIFY provides a mechanism for interprocess communication between the database and the service that is LISTENing to this notification channel.
- One or more services could be listening to this notification channel.
- Name of this channel is usually the name of the database. However the user is free to set suitable names for these channels.
- Any change in the table is captured by the DB and a trigger is initiated, which calls a function that formats the message to notify.
- This usually contains the table name and the payload string.
- The listening server registers to the channel and gets the message from the DB.
- The service can then use this message and perform operations on it.

Pros:

Simple to implement. Use a trigger and a function to notify. Implement a listen service.

Cons:

Weak reliability. There is always a risk of loss of message especially when the listening service is down. Messages in the queue do not persist.

Using Debezium
- Debezium is an open source tool used for capturing changes in the database tables based on the WAL (write ahead log).
- The tool provides connectors to connect to a variety of databases.
- The source connector is used to capture changes in the source database.
- The sync connector is used to sync data directly in the destination database.

Pros:

The changes are persistent as they can be streamed to kafka. Hence highly reliable.

Cons:

Debezium does not take into account changes in the schema, users need to update the schema changes. Otherwise there would be data loss.

Thank you
Aniketh Deshpande

DEV Community: Aniketh Deshpande

What Actually Happens When You Call spark.read? One Line of Python, a Thousand Tasks

Table of Contents

Prologue: Maya's first ticket

The setup: one cluster, one dataset

Chapter 1: The line that does (almost) nothing

1.1 Resolving the data source

1.2 Listing the files, which is Job 0

1.3 Reading one footer for the schema

1.4 What trips actually is

1.5 Why laziness is the whole design, not a trick

Chapter 2: The first action, or the anatomy of a job

2.1 Stop 1: four plans (Catalyst)

2.2 Stop 2: how many tasks? (file splitting)

2.3 Stop 3: stages and the DAGScheduler

2.4 Stop 4: shipping tasks to executors

2.5 Stop 5: inside one task

2.6 Stop 6: the second stage and the result

2.7 Stop 7: what the Spark UI shows

Chapter 3: Reading less: filters, pruning and pushdown

3.1 Partition pruning: never open the file

3.2 Column pruning: open the file, skip most of it

3.3 Predicate pushdown: let Parquet skip row groups and pages

3.4 Why there's still a Filter operator

3.5 The Spark UI after pruning

3.6 The five ways to lose pruning by accident

Chapter 4: Putting things in order: sort and shuffle

4.1 What "sorted" means on a cluster

4.2 Job 3: the sampling pass nobody asked for

4.3 Stage 3: the shuffle write (map side)

4.4 Stage 4: fetch, sort, write (reduce side)

4.5 Where did the 200 go? (AQE coalescing)

4.6 The twist: the top 10 took 10 seconds

4.7 Three sorts, three costs

Chapter 5: Meeting other tables: joins

5.1 The zones join: broadcast hash join

5.2 The drivers join: sort-merge join

5.3 AQE changes its mind at runtime

5.4 What if drivers had been 5 GB? (skew)

5.5 Dynamic partition pruning: the join that prunes the scan

5.6 How Spark picks a join

Chapter 6: The whole pipeline, end to end

6.1 The final plan, annotated

6.2 The stage map

6.3 A day in the life of task 812

6.4 Before and after

6.5 Epilogue: Thursday

Cheat sheets

Does it shuffle?

The settings that appeared in this story

Five questions to ask any plan

Reading the Spark UI like a story

Try it on your laptop

Further reading

The Distributed Compute Playground: Everything You Can Plug In and Tune in an Apache Spark Job

Table of Contents

How to read this article

The nine layers

The eleven dimensions

Three kinds of "dangerous"

Layer 1: Cluster manager

local[*]: one JVM

Spark Standalone

Hadoop YARN

Kubernetes

Managed serverless Spark

Choosing

Layer 2: Code and APIs

DataFrame / SQL API

RDD API

The Python UDF ladder

pandas API on Spark

Scala / Java UDF

Built-in functions only

Spark Connect

Thrift server (JDBC / BI)

Structured Streaming and its three companions

Choosing

Layer 3a: Execution engine

Catalyst + Tungsten (JVM): stock Spark

1.4 What `trips` actually is

3.4 Why there's still a `Filter` operator

`local[*]`: one JVM

`autoBroadcastJoinThreshold = 100 MB`

🧨 `autoBroadcastJoinThreshold = 2 GB`

Tune `spark.sql.shuffle.partitions`

🧨 `spark.sql.shuffle.partitions = 1`

`cache()` a reused DataFrame (`MEMORY_AND_DISK`)

⚠️ `persist(MEMORY_ONLY)` on a huge dataset

`MEMORY_ONLY_SER` and Kryo

Arrow for `toPandas()` and `createDataFrame()`

`spark.driver.maxResultSize` sized on purpose

🧨 `collect()` / `toPandas()` the whole table

`checkpoint()` long lineages

`spark.locality.wait = 0` on object storage

🧨 `spark.task.maxFailures = 1`