🔥 Spark & Databricks · Chapter 5

Partitioning, bucketing, broadcast joins

Three tools, three different problems: partitioning skips files, bucketing skips shuffles, broadcasting skips the big side entirely.

Study this lesson free in the app →

The basics

Partitioning, bucketing and broadcasting are three ways of laying out or moving data so Spark does less work.

Why they exist: the two most expensive things Spark does are reading data it did not need and shuffling data across the network. Each tool removes one of those:

• On-disk partitioning (partitionBy('ride_date')) writes one folder per value. A query filtering on ride_date opens only that folder — partition pruning.
• Bucketing (bucketBy(64, 'driver_id')) pre-hashes rows into a fixed number of files per table. Two tables bucketed the same way can be joined without a shuffle.
• Broadcast join copies a small table to every executor so the big table is joined where it sits — no shuffle of the big side.

Mental model: a library. Partitioning is shelving by year so you walk to one aisle. Bucketing is two libraries using the same shelf numbering so matching books are always on matching shelves. Broadcasting is photocopying the one-page index for every reader.

Explained

Meghalaya Cabs' analysts query rides almost always with WHERE ride_date = .... The table is one flat folder of 90,000 Parquet files, so every query lists and checks all of them.

Rewritten with partitionBy('ride_date'), a one-day query opens one folder of ~50 files.

Then someone 'improves' it to partitionBy('ride_date', 'city', 'hour'). Now there are 4 million files of 30KB, and the listing alone takes longer than the old full scan.

The library

Shelving by year — you want 2024 books, you walk to one aisle. But shelve by year and author initial and colour, and each shelf holds two books; you spend the day walking.

Two branches with the same numbering — the Shillong and Tura branches both put author 'K' books on shelf 11. To match their catalogues you compare shelf 11 with shelf 11. Nobody has to move books between branches first.

The photocopied index — the 40-page catalogue of rare books is copied for every reader. Nobody queues at one desk; each reader checks their own pile against their own copy.

Three layouts side by side

PARTITIONED (partitionBy ride_date)
s3://mc-lake/silver/rides/
  ride_date=2026-03-13/ part-000.parquet ... part-049
  ride_date=2026-03-14/ part-000.parquet ... part-049   ← WHERE ride_date='2026-03-14'
  ride_date=2026-03-15/ ...                                 reads only this folder

BUCKETED (bucketBy 64 driver_id) — rides and payouts both
  rides:   bucket 00 | bucket 01 | ... | bucket 63
  payouts: bucket 00 | bucket 01 | ... | bucket 63
            ↕ join      ↕ join           ↕ join      ← no Exchange

BROADCAST (drivers = 8MB)
  driver ──collect──► [drivers] ──copy──► executor 1..N
  each task: its rides partition ⋈ local drivers copy ← big side never moves

Writing each layout

# 1. On-disk partitioning: repartition by the same column first,
#    so each date is written by one task → few, large files
(rides.repartition('ride_date')
      .write.partitionBy('ride_date')
      .mode('overwrite').parquet('s3://mc-lake/silver/rides/'))

spark.read.parquet('s3://mc-lake/silver/rides/') \
     .filter(col('ride_date') == '2026-03-14').explain()
# ... PartitionFilters: [isnotnull(ride_date), (ride_date = 2026-03-14)]

# 2. Bucketing: needs a metastore table (saveAsTable), not a path
(rides.write.bucketBy(64, 'driver_id').sortBy('driver_id')
      .saveAsTable('mc.rides_bkt'))
(payouts.write.bucketBy(64, 'driver_id').saveAsTable('mc.payouts_bkt'))
spark.table('mc.rides_bkt').join(spark.table('mc.payouts_bkt'), 'driver_id')
# plan: SortMergeJoin with NO Exchange on either side

# 3. Broadcast
from pyspark.sql.functions import broadcast
rides.join(broadcast(drivers), 'driver_id')     # BroadcastHashJoin
spark.conf.get('spark.sql.autoBroadcastJoinThreshold')  # '10485760' = 10MB

