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

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.

---

Table of Contents

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

1. 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.
2. 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.
3. 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:

1. 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.
2. 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.
3. 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.
4. 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().
5. 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:

1. picks a destination partition by binary-searching the 199 boundaries (fare = 215.40 → partition 87),
2. 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,
3. 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:

1. asks the MapOutputTracker for the locations of block i from every one of the 1,364 map outputs,
2. 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),
3. 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,
4. 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:

1. AQE executes the plan one shuffle stage at a time, leaf-first. Stages 5 and 6 are both submitted.
2. 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.
3. 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.
4. 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

1. What's in PartitionFilters? If your date/region/tenant filter isn't there, you're scanning everything.
2. 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.
3. 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.
4. Which joins are BroadcastHashJoin? Anything that should be and isn't is a size-estimate problem; fix it with broadcast() or better statistics.
5. 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:

1. 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.
2. 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.

---

Further reading

Official docs

• Spark SQL performance tuning: caching, join hints, and the full AQE section
• Spark configuration reference and Spark SQL options
• Parquet data source options: schema merging, partition discovery, pushdown settings
• Apache Parquet file format and the page index spec

Source files, if you want to see the real thing (all under apache/spark)

• sql/core/.../execution/datasources/FilePartition.scala: maxSplitBytes and the bin-packing from Chapter 2
• sql/core/.../execution/DataSourceScanExec.scala: FileSourceScanExec, partition filters, dynamic pruning, scan metrics
• sql/core/.../execution/exchange/ShuffleExchangeExec.scala: how a shuffle dependency is built, including the RangePartitioner sampling
• core/src/main/scala/org/apache/spark/Partitioner.scala: RangePartitioner.sketch, the sampling job
• sql/core/.../execution/exchange/BroadcastExchangeExec.scala: the driver-side build, the 8 GB / 512 M row limits, the timeout
• sql/core/.../execution/SparkStrategies.scala: JoinSelection, the decision tree in Chapter 5
• sql/core/.../execution/adaptive/AdaptiveSparkPlanExec.scala: stage-by-stage re-planning, coalescing, skew handling
• core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala: stages cut at shuffle dependencies

Longer reads

• The Internals of Spark SQL by Jacek Laskowski: an operator-by-operator reference
• Spark: The Definitive Guide (Chambers & Zaharia), Part II, for the DataFrame execution model in long form

This is the first article in a series on distributed computing, told through one dataset and one team. Next up: what happens when you call .write, and why your output has 200 files. If something here doesn't match what you see in your Spark UI, tell me in the comments: versions move, and I'd rather fix it than be quoted wrong.