Broadcast joins for small tables in Apache Spark - Step-by-Step Execution

Concept Flow - Broadcast joins for small tables

Start with two tables

↓

Identify small table

↓

Broadcast small table to all nodes

↓

Perform join locally on each node

↓

Combine results from all nodes

↓

Output joined table

Broadcast join sends the small table to all worker nodes to join locally with the big table, speeding up the join.

Execution Sample

Apache Spark

from pyspark.sql import SparkSession
from pyspark.sql.functions import broadcast
spark = SparkSession.builder.getOrCreate()

big_df = spark.range(1000000).withColumnRenamed("id", "key")
small_df = spark.createDataFrame([(1, "A"), (2, "B")], ["key", "value"])

joined_df = big_df.join(broadcast(small_df), "key")
joined_df.show(5)

This code joins a big table with a small table using broadcast join to speed up the process.

Execution Table

Step	Action	Details	Result
1	Create big_df	Generate 1,000,000 rows with 'key' column	big_df with keys 0 to 999,999
2	Create small_df	Create small table with 2 rows (keys 1 and 2)	small_df with keys 1 and 2
3	Broadcast small_df	Send small_df to all worker nodes	small_df available locally on each node
4	Join big_df and small_df	Join on 'key' column locally on each node	Rows with keys 1 and 2 matched with values A and B
5	Collect results	Combine joined rows from all nodes	Joined DataFrame with matching rows
6	Show output	Display first 5 rows	Rows with keys 1 and 2 and their values shown
7	Exit	Join complete	Process ends

💡 Join completes after all matching rows are combined and displayed.

Variable Tracker

Variable	Start	After Step 1	After Step 2	After Step 3	After Step 4	After Step 5	Final
big_df	None	1,000,000 rows with keys 0-999,999	1,000,000 rows	1,000,000 rows	1,000,000 rows	1,000,000 rows	1,000,000 rows
small_df	None	None	2 rows with keys 1 and 2	Broadcasted to all nodes	Broadcasted	Broadcasted	Broadcasted
joined_df	None	None	None	None	Rows with keys 1 and 2 joined	Joined rows collected	Joined DataFrame with matched rows

Key Moments - 3 Insights

Why do we broadcast the small table instead of the big table?

Broadcasting the small table is efficient because it fits in memory and sending the big table would be too large and slow. See execution_table step 3 where only small_df is broadcasted.

What happens if the small table is not broadcasted?

Without broadcasting, Spark shuffles data across the network, which is slower. Broadcasting avoids this by sending the small table to all nodes upfront (execution_table step 3 vs step 4).

How does broadcasting improve join speed?

Broadcasting allows each node to join locally without waiting for data from others, reducing network traffic and speeding up the join (execution_table steps 3 and 4).

Visual Quiz - 3 Questions

Look at the execution_table, at which step is the small table sent to all worker nodes?

A Step 3

B Step 4

C Step 2

D Step 5

💡 Hint

Check the 'Action' column for broadcasting in execution_table.

According to variable_tracker, what is the state of joined_df after step 4?

A None

B Rows with keys 1 and 2 joined

C 1,000,000 rows

D Broadcasted small_df

💡 Hint

Look at joined_df row under 'After Step 4' in variable_tracker.

If the small table had 1 million rows, what would likely happen to the broadcast step?

A Big table would be broadcasted instead

B Broadcast would still be efficient

C Broadcast would fail or be slow due to size

D Join would happen without broadcast

💡 Hint

Recall why small tables are broadcasted from key_moments and execution_table step 3.

Concept Snapshot

Broadcast joins send a small table to all worker nodes.
This avoids shuffling the big table across the network.
Use broadcast() on the small DataFrame in Spark.
Join happens locally on each node.
Speeds up joins when one table is much smaller.

Full Transcript

Broadcast joins in Apache Spark work by sending the small table to all worker nodes. This lets each node join the small table locally with the big table, avoiding slow network shuffles. The process starts by creating the big and small tables. Then the small table is broadcasted to all nodes. Each node performs the join locally. Finally, results are combined and shown. Broadcasting is efficient only if the small table fits in memory. This method speeds up joins when one table is much smaller than the other.

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