Broadcast joins for small tables in Apache Spark - Mini Project: Build & Apply

Broadcast Joins for Small Tables in Apache Spark

📖 Scenario: You work at a retail company. You have a large sales dataset and a small product details dataset. You want to join them to get product names with each sale. Using broadcast join helps Spark run this join faster by sending the small product data to all worker nodes.

🎯 Goal: Learn how to perform a broadcast join in Apache Spark to efficiently join a large sales DataFrame with a small product DataFrame.

📋 What You'll Learn

Create a Spark DataFrame for sales data with columns sale_id, product_id, and quantity.

Create a Spark DataFrame for product data with columns product_id and product_name.

Use a broadcast join to join sales with product data on product_id.

Show the joined DataFrame output.

💡 Why This Matters

🌍 Real World

Broadcast joins are used in big data processing when one table is small enough to fit in memory. This speeds up joins by sending the small table to all worker nodes, avoiding expensive shuffles.

💼 Career

Data engineers and data scientists use broadcast joins to optimize Spark jobs for faster data processing and analysis, especially when working with mixed-size datasets.

1

Create the sales DataFrame

Create a Spark DataFrame called sales_df with these exact rows: (1, 101, 2), (2, 102, 1), (3, 103, 5). The columns must be sale_id, product_id, and quantity.

Apache Spark

from pyspark.sql import SparkSession

spark = SparkSession.builder.appName("BroadcastJoinExample").getOrCreate()
# Create sales_df DataFrame with specified rows and columns
# Your code here

Hint

Use spark.createDataFrame with a list of tuples and a list of column names.

2

Create the product DataFrame

Create a Spark DataFrame called product_df with these exact rows: (101, "Pen"), (102, "Notebook"), (103, "Eraser"). The columns must be product_id and product_name.

Apache Spark

from pyspark.sql import SparkSession

spark = SparkSession.builder.appName("BroadcastJoinExample").getOrCreate()
sales_data = [(1, 101, 2), (2, 102, 1), (3, 103, 5)]
sales_columns = ["sale_id", "product_id", "quantity"]
sales_df = spark.createDataFrame(sales_data, sales_columns)
# Create product_df DataFrame with specified rows and columns
# Your code here

Hint

Use spark.createDataFrame with a list of tuples and a list of column names for products.

3

Perform the broadcast join

Import broadcast from pyspark.sql.functions. Use sales_df.join() to join sales_df with product_df on product_id. Use broadcast(product_df) inside the join to broadcast the small product table. Save the result as joined_df.

Apache Spark

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

spark = SparkSession.builder.appName("BroadcastJoinExample").getOrCreate()
sales_data = [(1, 101, 2), (2, 102, 1), (3, 103, 5)]
sales_columns = ["sale_id", "product_id", "quantity"]
sales_df = spark.createDataFrame(sales_data, sales_columns)
product_data = [(101, "Pen"), (102, "Notebook"), (103, "Eraser")]
product_columns = ["product_id", "product_name"]
product_df = spark.createDataFrame(product_data, product_columns)
# Join sales_df with broadcast(product_df) on product_id and save as joined_df
# Your code here

Hint

Use broadcast(product_df) inside the join() method.

4

Show the joined DataFrame

Use print() and joined_df.show() to display the joined DataFrame output.

Apache Spark

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

spark = SparkSession.builder.appName("BroadcastJoinExample").getOrCreate()
sales_data = [(1, 101, 2), (2, 102, 1), (3, 103, 5)]
sales_columns = ["sale_id", "product_id", "quantity"]
sales_df = spark.createDataFrame(sales_data, sales_columns)
product_data = [(101, "Pen"), (102, "Notebook"), (103, "Eraser")]
product_columns = ["product_id", "product_name"]
product_df = spark.createDataFrame(product_data, product_columns)
joined_df = sales_df.join(broadcast(product_df), on="product_id")
# Print the joined DataFrame
# Your code here

Hint

Use joined_df.show() to display the DataFrame rows.

Practice

1.

What is the main benefit of using broadcast() in Spark when joining tables?

easy

A. It compresses the large table to save memory.

B. It sorts both tables before joining.

C. It sends the small table to all worker nodes to speed up the join.

D. It duplicates the large table across all nodes.

Solution

1. Step 1: Understand broadcast join purpose

   Broadcast join sends the small table to all worker nodes to avoid shuffling the large table.

2. Step 2: Identify the benefit

   This reduces network data movement and speeds up the join operation.

