What is Bucketing?

How it works

Partitioning and bucketing answer different questions. partitionBy creates one folder per value and suits low-cardinality columns you filter on, such as a date. bucketBy creates a fixed number of files per partition by hashing the column, and suits high-cardinality columns you join or group on, such as a customer id.

In Spark, bucketing needs a table in a catalog, so the write must use saveAsTable; a plain save(path) does not record the bucket spec. When two tables are bucketed on the join key with the same number of buckets, bucket i of one can only match bucket i of the other. Spark reads both as already partitioned and removes the Exchange (the shuffle) from both sides of the join. The shuffle is paid once, at write time, instead of on every query.

The limits are real:

• Each writing task writes its own file for every bucket it holds rows for, so the file count can reach tasks multiplied by buckets. Repartition by the bucket column to the bucket count before writing.
• Bucket specs are a Spark and Hive-style catalog feature; other engines may ignore them, and Spark’s bucket hash differs from Hive’s.
• Changing the bucket count means rewriting the table.
• Delta Lake does not support bucketBy; for Delta tables use partitioning, Z-ordering or liquid clustering.

Example

Two tables bucketed into 4 buckets on customer_id. The join plan has no Exchange, so neither side is shuffled (PySpark 4.2.0, local warehouse folder).

import tempfile
from pyspark.sql import SparkSession

spark = (SparkSession.builder.master("local[2]").appName("glossary-bucketing")
         .config("spark.sql.warehouse.dir", tempfile.mkdtemp())
         .config("spark.sql.autoBroadcastJoinThreshold", "-1")   # force a shuffle-style join
         .getOrCreate())
spark.sparkContext.setLogLevel("ERROR")

orders = spark.range(1000).selectExpr("id AS order_id", "id % 50 AS customer_id")
customers = spark.range(50).selectExpr("id AS customer_id", "'segment' AS segment")
for name, df in [("orders_b", orders), ("customers_b", customers)]:
    df.repartition(4, "customer_id").write.bucketBy(4, "customer_id").sortBy("customer_id").saveAsTable(name)

joined = spark.table("orders_b").join(spark.table("customers_b"), "customer_id")
plan = joined._jdf.queryExecution().executedPlan().toString()
print("Exchange in plan:", "Exchange" in plan)
print("rows:", joined.count())
spark.stop()

Exchange in plan: False
rows: 1000

Common mistakes

• Bucketing on a column nobody joins or groups on, which adds write cost for no benefit.
• Using different bucket counts on the two sides of a join, so the shuffle comes back.
• Writing without repartitioning first and ending up with thousands of small files.
• Expecting Trino, Athena or Delta Lake to honour a Spark bucket spec.