Databricks Production: Broadcast Joins Appropriately

Broadcast joins in Databricks are a powerful optimization technique to speed up joins between a large table (the “reducer”) and a smaller table (the “broadcaster”). They’re especially effective when the smaller table can fit comfortably in the memory of each worker node. This tutorial will guide you through understanding and applying broadcast joins effectively.

Understanding Broadcast Joins

A broadcast join works by creating a copy of the smaller table and joining it to the larger table using this copy. This avoids shuffling large amounts of data across the network, significantly reducing the execution time. It’s crucial to determine if your smaller table is large enough to be broadcasted effectively. The general rule of thumb is that the smaller table should be significantly smaller than the larger table – ideally less than 1GB for optimal performance. Databricks automatically detects and utilizes broadcast joins when appropriate.

Example 1: Simple Broadcast Join

Let’s start with a simple example demonstrating a basic broadcast join. We’ll create two tables: a large table containing sales data and a smaller table containing product categories.

from pyspark.sql import SparkSession
from pyspark.sql.functions import broadcast

# Create SparkSession
spark = SparkSession.builder.appName("BroadcastJoinExample").getOrCreate()

# Create a large sales data table
sales_data = [
    (1, "Product A", 100),
    (2, "Product B", 200),
    (3, "Product A", 150),
    (4, "Product C", 300),
    (5, "Product B", 250)
]
sales_df = spark.createDataFrame(sales_data, ["sale_id", "product_name", "sales_amount"])

# Create a smaller product categories table
product_categories = [
    (1, "Electronics"),
    (2, "Clothing"),
    (3, "Electronics"),
    (4, "Home Goods")
]
product_categories_df = spark.createDataFrame(product_categories, ["category_id", "category_name"])

# Perform a broadcast join
joined_df = sales_df.join(broadcast(product_categories_df), "category_id")

# Show the result
joined_df.show()

# Stop SparkSession
spark.stop()

In this example:

• We create a SparkSession.
• `sales_data` and `product_categories` are defined as lists.
• `sales_df` and `product_categories_df` are created from these lists using `spark.createDataFrame()`.
• `broadcast(product_categories_df)` creates a broadcast variable containing a copy of `product_categories_df`.
• `sales_df.join(…)` performs a join using the broadcasted table.
• `joined_df.show()` displays the result.

Important: The `broadcast()` function is crucial. Without it, a regular join would be performed, which is less efficient for this scenario. Note that spark automatically detects the use of broadcast joins.

# Output
# +-------+-----------+-------------+------------------+
# |sale_id|product_name|sales_amount |category_name    |
# +-------+-----------+-------------+------------------+
# |      1|Product A  |         100 |Electronics       |
# |      2|Product B  |         200 |Clothing          |
# |      3|Product A  |         150 |Electronics       |
# |      4|Product C  |         300 |Home Goods        |
# |      5|Product B  |         250 |Clothing          |
# +-------+-----------+-------------+------------------+

Example 2: Handling Nulls

Let’s examine what happens when there are nulls in the smaller table. Broadcast joins gracefully handle nulls, effectively joining rows based on matching values, and filling in nulls from the broadcaster table where there’s no match in the larger table.

from pyspark.sql import SparkSession
from pyspark.sql.functions import broadcast

# Create SparkSession
spark = SparkSession.builder.appName("BroadcastJoinWithNulls").getOrCreate()

# Create sales data table with a null category_id
sales_data = [
    (1, "Product A", 100, None),
    (2, "Product B", 200, 2),
    (3, "Product A", 150, 1),
    (4, "Product C", 300, 4)
]
sales_df = spark.createDataFrame(sales_data, ["sale_id", "product_name", "sales_amount", "category_id"])

# Create product categories table
product_categories = [
    (1, "Electronics"),
    (2, "Clothing"),
    (3, "Electronics"),
    (4, "Home Goods")
]
product_categories_df = spark.createDataFrame(product_categories, ["category_id", "category_name"])

# Perform a broadcast join
joined_df = sales_df.join(broadcast(product_categories_df), "category_id")

# Show the result
joined_df.show()

# Stop SparkSession
spark.stop()

In this version, we introduced a `None` value in the `category_id` column of the `sales_df`. The join will still function, and the `category_name` will be populated with the corresponding values from the `product_categories_df`.

# Output
# +-------+-----------+-------------+------------------+
# |sale_id|product_name|sales_amount |category_name    |
# +-------+-----------+-------------+------------------+
# |      1|Product A  |         100 |Electronics       |
# |      2|Product B  |         200 |Clothing          |
# |      3|Product A  |         150 |Electronics       |
# |      4|Product C  |         300 |Home Goods        |
# +-------+-----------+-------------+------------------+

Example 3: Optimizing Broadcast Join Size

Sometimes, even though a table seems small, it might be too large to broadcast efficiently. You can control the memory allocated to the broadcasted table using the `spark.sql.shuffle.partitions` configuration. Increasing this value will allow for more memory allocation per worker node.

from pyspark.sql import SparkSession
from pyspark.sql.functions import broadcast

# Create SparkSession
spark = SparkSession.builder.appName("BroadcastJoinSizeOptimization").getOrCreate()

# Create sales data table
sales_data = [(1, "Product A", 100), (2, "Product B", 200)]
sales_df = spark.createDataFrame(sales_data, ["sale_id", "product_name", "sales_amount"])

# Create product categories table
product_categories = [(1, "Electronics"), (2, "Clothing")]
product_categories_df = spark.createDataFrame(product_categories, ["category_id", "category_name"])

# Set shuffle partitions to 8.  This will increase memory usage.
spark.conf.set("spark.sql.shuffle.partitions", "8")

# Perform a broadcast join
joined_df = sales_df.join(broadcast(product_categories_df), "category_id")

# Show the result
joined_df.show()

# Stop SparkSession
spark.stop()

By setting `spark.sql.shuffle.partitions` to 8, we’re instructing Spark to use 8 partitions for the shuffle operations. This will increase the memory available for the broadcasted table. Adjust this value based on your data and cluster resources. Experimentation is key.

# Output
# +-------+-----------+-------------+------------------+
# |sale_id|product_name|sales_amount |category_name    |
# +-------+-----------+-------------+------------------+
# |      1|Product A  |         100 |Electronics       |
# |      2|Product B  |         200 |Clothing          |
# +-------+-----------+-------------+------------------+