3. Final Answer:

   It sends the small table to all worker nodes to speed up the join. -> Option C

4. Quick Check:

   Broadcast join = send small table to all nodes [OK]

Hint: Broadcast small table to all nodes to avoid shuffle [OK]

Common Mistakes:

• Thinking broadcast compresses large table
• Assuming broadcast sorts tables
• Believing broadcast duplicates large table

2.

Which of the following is the correct syntax to broadcast a small DataFrame small_df in Spark before joining?

from pyspark.sql.functions import broadcast

result = large_df.join(____, 'id')

easy

A. small_df.join(broadcast())

B. small_df.broadcast()

C. broadcast(large_df)

D. broadcast(small_df)

Solution

1. Step 1: Recall broadcast syntax

   In Spark, you use broadcast() function around the small DataFrame to mark it for broadcasting.

2. Step 2: Apply to join

   The join should be large_df.join(broadcast(small_df), 'id') to broadcast the small table.

3. Final Answer:

   broadcast(small_df) -> Option D

4. Quick Check:

   Use broadcast() function on small_df [OK]

Hint: Use broadcast() around small DataFrame in join [OK]

Common Mistakes:

• Calling broadcast as a method on DataFrame
• Broadcasting the large DataFrame instead
• Incorrect join syntax with broadcast

3.

Given the following Spark code, what will be the output count?

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

spark = SparkSession.builder.getOrCreate()

large_data = [(1, 'A'), (2, 'B'), (3, 'C')]
small_data = [(1, 'X'), (2, 'Y')]

large_df = spark.createDataFrame(large_data, ['id', 'val1'])
small_df = spark.createDataFrame(small_data, ['id', 'val2'])

joined_df = large_df.join(broadcast(small_df), 'id')
print(joined_df.count())

medium

A. 2

B. 5

C. 3

D. 0

Solution

1. Step 1: Analyze join keys and data

   large_df has ids 1, 2, 3; small_df has ids 1, 2. Join on 'id' keeps only matching ids.

2. Step 2: Count matching rows

   Ids 1 and 2 match in both tables, so joined_df has 2 rows.

3. Final Answer:

   2 -> Option A

4. Quick Check:

   Join count = matching ids = 2 [OK]

Hint: Count matching keys in both tables after join [OK]

Common Mistakes:

• Counting all rows from large_df ignoring join
• Assuming broadcast changes join result count
• Counting unmatched rows

4.

Identify the error in the following Spark code that attempts a broadcast join:

from pyspark.sql.functions import broadcast

result = large_df.join(small_df.broadcast(), 'id')

medium

A. Missing import for broadcast function.

B. Using broadcast as a method on DataFrame is incorrect.

C. Join key 'id' does not exist in both DataFrames.

D. Broadcasting large_df instead of small_df.

Solution

1. Step 1: Check broadcast usage

   Broadcast is a function, not a method on DataFrame, so small_df.broadcast() is invalid syntax.

2. Step 2: Correct usage

   Use broadcast(small_df) instead to mark small_df for broadcasting.

3. Final Answer:

   Using broadcast as a method on DataFrame is incorrect. -> Option B

4. Quick Check:

   broadcast() is a function, not a DataFrame method [OK]

Hint: Use broadcast() function, not DataFrame.broadcast() [OK]

Common Mistakes:

• Calling broadcast as a DataFrame method
• Ignoring import of broadcast function
• Confusing which DataFrame to broadcast

5.

You have a large DataFrame orders with millions of rows and a small DataFrame countries with 10 rows. You want to join them on country_code. Which approach will give the best performance?

hard

A. Use orders.join(broadcast(countries), 'country_code') to broadcast the small table.

B. Use orders.join(countries, 'country_code') without broadcasting.

C. Broadcast the large orders DataFrame instead.

D. Collect both DataFrames to driver and join locally.

Solution

1. Step 1: Identify table sizes

   Since countries is small, broadcasting it avoids shuffling the large orders DataFrame.

2. Step 2: Choose best join method

   Broadcast join sends the small table to all nodes, speeding up the join and reducing network traffic.

3. Final Answer:

   Use orders.join(broadcast(countries), 'country_code') to broadcast the small table. -> Option A

4. Quick Check:

   Broadcast small table for best join performance [OK]

Hint: Broadcast the small table in large-small joins [OK]

Common Mistakes:

• Broadcasting the large DataFrame wastes memory
• Not broadcasting causes expensive shuffle
• Collecting large data to driver causes failure