Bucketing

01 · What it is

Pre-shuffle a table by its key, once, on write

Bucketing physically pre-shuffles a table by a key once, when you write it, and stores the result as a fixed number of hash-partitioned files — the "buckets." Any later join or aggregation on that key can then skip the shuffle, because the data already sits exactly where the engine would otherwise have moved it.

analogy A shuffle join is two delivery firms each re-sorting all their parcels by postcode at the depot, every morning, before they can be matched. Bucketing is pre-sorting the warehouse shelves by postcode once — every future match just reads the right shelf.

Three moves make a bucketed table, and the rest of this lesson keeps pointing back to them rather than re-listing:

• You write with df.write.bucketBy(n, "key").sortBy("key").saveAsTable("cat.sch.tbl").
• Spark sends every row to bucket hash(key) % n, and (with sortBy) sorts within each bucket.
• Two tables bucketed by the same key into the same n have matching layouts — corresponding buckets are co-located — so the planner omits the Exchange before the join.

02 · Why it matters

The one tool for big ⋈ big on a repeated key

• The shuffle is the most expensive thing a wide operation does — network I/O + disk spill + serialization + a stage boundary where every executor waits for the slowest. Eliminating it is the point of the whole track.
• Broadcast (Lesson 02) only helps when one side is small. When you join big ⋈ big repeatedly on the same key — a daily transactions ⋈ accounts — neither side broadcasts and you re-pay the double shuffle every run. Bucketing is the fix nothing else covers.
• It's a classic senior probe: "You can't broadcast and the join is the same every day — what do you do?" The answer is "bucket both tables on the join key into matching buckets so the Exchange disappears" — and you prove it in df.explain(). The how is the next section.

03 · How it works — deep dive

Write once, and every future join inherits the layout

A sort-merge join (Lesson 02) needs matching keys in the same partition on both sides; it gets there by shuffling both sides into spark.sql.shuffle.partitions (200) hash buckets, then sorting and merging. Bucketing performs that exact hash-partitioning ahead of time and stores it. Each idea below is explained here once — later sections show it, verify it, and help you decide, but don't re-explain it.

3.1 · Writing a bucketed table (bucketBy + sortBy + saveAsTable)

• Mechanism. bucketBy(n, "key") hash-partitions the output into exactly n files using HashPartitioning(key). sortBy("key") sorts each bucket, so a later join does an efficient sort-merge inside each bucket with no separate sort stage. saveAsTable(...) records the bucket spec (key + count) in the metastore / Unity Catalog for every future query.
• Why it works. When the planner sees a join whose required distribution is HashPartitioning(key, n) and the table already advertises exactly that, it knows the data is already where it needs to be and drops the pre-join Exchange.
• Trade-off. You pay a full shuffle + sort at write time, and re-pay it on every full rewrite. It only pays off when the table is joined on that key many times.

use case a 2 TB transactions table joined to a 300 GB accounts table on account_id every day — bucket both by account_id into the same n once; every daily join then runs with no Exchange.

3.2 · Matching bucket counts → no Exchange (the core mechanism)

• Mechanism. For the planner to skip the shuffle, two conditions must hold, straight from the SQL internals: (1) the number of partitions on both sides must be exactly the same (same bucket count), and (2) both join operators must use HashPartitioning on the join key — which is precisely what bucketBy produces.
• Why it works. Bucket i of table A and bucket i of table B hold the same set of keys (same hash, same modulus), so they join locally, partition-for-partition, with no data movement.
• Trade-off. It's brittle — it holds only when the key and the count line up on both sides. Bucket into 256 on one and 200 on the other and the planner shuffles one side to reconcile them (§3.4).

analogy two decks of cards already sorted into the same 256 numbered trays — to find matching pairs you compare tray-to-tray; you never re-deal.

3.3 · Bucketing vs partitioning

These are the two on-disk layout tools, and they solve different problems — the single distinction people most often blur:

