Understanding Distributed Joins

In F03.1 and F03.2, we learned how distributed systems use shuffles to bring records with the same key together for aggregation.

But aggregation isn’t the only operation that requires related records to meet. Another common operation is a join. Let’s understand this using our orders platform.

Calculating Order Delivery Time

The reporting team wants to calculate how long each order takes to reach the customer. For every delivered order, we need two events:

1. 01order_placed: When the customer placed the order.
2. 02order_delivered: When the order was delivered.

The delivery duration is calculated as:
Delivery time = Delivery timestamp − Order placement timestamp

However, these events are stored as separate records. They may also be distributed across different partitions and processed by different workers. To calculate delivery time, we need to match the placement and delivery events belonging to the same order. Both events contain a common field: order_id. This makes order_id the appropriate join key for our requirement.

What Should the Join Produce?

Our smaller dataset contains:

• 4,400 placed orders
• 3,977 delivered orders

For this exercise, every delivered order has exactly one matching placement event. Therefore, joining the two event sets using order_id should produce: 3,977 matching records, one for each delivered order. The simulator scales this workload to approximately:

• 12.5 million placed-order events
• 11.3 million delivered-order events

The correct join produces 11,280,136 records, representing delivered orders that have matching placement events.

What Went Wrong?

The reporting team runs the job, but the results are unexpectedly large. Instead of approximately 11.3 million records, the join produces: 475,518,578 records
The output grows from an expected 0.68 GB to 28.53 GB. One reduce task takes more than 334 seconds to finish, significantly delaying the entire job. What caused this? The developer joined the two event sets using customer_id instead of order_id. Both columns exist in the data, and the join executes successfully. However, they represent different relationships:

• order_id matches events belonging to the same order.
• customer_id matches events belonging to the same customer, including events from different orders.

A customer may have placed hundreds of orders, so joining by customer can produce many combinations that don’t represent valid order deliveries.

Why Does This Matter?

A join can be syntactically valid and execute successfully while producing logically incorrect results. Choosing the wrong key can cause:

• Incorrect output: Unrelated records are matched together.
• Unexpected data growth: The join produces far more rows than expected.
• Longer execution times: Workers must process and write additional records.
• Uneven workloads: High-volume keys generate significantly more matches than others.

In this lesson, we’ll learn how to predict the number of matching records, understand how distributed joins execute, and identify why some join strategies are more efficient than others.

Let’s understand how distributed systems match records from two datasets when those records are spread across different workers. We’ll begin with our delivery-time report.

Step 1: Understand How Matching Records Are Produced

Consider a customer who places multiple orders. For example, suppose cust-001 placed three orders, and two have been delivered.
Order placement events

Order delivery events

If we join using order_id, each delivered order matches its corresponding placement event. The result contains 2 records:

• ord-101 matches ord-101.
• ord-103 matches ord-103.

Order ord-102 has not been delivered, so it has no matching delivery event. Now imagine joining these datasets using customer_id instead. All five records have the same customer ID. Every placement record matches both delivery records, producing: 3 placed orders × 2 delivered orders = 6 matching pairs
Only two of those pairs represent the same order. The other four combine events belonging to different orders. This demonstrates a fundamental property of joins:
When a key appears multiple times on both sides, the join produces every matching combination of those records.

Step 2: Calculate the Number of Matching Pairs

For an inner join, the number of output rows for a particular key is:
Matching pairs = Left-side records for the key × Right-side records for the key

The total join output is the sum of these products across all matching keys. For example:

Now let’s apply this to our orders dataset. When joining by order_id, every delivered order has exactly one corresponding placement record. Therefore, every delivered order produces one matching pair.

Our nightly workload contains:

• Approximately 12.5 million placed-order events.
• Approximately 11.3 million delivered-order events.

Step 3: How Does the Join Execute Across Workers?

Now that we understand how matching pairs are produced, let’s examine how a distributed system performs the join. Our job uses a shuffle join. In this strategy, both inputs are redistributed using the same partitioning rule. For our delivery-time report, the key is order_id. The execution works as follows:

1. Map tasks read both inputs.
The first set of map tasks processes order-placement events. The second set processes order-delivery events. In our simulator, each side is processed using 12 map tasks.

2. Both inputs are partitioned by order_id.
The system uses the same hash-based rule for both inputs: Reduce partition = stable_hash(order_id) % 12
This ensures that placement and delivery events belonging to the same order are sent to the same reduce task.

3. The shuffle moves records to their destination tasks.
Because matching records may originate from different nodes, the shuffle transfers the necessary data across the cluster.

