How to Use Bucketing in Spark for Efficient Data Processing

In Spark, bucketing organizes data into fixed number of buckets based on a column, improving join and query performance. You use bucketBy(numBuckets, colName) when writing a DataFrame and then save it as a table or file. Bucketing helps Spark avoid full data shuffles during joins.

Syntax

The main syntax for bucketing in Spark is used when writing a DataFrame. You specify the number of buckets and the column(s) to bucket by.

• bucketBy(numBuckets, colName): Defines how many buckets and which column to use.
• sortBy(colName) (optional): Sorts data within each bucket.
• saveAsTable(tableName) or save(path): Saves the bucketed data as a table or file.

scala

df.write.bucketBy(4, "user_id").sortBy("timestamp").saveAsTable("bucketed_table")

Example

This example shows how to create a bucketed table by the column user_id with 4 buckets, then how to read and join bucketed tables efficiently.

python

from pyspark.sql import SparkSession

spark = SparkSession.builder.appName("BucketingExample").enableHiveSupport().getOrCreate()

# Sample data
data = [(1, "Alice", 100), (2, "Bob", 200), (3, "Cathy", 300), (4, "David", 400)]
columns = ["user_id", "name", "score"]
df = spark.createDataFrame(data, columns)

# Write bucketed table
(df.write
   .bucketBy(4, "user_id")
   .sortBy("user_id")
   .mode("overwrite")
   .saveAsTable("bucketed_users"))

# Read bucketed table
bucketed_df = spark.table("bucketed_users")
bucketed_df.show()

# Join example (bucketed join optimization)
other_data = [(1, "NY"), (2, "LA"), (3, "SF"), (4, "TX")]
other_df = spark.createDataFrame(other_data, ["user_id", "city"])

joined_df = bucketed_df.join(other_df, "user_id")
joined_df.show()

Output

+-------+-----+-----+
|user_id| name|score|
+-------+-----+-----+
|      1|Alice|  100|
|      2|  Bob|  200|
|      3|Cathy|  300|
|      4|David|  400|
+-------+-----+-----+
+-------+-----+-----+----+
|user_id| name|score|city|
+-------+-----+-----+----+
|      1|Alice|  100|  NY|
|      2|  Bob|  200|  LA|
|      3|Cathy|  300|  SF|
|      4|David|  400|  TX|
+-------+-----+-----+----+

Common Pitfalls

• Not enabling Hive support: Bucketing with saveAsTable requires Hive support enabled in SparkSession.
• Mismatch in bucket number or columns: When joining bucketed tables, both must have the same number of buckets and bucket columns for optimization.
• Using bucketing with file formats that don't support it well: Bucketing works best with Hive tables or formats like Parquet.
• Forgetting to use sortBy: Sorting inside buckets can improve query performance but is optional.

scala

## Wrong: Different bucket numbers
(df.write.bucketBy(4, "user_id").saveAsTable("table1"))
(df.write.bucketBy(6, "user_id").saveAsTable("table2"))

## Right: Same bucket numbers
(df.write.bucketBy(4, "user_id").saveAsTable("table1"))
(df.write.bucketBy(4, "user_id").saveAsTable("table2"))

Quick Reference

Use this quick guide to remember bucketing steps and tips:

Step	Description
Enable Hive support	Required for saveAsTable with bucketing.
Use bucketBy(numBuckets, colName)	Define number of buckets and bucket column.
Optionally use sortBy(colName)	Sort data inside each bucket for faster queries.
Save as table or file	Use saveAsTable or save to store bucketed data.
Match buckets for joins	Ensure bucket count and columns match for bucketed join optimization.

Key Takeaways

Bucketing splits data into fixed buckets based on columns to optimize joins and queries.

Always enable Hive support in SparkSession to use bucketing with tables.

Bucketed tables must have matching bucket counts and columns for efficient joins.

Use sortBy to order data inside buckets and improve query speed.

Bucketing works best with Hive tables and file formats like Parquet.