Definition
- Bucket pruning – an equality filter on the bucketing column computes which single bucket the value hashes to and reads only that file.
- Shuffle-free aggregation – a
groupByon the bucketing column needs no shuffle because every key's rows already live in one bucket. - Bucket-aware join – two tables bucketed on the same column with the same bucket count join bucket-to-bucket without shuffling either side.
- Invariant – all three optimizations flow from one fact: the same key always hashes to the same bucket.
- Trade-off – bucketing is a write-time investment that only benefits queries on the bucketed column.
The National Bank of India keeps account records in 32 drawers chosen by a fixed hash rule on account number. Bucket pruning: asking for one account opens only one drawer. Shuffle-free aggregation: totalling each account's transactions needs no gathering because slips are already together. Bucket-aware join: matching two record sets that both use the same 32-drawer rule only pairs drawer 1 with drawer 1.
Understanding the Concepts
What is bucketing?
Bucketing pre-distributes data at write time by hashing a column into a fixed number of files (buckets). When Spark later reads the table, it knows which bucket contains each key value, enabling three powerful optimizations.
from pyspark.sql import SparkSession, functions as F
spark = SparkSession.builder \
.appName("NBI-Bucketing") \
.config("spark.sql.sources.bucketing.enabled", "true") \
.getOrCreate()
# Write transactions bucketed by accountId into 32 buckets
txns = spark.read.parquet("s3://nbi-data-lake/transactions/")
txns.write \
.bucketBy(32, "accountId") \
.sortBy("accountId") \
.saveAsTable("nbi.transactions_bucketed")
# Write accounts bucketed by the SAME column and SAME count
accounts = spark.read.parquet("s3://nbi-data-lake/accounts/")
accounts.write \
.bucketBy(32, "accountId") \
.sortBy("accountId") \
.saveAsTable("nbi.accounts_bucketed")
How does bucket pruning work?
When a query filters on the bucketing column with an equality predicate, Spark computes hash(value) % numBuckets to find the one bucket that can contain that value and reads only that file, skipping the other 31.
# Read the bucketed table through the catalog
txns_b = spark.table("nbi.transactions_bucketed")
# Equality filter on bucketing column → bucket pruning
result = txns_b.filter(F.col("accountId") == "SAV-100042")
result.explain("formatted")
# Only 1 of 32 bucket files is read!
# This is like partition pruning but at the file level inside the table.
== Physical Plan ==
*(1) Filter (accountId = SAV-100042)
+- FileScan parquet nbi.transactions_bucketed
SelectedBucketsCount: 1 out of 32
How does shuffle-free aggregation work?
A groupBy on the bucketing column normally requires a shuffle to bring all rows for each key together. But in a bucketed table, rows for each key already live in the same bucket. Spark recognizes this and skips the shuffle.
# GroupBy on bucketing column → NO shuffle
per_account = txns_b.groupBy("accountId").agg(
F.sum("amount").alias("total"),
F.count("*").alias("txnCount")
)
per_account.explain("formatted")
== Physical Plan ==
*(1) HashAggregate(keys=[accountId], functions=[sum(amount), count(1)])
+- *(1) HashAggregate(keys=[accountId], functions=[partial_sum(amount), partial_count(1)])
+- FileScan parquet nbi.transactions_bucketed
No Exchange (shuffle) node! Aggregation is local per bucket.
How does bucket-aware join work?
When two tables are bucketed on the same column with the same bucket count, Spark joins bucket-to-bucket with no shuffle of either side. If both are also sorted, the join is a local sort-merge with no sort step.
# Both tables bucketed on accountId with 32 buckets
txns_b = spark.table("nbi.transactions_bucketed")
accts_b = spark.table("nbi.accounts_bucketed")
# Join on the bucketing column → NO shuffle
result = txns_b.join(accts_b, "accountId")
result.explain("formatted")
== Physical Plan ==
*(1) SortMergeJoin [accountId], [accountId], Inner
:- FileScan parquet nbi.transactions_bucketed
+- FileScan parquet nbi.accounts_bucketed
No Exchange nodes! Bucket 0 joins with bucket 0, etc.
What are the requirements for bucket-aware join?
Both tables must share the exact same bucket count and bucketing column. A mismatch forces Spark to re-shuffle one or both sides.
| Condition | Result |
|---|---|
| Same column, same count | Shuffle-free join |
| Same column, different count | Spark may coalesce if one is a multiple of the other, but do not rely on it |
| Different columns | Full shuffle, no benefit |
| Reading via path (not catalog) | Bucketing metadata lost, full shuffle |
# WRONG: different bucket counts
txns.write.bucketBy(32, "accountId").saveAsTable("nbi.txns_32")
accounts.write.bucketBy(64, "accountId").saveAsTable("nbi.accts_64")
# Join will shuffle one side to match the other!
# RIGHT: same bucket count
txns.write.bucketBy(32, "accountId").saveAsTable("nbi.txns_32")
accounts.write.bucketBy(32, "accountId").saveAsTable("nbi.accts_32")
# Shuffle-free join!
When does bucket pruning NOT help?
Bucket pruning works only for equality and IN predicates. Range filters (greater than, between) span many hash buckets and cannot benefit.
# GOOD: equality filter → bucket pruning
txns_b.filter(F.col("accountId") == "SAV-100042") # 1 bucket read
# GOOD: IN filter → bucket pruning
txns_b.filter(F.col("accountId").isin("SAV-100042", "CUR-200001")) # 2 buckets
# NO BENEFIT: range filter spans many hash buckets
txns_b.filter(F.col("accountId") > "SAV-100000") # All buckets read
How do you choose the bucket count?
The bucket count should roughly match the cluster's parallelism. Too few buckets under-utilize cores. Too many create excessive small files.
# Cluster: 30 executors x 5 cores = 150 total cores
# Good bucket count: 128 or 256 (power of 2 near total cores)
txns.write \
.bucketBy(128, "accountId") \
.sortBy("accountId") \
.saveAsTable("nbi.transactions_bucketed_128")
# Each bucket becomes one task, so 128 buckets → 128 tasks
# With 150 cores, all buckets run in one wave
What should a senior engineer know?
A senior engineer knows that bucketing metadata is only visible when reading through the catalog (table name), not when reading via a raw file path. They also know that bucketing is a write-time investment: writing takes longer because of hashing and sorting, but reads are dramatically faster for the specific bucketed column.
# WRONG: reading via path loses bucketing metadata
txns_path = spark.read.parquet("/warehouse/nbi/transactions_bucketed/")
# Spark does NOT know this is bucketed! No optimizations.
# RIGHT: reading via catalog preserves bucketing metadata
txns_table = spark.table("nbi.transactions_bucketed")
# Spark sees bucketing metadata → all optimizations apply
Visual Overview
Bucketing write (one-time cost):
Raw transactions
┌─────────────────────────────────┐
│ SAV-1001, 500 │ CUR-2001, 800 │
│ SAV-1002, 300 │ SAV-1001, 200 │
│ CUR-2001, 100 │ SAV-1002, 400 │
└─────────────────────────────────┘
│ hash(accountId) % 32
▼
Bucket 0 Bucket 1 ... Bucket 31
┌────────┐ ┌────────┐ ┌────────┐
│SAV-1001│ │CUR-2001│ │SAV-1002│
│SAV-1001│ │CUR-2001│ │SAV-1002│
└────────┘ └────────┘ └────────┘
Bucket-aware join (no shuffle):
transactions_bucketed accounts_bucketed
Bucket 0 ◄──────────► Bucket 0 (local join)
Bucket 1 ◄──────────► Bucket 1 (local join)
... ...
Bucket 31 ◄─────────► Bucket 31 (local join)
No data moves across the network!
Key Takeaway
Bucketing optimizes queries in three ways, all flowing from one invariant: the same key always hashes to the same bucket. Bucket pruning reads only the single bucket matching an equality filter. Shuffle-free aggregation works because all rows for a key already live together. Bucket-aware joins match bucket-to-bucket without shuffling either side.
The requirements are strict: both join sides must share the exact same bucket count and column, queries must read through the catalog table so Spark sees the bucketing metadata, and pruning only works for equality and IN predicates, not ranges. Bucketing is a write-time investment that pays off only for the specific column you bucketed on.
At the National Bank of India we bucketed the transactions and accounts tables by accountId into 32 buckets each. Customer-service lookups by account read a single bucket. Nightly per-account totals aggregated without shuffling. The daily transactions-to-accounts join stopped shuffling two huge tables and ran bucket-to-bucket, which was the biggest performance win.
The senior nuance is that reading via a raw file path instead of a catalog table name loses all bucketing metadata, so no optimizations apply. Always read bucketed tables through the catalog.
Remember one sentence: bucketing pre-distributes data by key at write time so reads can prune to one bucket, aggregate without shuffle, and join bucket-to-bucket without shuffle.
---