4. Reduce tasks perform the join.
Each reduce task receives the placement and delivery records for the order IDs it owns. In our simplified implementation, the reduce task builds an in-memory lookup table from the placement records. It then examines each delivery record and searches for the corresponding placement event. When the keys match, the task produces a joined record.

5. The final output is produced.
The reduce tasks collectively produce 11,280,136 matching records, one per delivered order.

Step 4: What Does the Join Cost?

A shuffle join introduces additional data movement because both inputs must be redistributed by their join keys. In our reference simulator:

• The map stage takes 47.9 seconds.
• The reduce stage takes 11.9 seconds.
• Approximately 4.79 GB moves across cluster nodes.

Including serial work, startup, and worker coordination, the total runtime is:
Total runtime = 10 + 40 + 36 + 47.9 + 11.9 = 145.8 seconds

The total run costs approximately 2.66 units in our illustrative pricing model.

What Should We Learn from This?

A shuffle join brings matching records together by distributing both inputs using the same join key. This makes it possible to match records that were originally stored or processed on different machines. However, the join’s performance depends on several factors:

• How many records must be redistributed.
• How frequently each join key appears.
• How many matching pairs each key produces.
• How evenly the resulting work is distributed across tasks.

In our delivery-time example, joining by order_id produces a predictable result because each delivered order matches exactly one placement event. But what happens when we change the join key?

We’ve seen that joining placement and delivery events by order_id produces one record for each delivered order. Now suppose a developer changes the join key from order_id to customer_id. The code still runs, and both inputs contain the customer_id column. However, the relationship has changed. Instead of matching events belonging to the same order, the join now matches every placement event with every delivery event belonging to the same customer.

Let’s examine the consequences.

Step 1: Calculate the Output for One Customer

Consider cust-0001, one of the most active customers in our dataset. This customer has:

• 263 order-placement events
• 236 order-delivery events

If we join by order_id, each delivered order matches its corresponding placement event. The expected result is: 236 matching pairs

But what happens if we join by customer_id? Every one of the 263 placement events matches all 236 delivery events. The number of matching pairs becomes: 263 × 236 = 62,068 pairs
Instead of producing 236 records, the join produces 62,068 records for this customer alone. Most of these records incorrectly associate a delivery with a different order placed by the same customer.

Our dataset contains the following records for cust-0001:

• 263 placed orders
• 236 delivered orders

Step 2: What Happens Across the Entire Dataset?

Now let’s compare the two join keys. Our smaller fixture produces:

• 3,977 matching pairs when joining by order_id.
• 167,652 matching pairs when joining by customer_id.

The customer-based join produces approximately 42 times as many rows. The same pattern appears in our scaled nightly workload.

Notice how changing a single join key affects both correctness and performance.

Step 3: Why Does One Reduce Task Become So Slow?

Our shuffle join partitions records using the join key. When the key is customer_id, all records belonging to the same customer are routed to the same reduce task. This includes customers who generate unusually large numbers of orders. In our scaled workload, cust-0001 alone produces approximately 176 million matching pairs.

That customer shares a reduce partition with other customers, creating a task responsible for approximately 280 million output pairs. This is significantly more work than other reduce tasks receive. As a result, the reduce stage takes 334.2 seconds, compared with just 11.9 seconds when joining by order_id. This is the data-skew problem we studied in F02, made much worse by the number of matching combinations generated by the incorrect join key.

Step 4: Why Didn’t the Shuffle Volume Increase Significantly?

Here’s an interesting observation. The amount of data redistributed across the cluster remains similar:

• Join by order_id: Approximately 4.79 GB moved.
• Join by customer_id: Approximately 4.76 GB moved.

Why doesn’t the shuffle become 42 times larger? Because the shuffle moves the input records, not all the matching pairs that will eventually be produced. The additional records are generated during the join itself, inside the reduce tasks. Therefore, changing the join key produces a massive increase in output size and processing work, even though the amount of shuffled input data remains nearly unchanged. This distinction is important when diagnosing distributed join performance. A join can become significantly slower because it produces too many matching pairs, even when network traffic stays relatively stable.

How Could We Detect This Before Running the Job?

We can estimate join output by examining how often each key appears in both inputs.
For each key: Expected matching pairs = Left record count × Right record count. Then add those products across all matching keys.

For example, cust-0001 alone produces: 263 × 236 = 62,068 pairs