How Spark picks a join strategy

For an equi-join, roughly in this order:

1. Broadcast hash join — if a hint says so, or one side's estimated size is under autoBroadcastJoinThreshold (10MB), and the join type allows that side to be the build side
2. Sort-merge join — the default for two large sides: Exchange + Sort both, then merge
3. Shuffled hash join — shuffle both, hash the smaller per partition; chosen when preferred via config/hint or by AQE when partitions are small
4. Broadcast nested loop / cartesian — non-equi joins; dangerous at scale

Build-side rule: broadcast must be the side whose unmatched rows are not kept. Inner: either. Left outer/semi/anti: right only. Right outer: left only. Full outer: no broadcast hash join.

The estimate comes from file sizes and table statistics. A 2GB table filtered to 3MB may still be 'too big' to the planner — unless AQE sees the real post-filter size at runtime and converts the sort-merge join into a broadcast join.

At 100×

The small-files problem. Files written by a partitioned write ≈ tasks × distinct partition values per task. 200 shuffle tasks × 365 dates = up to 73,000 files. Repartition by the partition column (or let Delta's optimized writes / OPTIMIZE compact) before writing.

Broadcast memory is driver memory first. The small side is collected to the driver, then shipped. Broadcasting a 1GB table to 200 executors costs driver heap and 200GB of network; there is a hard 8GB limit, and a 5-minute broadcastTimeout. Keep broadcasts in the tens to low hundreds of MB.

Bucketing is fragile. Both tables need the same bucket count and key, written via saveAsTable; each writing task can produce one file per bucket (200 × 64 = 12,800 files). On Databricks/Delta, bucketing is not supported — use Z-order or liquid clustering for data skipping, and rely on broadcast + AQE for joins.

Dynamic partition pruning (Spark 3.0+): joining rides (partitioned by ride_date) to a filtered calendar dimension prunes rides' partitions at runtime.

Gotchas

• Partitioning by a high-cardinality column (ride_id, driver_id) — millions of folders of one file each
• Filtering on a related column instead of the partition column — partitioned by ride_date, queried with WHERE pickup_ts >= '2026-03-14': Spark does not know the two are linked, so nothing is pruned. Add the ride_date predicate explicitly
• repartition(n) mistaken for partitionBy — the first changes tasks in memory, only the second creates folders
• Broadcasting the preserved side of an outer join — the hint is silently ignored and you get a sort-merge join
• Trusting the 10MB estimate — stale or missing stats make Spark broadcast a table that is really 3GB; driver OOM. Use ANALYZE TABLE ... COMPUTE STATISTICS or explicit hints
• Expecting bucketing to survive a write to a plain path — bucketBy only works with saveAsTable

Partition to skip files. Bucket to skip shuffles. Broadcast to skip the big side.

Partition columns: low cardinality, always filtered. Broadcast: small, non-preserved side.

Words to know

In two minutes

• Spark has two kinds of partitioning: in-memory partitions (tasks; repartition/coalesce) and on-disk partitions (folders; write.partitionBy). Don't confuse them.
• partitionBy on a low-cardinality column that queries filter on gives partition pruning — Spark lists and reads only matching folders.
• Over-partitioning (city × date × hour) creates millions of tiny files; aim for partitions of hundreds of MB to a few GB each.
• Bucketing hashes rows into N files by a key at write time (saveAsTable only); joining two tables bucketed identically on the join key skips the Exchange.
• Broadcast hash join: tables estimated under spark.sql.autoBroadcastJoinThreshold (10MB default) are collected to the driver and shipped to every executor; broadcast() forces it.
• Broadcast only works on the non-preserved side: inner either side, left outer → broadcast right, right outer → broadcast left, full outer → never.
• AQE can switch a sort-merge join to broadcast at runtime when a side turns out small after filtering; Delta tables use Z-order or liquid clustering instead of bucketing.

Test yourself

There are 8 questions, 3 puzzles and 5 recall cards for this lesson in the app — the recall cards come back on later days, just before you would forget.