They stack, they don't compete: partitionBy("event_date").bucketBy(256, "user_id").sortBy("user_id") prunes by date and removes the shuffle on user_id joins.

use case an events table partitioned by event_date (one directory per day) and bucketed by user_id (no shuffle on the join to users).

3.4 · Mismatched counts & coalesceBucketsInJoin (Rung 3)

• Mechanism. Different bucket counts → mismatched HashPartitioning → the planner shuffles one side. Since 3.1, spark.sql.bucketing.coalesceBucketsInJoin.enabled lets Spark coalesce the larger side down to the smaller count when one count is an exact multiple of the other (e.g. 256 → 64), avoiding the shuffle.
• Interview gotcha. The default is false (since 3.1.0). It's easy to assume it's on — it is not. And it only helps for multiple-of counts; 200 vs 256 still shuffles.
• Trade-off. Coalescing reads the larger side as fewer, bigger buckets — convenient, but the best practice is to pick one bucket count and use it on both tables.

3.5 · Classic bucketing vs the DataSource-V2 flag (don't confuse them)

• spark.sql.sources.bucketing.enabled = true (default, since 2.0.0) — the classic file-source bucketing this lesson is about.
• spark.sql.sources.v2.bucketing.enabled = true (since 3.3.0) — a different feature: the DataSource-V2 storage-partitioned join for V2 connectors that report their own partitioning. Same word, different machinery — don't cite the V2 flag for classic bucketing.

Figure A · where the rows go

See it run across the cluster

Step a daily transactions ⋈ accounts join on account_id across 1 driver + 3 executors. First the unbucketed shuffle moves both tables over the network; then bucketing pre-places matching keys so the buckets are co-located and the join runs locally with no Exchange; finally watch what a bucket-count mismatch brings back. Use Prev / Play / Next — the thing to watch is whether the wire between nodes lights up (a shuffle) or stays dark (co-located, local join).

04 · The code — and how to verify it

Four moves, each with its proof

Every technique below is paired with how to prove it worked — the .explain() plan node or the Spark-UI signal. The proof for bucketing is always the absence of Exchange under the join. Apply, then verify; never assume.

Write two tables bucketed by the same key into the same N

python# Bucket BOTH tables by the join key (account_id) into the SAME number of buckets.
# bucketBy(n, key) -> HashPartitioning(account_id) into n files (the pre-shuffle)
# sortBy(key)      -> sort within each bucket so the read-side join skips its Sort too
# saveAsTable(...) -> REQUIRED: bucketing only works through a metastore/UC table
N = 256

(transactions.write
    .bucketBy(N, "account_id").sortBy("account_id")
    .mode("overwrite")
    .saveAsTable("main.pyspark_perf.transactions_bucketed"))

(accounts.write
    .bucketBy(N, "account_id").sortBy("account_id")   # SAME key, SAME N
    .mode("overwrite")
    .saveAsTable("main.pyspark_perf.accounts_bucketed"))

# VERIFY the stored bucket spec:
spark.sql("DESCRIBE EXTENDED main.pyspark_perf.transactions_bucketed").show(50, False)
#   -> Num Buckets: 256   Bucket Columns: [account_id]   Sort Columns: [account_id]

spark.sql.sources.bucketing.enabled is true by default — you don't set it to use bucketing; you'd only set it to false to prove the before-state.

Join the bucketed tables and confirm the Exchange is gone

pythont = spark.table("main.pyspark_perf.transactions_bucketed")
a = spark.table("main.pyspark_perf.accounts_bucketed")

joined = t.join(a, "account_id")        # same key both tables are bucketed on

# VERIFY: a SortMergeJoin with NO Exchange under either side = shuffle eliminated.
joined.explain(mode="formatted")
#   == Physical Plan ==
#   * SortMergeJoin [account_id], Inner
#   :- * FileScan ... transactions_bucketed   <- no Exchange
#   +- * FileScan ... accounts_bucketed        <- no Exchange

Contrast: the unbucketed join (the shuffle you're removing)

python# Unbucketed big-vs-big join: both sides shuffled (two Exchange nodes).
slow = transactions.join(accounts, "account_id")
slow.explain(mode="formatted")
#   * SortMergeJoin [account_id], Inner
#   :- * Sort  +- Exchange hashpartitioning(account_id, 200)  <- 2 TB shuffled
#   +- * Sort  +- Exchange hashpartitioning(account_id, 200)  <- 300 GB shuffled
# VERIFY: Spark UI SQL tab — `slow` has two Exchange nodes; the bucketed join has none.

Stack partitioning (prune) with bucketing (no shuffle)

python# Prune by a low-cardinality filter column AND remove the shuffle on the join key.
(events.write
    .partitionBy("event_date")                     # directories -> partition pruning (L07)
    .bucketBy(256, "user_id").sortBy("user_id")     # hash files  -> no shuffle on user_id
    .mode("overwrite")
    .saveAsTable("main.pyspark_perf.events_bpb"))

# VERIFY: the plan shows PartitionFilters on event_date AND no Exchange on the user_id join.

Equivalent Spark SQL — bucketing declared in DDL:

sql-- CLUSTERED BY ... INTO n BUCKETS is the SQL form of bucketBy/sortBy.
CREATE TABLE main.pyspark_perf.accounts_bucketed
CLUSTERED BY (account_id) SORTED BY (account_id) INTO 256 BUCKETS
AS SELECT * FROM main.pyspark_perf.accounts_raw;

Figure B · the shuffle eliminator

Watch the Exchange disappear

This figure isolates a different thing from Figure A: not where the rows land, but what the plan does. Toggle "bucket both by the key" and watch the Exchange vanish from the join plan. Then mismatch the bucket counts — the shuffle returns — and flip coalesceBucketsInJoin to rescue the multiple-of case. Watch the "Exchange before join?" readout and the plan box.

05 · Decision table

Bucketing vs the layouts it sits beside

Read this by the problem you're solving — it answers "which tool removes which cost?" at a glance. (The why is §3; this is the which.)

06 · Edge cases & limitations

Where it gets subtle

New nuance an interviewer probes — not a re-run of §3:

• Bucketing only one side. Both tables must share the same key and the same N — bucketing one alone doesn't help; the other side still shuffles to match, and you've paid a write-time shuffle for nothing.
• Bucketing a column you never join on. Bucketing on customer_id does nothing for a join on account_id. The bucket key must be the key you actually join / aggregate by, repeatedly.
• AQE doesn't retro-fit bucketing. AQE coalesces post-shuffle partitions; if only one side is bucketed, the other still shuffles. Bucketing acts at the layout layer, before any shuffle exists to coalesce.
• Over- or under-bucketing. Bucket count drives file size — too many → tiny files, too few → giant, skew-prone buckets. Size N to the table's size and key cardinality.

Hard limits worth stating plainly:

• saveAsTable() only. Bucketing is not supported for DataFrameWriter.save, .insertInto, or .jdbc — you must write a metastore / UC-registered table, not plain files in a path.
• Re-bucketing is a full rewrite. Changing the key or count means re-shuffling the whole table — a real, recurring cost on every full rewrite.
• Classic ≠ DataSource-V2. spark.sql.sources.bucketing.enabled (this lesson) is not spark.sql.sources.v2.bucketing.enabled (since 3.3.0, the storage-partitioned-join flag).

07 · Common mistakes

Gotchas — the mistake and its one-line fix

08 · References

Sources

• Apache Spark — SQL Performance Tuning (bucketing, join strategies, AQE)
• Apache Spark — Configuration (bucketing.enabled, coalesceBucketsInJoin, v2.bucketing.enabled, shuffle.partitions)
• Apache Spark — DataFrameWriter.bucketBy / sortBy / saveAsTable (Python API)
• Azure Databricks — Adaptive Query Execution
• Azure Databricks — Unity Catalog (saveAsTable, three-level names)