Applying the same calculation to all customers reveals that the customer-based join produces far more rows than the delivery report requires. This is a useful validation technique because it can expose unexpected many-to-many relationships before running an expensive distributed join. For our delivery-time report, the expected output is one row per delivered order. If the proposed join produces many more records than that, we should investigate whether the chosen key represents the intended relationship.

In the previous section, we saw how choosing the wrong join key can produce incorrect results and significantly increase processing time. Now let’s consider another common challenge: choosing an efficient join strategy when one dataset is much smaller than the other. Our reporting team wants to add each customer’s region to their order events.

We have two datasets:

1. 01Order events: 60 million records, occupying approximately 16.19 GB.
2. 02Customers: 2.26 million records, occupying approximately 0.206 GB.

Every order event contains a customer_id, which we can use to look up the corresponding customer information. Unlike our previous example, joining by customer_id is correct here because the customers table contains only one row per customer. The team initially uses a shuffle join, which redistributes both datasets by customer_id. But is moving both datasets necessary when the customers table is much smaller?

Understanding the Two Join Strategies

Let’s compare two approaches.
1. Shuffle Join
A shuffle join redistributes both datasets using the same join key. For our example:

• Order events are partitioned by customer_id.
• Customer records are partitioned using the same rule.
• Records with matching customer IDs are sent to the same reduce task.
• Reduce tasks combine the matching records.

This works even when both datasets are large, but it can require substantial network traffic because records from both sides may need to move between nodes.

2. Broadcast Join
A broadcast join takes a different approach. Instead of redistributing both datasets, it sends a copy of the smaller customers table to every worker. Each worker then joins its assigned order events with the customer information available locally.

This means:

• The large orders dataset doesn’t need to be redistributed for the join.
• Each worker has access to the customers table.
• The join can be performed within the worker’s existing processing task.
• No separate reduce stage is needed in our simulator.

However, each worker must have enough memory to hold its copy of the customers table.

Our cluster has 12 workers running across 4 nodes.

We want to join:

• Order events: 16.19 GB
• Customers: 0.206 GB

Why Does the Shuffle Join Move More Data?

Our simulator assumes that approximately 75% of the shuffled data must cross between nodes. Since both datasets are redistributed,
the total shuffled input is: 16.19 GB + 0.206 GB = 16.396 GB
The estimated cross-node transfer is: 16.396 × 0.75 ≈ 12.29 GB

Most of this traffic comes from the much larger orders dataset.

Why Does the Broadcast Join Move Less Data?

With a broadcast join, only the customers table needs to be distributed. Our simulator sends one complete copy to each of the 12 workers.
Broadcast transfer = Customers table size × Number of workers = 0.206 GB × 12 ≈ 2.47 GB

The order events remain in their existing partitions. Each worker uses its local copy of the customers table to find the matching customer for every order event it processes. Because each customer has exactly one row in the customers table, the join produces 60 million matching records under the fixture’s assumptions, regardless of which strategy we choose.

Comparing the Results

Broadcasting the customers table:

• Reduces network transfer from 12.29 GB to 2.47 GB.
• Eliminates the separate reduce stage in our simulator.
• Reduces runtime from 270.9 to 218.7 seconds, saving approximately 52 seconds.
• Reduces cost from 5.78 to 2.81 units.

The broadcast join is more efficient for this particular workload because the smaller dataset can be copied to workers without requiring the much larger dataset to move. However, this doesn’t mean broadcast joins are always better. What happens when the smaller table grows too large to fit in memory? We’ll explore that next.

We’ve seen that broadcasting the customers table reduces network traffic and improves runtime. But before choosing a broadcast join, we need to answer two important questions:

1. 01Can every worker hold the broadcast table in memory?
2. 02How much data will broadcasting transfer as the number of workers increases?

These two factors determine whether broadcasting remains a practical choice.

1. Check Whether the Table Fits in Memory

A broadcast join requires each worker to have access to the complete smaller dataset. In our simulator, each worker builds an in-memory hash table from the broadcast data so it can quickly find matching customer records. However, the in-memory representation can be larger than the original data stored on disk.
For this lesson, our model assumes that the in-memory hash table requires four times the table’s stored size.
Our customers table occupies: 0.206 GB
Therefore, the estimated memory requirement per worker is: 0.206 × 4 = 0.824 GB
Each worker has 2 GiB of memory, approximately 2.15 GB.

Since 0.824 GB is less than 2.15 GB, the customers table fits in memory under our model’s assumptions. But what happens as the customers table grows?

Why Does This Matter?

A table that is small today may not remain small as the business grows. For example:

• A table containing country codes may remain relatively small.
• A customers table may grow as new customers register.
• A product catalogue may become significantly larger over time.

A broadcast strategy that works today may become unsuitable as its smaller input grows.

Important: Our model assumes the entire worker memory budget can be used for the broadcast table. Real processing engines also need memory for other tasks, data structures, and execution overhead. Therefore, the practical broadcast limit may be lower.

2. Understand How Worker Count Affects Broadcast Cost

Memory isn’t the only limitation. A broadcast join must distribute the smaller dataset to every worker. As we add workers, the number of copies increases.
In our model: Broadcast transfer = Small table size × Number of workers
The customers table is 0.206 GB, so:

• 4 workers receive approximately 0.82 GB in total.
• 12 workers receive approximately 2.47 GB.
• 24 workers receive approximately 4.94 GB.
• 48 workers receive approximately 9.89 GB.

The amount of data transferred increases with the worker count, even though the customers table itself hasn’t changed. A shuffle join behaves differently. In our four-node simulator, the cross-node shuffle transfer remains near 12.3 GB for this workload because the volume of input data stays approximately the same.

Let’s compare both strategies.

Comparing Broadcast and Shuffle Across Worker Counts

Notice two important patterns:
1. Broadcast transfer increases as workers are added.
Every additional worker needs another copy of the customers table. At smaller worker counts, this requires far less data movement than a shuffle. However, the difference becomes smaller as we add more workers.

2. More workers don’t always reduce runtime.
Both strategies initially become faster as we increase worker count, but eventually their runtimes start increasing again. This happens because additional workers introduce coordination and resource-management overhead, as we learned in F01. In our comparison, the broadcast join has its shortest observed runtime at 24 workers.

Understanding the Crossover Point

The crossover point is the worker count at which both strategies transfer approximately the same amount of data. For our simplified model:
Broadcast transfer = Table size × Worker count
Shuffle transfer ≈ 75% × Combined input size

We can estimate the crossover by setting these values equal:

0.206 × W = 0.75 × (16.19 + 0.206)

Solving for (W):

W = (0.75 × 16.396) ÷ 0.206 ≈ 59.7

Therefore, the estimated crossover is approximately 60 workers. Below this count, broadcasting the customers table transfers less data than shuffling both inputs in our model. Above it, broadcasting transfers more data.

Does This Mean Shuffle Is Always Better Beyond 60 Workers?

No. The crossover tells us when the two strategies transfer the same amount of data, not when they have identical runtimes or costs. For example, with 64 workers:

• Broadcast: Transfers 13.19 GB and completes in 270.7 seconds.
• Shuffle: Transfers 12.30 GB and completes in 294.5 seconds.

Broadcasting transfers slightly more data but still completes sooner because it avoids the separate reduce stage in our simulator. At 96 workers, broadcasting also becomes slightly more expensive overall:

• Broadcast cost: 33.68 units.
• Shuffle cost: 33.49 units.

This illustrates why we need to consider multiple factors when choosing a join strategy.

3. Choosing the Right Join Strategy

Let’s summarise the important considerations.

For our current workload, broadcasting the customers table is the better option among the tested strategies because it fits in memory and reduces both execution time and cost. However, we should re-evaluate that choice if:

• The customers table grows substantially.
• Worker memory becomes constrained.
• The number of workers increases.
• The join output becomes unexpectedly large.
• The cost of broadcasting approaches or exceeds the cost of shuffling.

Our experiments demonstrate three important lessons about distributed joins.

1. 01Join keys determine output cardinality: The number of matching rows depends on how often each key appears in both datasets. Choosing the wrong key can multiply output records and create severe performance problems.
2. 02Broadcast joins can avoid expensive shuffles: When one dataset is small enough to fit in worker memory, broadcasting it can reduce network transfers and eliminate the need to redistribute the larger dataset.
3. 03Join strategies have different resource limits: A broadcast join depends on per-worker memory and the cost of copying data to workers. A shuffle join depends on data movement, partitioning, and the work required to match records.

Limitations of the Model

Our simulator makes several simplifying assumptions.

• Proportional workload scaling: The model scales the smaller fixture to represent a much larger job. For example, cust-0001 produces 62,068 matching pairs in the fixture and approximately 176 million in the scaled workload. Real join cardinality can grow differently, particularly when records on both sides increase independently.
• Fixed data distribution: The model assumes the distribution of orders and customers remains similar to the reference dataset. Changes in customer activity or the arrival of new high-volume customers can affect join size and skew.
• Simplified broadcast memory usage: The model assumes a broadcast hash table uses four times its stored size in memory and that each worker has the full modelled memory budget available. Real memory overhead depends on the execution engine and data representation.
• One broadcast copy per worker: Our model distributes a separate copy of the smaller table to each worker. Some distributed engines may share broadcast data among tasks or processes running on the same node.
• Simplified join execution: The model doesn’t include sort-merge joins, disk spilling, or the full range of join algorithms used by production processing engines. We’ll explore these mechanisms in the Spark curriculum.
• No failures or retries: Every task completes successfully. We’ll study how distributed systems handle failed tasks, re-computation, and recovery in F04.

Why Join Cardinality Matters

Before optimising a distributed join, we must first confirm that it produces the correct relationships. For example, our delivery-time report should produce approximately 11.3 million matched orders. Joining on customer_id instead of order_id produces 475.5 million records, even though the code executes successfully. No join optimisation can make those incorrect matches valid. Similarly, broadcasting the wrong dataset or choosing a strategy without checking memory requirements can turn an otherwise correct join into an expensive or failing operation.

We’ve learned how join keys, data sizes, and worker memory affect distributed join performance. Let’s examine two more situations where choosing the wrong approach can lead to unnecessary data movement, memory failures, or higher processing costs.

Mistake 1: Broadcasting the Larger Dataset

In the previous section, we saw that broadcasting the smaller customers table improved performance because each worker could hold a copy in memory. But what happens if we broadcast the larger orders dataset instead?

Let’s recall our two datasets:

• Order events: 16.19 GB
• Customers: 0.206 GB

A developer accidentally configures the join to broadcast the order events rather than the customers table. Since a broadcast join sends the entire selected dataset to every worker, each worker must now hold all 16.19 GB of order events in memory. Our simulator assumes that the in-memory representation requires four times the stored data size.

The team broadcasts the 16.19 GB orders dataset instead of the customers table. Each worker has approximately 2.15 GB of memory, and our model assumes that a broadcast hash table requires four times its stored size.

Mistake 2: Joining Before Aggregating

Now let’s revisit the customer-region report from Section 04. The reporting team wants to calculate total event counts per customer, along with each customer’s region. One approach is to join all 60 million events with the customers table, attach the region to every event, and then aggregate the results. But do we really need to attach customer information to every individual event?
Consider an alternative:

1. 01Aggregate the 60 million events to produce one result per customer.
2. 02Join those aggregated results with the customers table to retrieve each customer’s region.

This changes how much data the join needs to process.

Why Does Aggregating First Help?

Our scaled workload contains:

• 60 million order events
• 2,260,565 customers

If we join before aggregating, the join processes all 60 million events. If we aggregate first, the join processes only 2,260,565 customer-level results. That is a substantial reduction in the number of rows participating in the join. Let’s compare the join strategies after aggregation.

These are the simulator’s modeled join-step results after customer-level aggregation, not the complete pipeline runtime including the earlier aggregation.

Notice two important things.
1. Aggregating first reduces the data involved in the join.
The join now operates on customer-level totals rather than individual events. This can significantly reduce the amount of data that needs to be processed and transferred.

2. The fastest join isn’t necessarily the cheapest.
After aggregation:

• The broadcast join completes in 92.8 seconds, making it slightly faster.
• The shuffle join completes in 97.8 seconds, but costs less and transfers significantly less data.

The difference is only five seconds, but the shuffle join is cheaper in our model.

When Is This Optimization Valid?

Aggregating before joining works when the business requirement allows it. In our example, each customer has one region, and the report needs one total per customer. We can calculate the totals first and attach the region afterward. However, if customer attributes vary per event or historical versions must be preserved, joining after aggregation may change the meaning of the result.

How can we recognise when a distributed join is producing incorrect results or consuming unnecessary resources? In our simulator, look for four warning signs:

1. 01Unexpectedly large output: The join produces far more records than expected, suggesting an incorrect join key or an unintended many-to-many relationship.
2. 02Overloaded join tasks: One or a few tasks process significantly more matching pairs than the others.
3. 03Excessive data movement: Large amounts of data are shuffled or broadcast across the cluster.
4. 04Memory failures: A broadcast join fails because the selected dataset is too large to fit in worker memory.

What Does This Look Like in Production?

Similar problems may appear as:

• A join produces many times more rows than expected.
• One join task continues running long after the other tasks finish.
• Shuffle read and write volumes are unexpectedly high.
• Workers run out of memory when attempting to broadcast a large dataset.
• A previously successful broadcast join starts failing as the smaller table grows.
• Increasing the number of workers raises network traffic without meaningful performance improvements.

For example, our delivery-time report produced 475.5 million records instead of 11.3 million because it joined on customer_id rather than order_id. The incorrect join also created an overloaded reduce task that took 334.2 seconds to finish.

What Should You Investigate?

When a join behaves unexpectedly, start by examining:

• Join keys: Do they represent the relationship the business needs?
• Key cardinality: How many records share each key on both sides?
• Output size: Does the number of matching records agree with the expected relationship?
• Data movement: How much data is shuffled or broadcast?
• Task performance: Are a few keys responsible for most of the join work?
• Memory usage: Does the broadcast table fit comfortably within the available worker memory?

These measurements help distinguish correctness problems from execution bottlenecks.

Before running a large distributed join, consider these five questions:

1. 01Are we joining on the correct key? Verify that the join columns represent the intended relationship between the datasets.
2. 02How many matching records will the join produce? Examine key frequencies on both sides and estimate the output using the number of matching pairs.
3. 03Can the smaller dataset fit in worker memory? Estimate the memory required for a broadcast hash table, including its in-memory expansion and other execution overhead.
4. 04Which strategy moves less data? Compare the cost of broadcasting the smaller dataset with redistributing both inputs through a shuffle.
5. 05Can we reduce the input before joining? Consider whether filtering or aggregating records first would reduce the amount of data involved in the join without changing the expected result.

For example, our customer-region report originally joined 60 million events with the customers table. Aggregating the events first reduced the join input to approximately 2.26 million customer-level results. This changed both the amount of data transferred and the relative cost of broadcast and shuffle joins.

Let’s return to our customer-region join. Previously, we compared broadcast and shuffle joins on a cluster with 4 nodes. Our model assumes that roughly 75% of shuffled data crosses between nodes. Now suppose the cluster has 8 nodes instead of 4. We continue using a placement model where the source and destination nodes of shuffled records are evenly distributed. What happens to the amount of shuffled data that must cross the network?

How Does This Affect the Broadcast Crossover?

Remember that our crossover point is where the broadcast and shuffle strategies transfer approximately the same amount of data.
With eight nodes, the estimated shuffle transfer is: Shuffle transfer = 0.875 × (16.19 + 0.206) ≈ 14.35 GB

The customers table is still 0.206 GB. Therefore: Crossover workers = 14.35 ÷ 0.206 ≈ 69.6

The crossover moves from approximately 60 workers to 70 workers. This happens because the shuffle transfers a larger share of its input across nodes, while our broadcast model still sends one copy of the customers table to every worker.

Try It Yourself

When the F03 workbench is available, run the delivery-time join using the correct key:

The reference result should show:

• Matching pairs: 11,280,136
• Output size: 0.68 GB
• Data transferred: 4.79 GB
• Total runtime: 145.8 seconds

Now change the join key from order_id to customer_id. The incorrect join produces:

• Matching pairs: 475,518,578
• Output size: 28.53 GB
• Reduce-stage duration: 334.2 seconds

Compare the results and identify why changing the join key causes such a large increase in output records.

Note: The F03 workbench is planned. These commands are intended for use once it becomes available.

Distributed joins combine records from different datasets using shared keys. When those datasets are distributed across multiple workers, the system must bring matching records together.

In this lesson, we learned that:

1. 01Join keys determine correctness and output size: An incorrect key can create unintended many-to-many matches and multiply output records.
2. 02Shuffle joins redistribute data: Both inputs may need to move across workers so matching records can be processed together.
3. 03Broadcast joins replicate the smaller dataset: This can avoid shuffling the larger dataset, provided the smaller input fits in worker memory.
4. 04Broadcasting has limits: Memory requirements and transfer costs increase as the broadcast dataset or worker count grows.
5. 05Reducing data before joining can improve efficiency: Filtering or aggregating first can reduce the number of records participating in the join.

Our Delivery-Time Example

Compare the results of joining placement and delivery events.

The incorrect join key produced approximately 42 times more records and created a severe processing bottleneck.

Our Customer-Region Example

When joining 60 million events with the customers table:

• Shuffle join: 12.29 GB transferred, 270.9 seconds.
• Broadcast join: 2.47 GB transferred, 218.7 seconds.

Broadcasting the smaller customers table was more efficient for this workload. However, its memory requirement and network usage must be reconsidered as the table or worker count grows.