Broadcast and Skew Joins

Spark tutorial · PySpark.in

Broadcast and Skew Joins

1. Broadcast Join

In distributed computing, a Broadcast Join is a technique where the smaller dataset is sent to all worker nodes, so that a join can be performed without shuffling the larger dataset. In Spark, broadcast joins can dramatically speed up join operations by avoiding the expensive shuffle of a large DataFrame across the cluster. PySpark can automatically decide to broadcast a small DataFrame (based on a size threshold, e.g. tens of MB), but you can also hint or explicitly mark a DataFrame to be broadcast.

How Broadcast Joins Work

Normally, a join between two big tables is a shuffle join (also called a sort-merge join in Spark) – both DataFrames are shuffled and partitioned by the join key, so that matching keys end up on the same partition. Shuffle joins are expensive when data is large.

With a broadcast join, Spark will distribute the entire smaller DataFrame to every partition of the larger DataFrame. Then each partition of the large DataFrame can join with the local copy of the small DataFrame without further network shuffle. This is very effective when one DataFrame is relatively tiny and the other is huge.

When to use: A common guideline is to broadcast the smaller side if it’s on the order of tens of megabytes or less (say < 100MB), though the exact threshold can be tuned and depends on your cluster memory. Spark’s config spark.sql.autoBroadcastJoinThreshold (default 10MB in many versions) controls automatic broadcasting. You can override or force it with hints.

Example Usage

In our running example, the Departments DataFrame is very small (only 3 records). In a real scenario, imagine df_departments is a small dimension table and df_employees is huge (millions of rows). A broadcast join would be beneficial. We can force a broadcast join in PySpark like this:

```

from pyspark.sql.functions import broadcast

from pyspark.sql import SparkSession

spark = SparkSession.builder.appName("JoinsTutorial").getOrCreate()

# Create Employees DataFrame

employees = [

(1, "Alice", 101),

(2, "Bob", 102),

(3, "Charlie", None), # Charlie has no dept (null)

(4, "David", 104), # David's dept_id 104 doesn't exist in Departments

(5, "Eve", 101)

]

df_employees = spark.createDataFrame(employees, ["emp_id", "name", "dept_id"])

df_employees.show()

# Create Departments DataFrame

departments = [

(101, "HR"),

(102, "IT"),

(103, "Marketing")

]

df_departments = spark.createDataFrame(departments, ["dept_id", "dept_name"])

broadcast_join_df = df_employees.join(broadcast(df_departments), on="dept_id", how="inner")

broadcast_join_df.show()

```

This code explicitly marks df_departments to be broadcast to all nodes. The result of broadcast_join_df will be the same as an inner join we did before (in fact, we still specify the join type as "inner"). Broadcasting does not change the result of the join, only the way Spark executes it. In our small example, the output is identical to the earlier inner join result:

But behind the scenes, Spark did not shuffle the employees by dept_id. Instead, it broadcast the tiny departments DataFrame to every executor, and each executor joined its portion of employees with that broadcasted data. This is much faster when df_employees is large, because shuffling a huge DataFrame is avoided.

Alternative: Join Hints

PySpark also allows join hints. For example:

```

hide

df_employees.hint("broadcast").join(df_departments, "dept_id")

```

would hint to Spark to broadcast the employees DataFrame (not what we want in this case, we’d rather broadcast the smaller one). You can also hint both or neither, but Spark will decide which side to broadcast or neither. The broadcast() function from pyspark.sql.functions is more explicit and guarantees broadcasting the specified DataFrame.

Use Cases

Small dimension table with a large fact table: e.g., joining a small lookup table (country codes, product taxonomy, etc.) with a massive log of events.
Star schema optimization: In data warehouse style joins (fact table to multiple dimension tables), broadcasting each dimension (if small) will greatly speed up joins.
When the join key has low cardinality but one side is small: If one side has only a few distinct keys and is small, broadcasting can avoid skew issues by not shuffling by key.

Pitfalls

Memory Overhead: Broadcasting a table means making copies of it on each worker. If the table is larger than expected, you could run into memory issues or even crash executors. Avoid broadcasting DataFrames that are too big to comfortably fit in each executor’s memory.
Broadcast timeout: Very large broadcasts might take time and Spark has a timeout for broadcast (config spark.sql.broadcastTimeout). If a broadcast is slow (or too large), the job can fail.
Unnecessary broadcasting: If both tables are large, forcing a broadcast can backfire. Let’s say you mistakenly broadcast a 1GB DataFrame to join with a 10GB one – that means every executor tries to get a 1GB copy, which is inefficient. In such cases, a shuffle join might have been better. It's best to broadcast only when you're confident one side is small.
Skew Considerations: Broadcast joins can alleviate some skew problems because the large side is not shuffled by key at all (so a skewed key won’t concentrate all data on one partition in the shuffle). However, if the large DataFrame is itself heavily skewed in how data is originally partitioned (e.g., all records already in one partition due to previous transformations), broadcast won’t fix that; you might need to repartition the large DataFrame properly.

Best Practices

Let Spark Decide (Usually): Spark’s automatic broadcast mechanism is usually effective. It will broadcast a DataFrame if it estimates it to be under the threshold. Use df.explain() to see if a join is broadcast or not.
Use Hints for Control: When you know better – e.g., you have a slightly larger table that you know is used often and fits in memory – you can hint Spark to broadcast it. Or if Spark is broadcasting something inappropriately, you can disable it by increasing the threshold to a huge value or using hint("SHUFFLE_HASH") to force a shuffle join.
Combine with Partitioning: If you cannot broadcast (both sides large), consider other techniques (next section on skew handling). But if one side is borderline, sometimes filtering it down or caching it can make broadcasting feasible.
Monitor Memory: Keep an eye on Spark executors’ memory usage when broadcasting. If you see OOM errors or large GC pauses, you might be broadcasting too much data.

2. Skew Join Handling

Data skew in joins occurs when certain join key values are very common (disproportionately large number of records), causing an imbalance in the data distribution. For example, imagine joining on country_id and 50% of your data has country_id = USA – then in a normal shuffle join, one partition might end up handling 50% of the data (all the USA records land in the same partition for matching), becoming a performance bottleneck. Skew can lead to some tasks taking much longer than others, or even running out of memory, while other tasks finish quickly.

Spark has a few strategies to handle skewed joins:

2.1 Identify and Manage Skew

Detecting skew: A first step is to identify if skew exists. You can do this by computing group counts on the join key (e.g., df.groupBy("key").count() and looking at the distribution) or by looking at the Spark UI’s task metrics (if one reducer task is dramatically larger). Some keys having significantly more records than others is a sign of skew.

Basic mitigation strategies:

Broadcast the smaller table (if applicable): If one side of the join is small enough to broadcast, this often sidesteps the skew problem entirely. In a broadcast join, the large table is not shuffled by key, so a skewed key does not concentrate data into one partition; the data stays distributed as it was originally. For example, if joining a huge "sales" table with a small "countries" table, broadcasting "countries" avoids shuffling "sales" by country_id, thus avoiding one massive "USA" partition in the shuffle.
Increase parallelism: Sometimes just increasing spark.sql.shuffle.partitions can help distribute a skewed key’s data across more partitions (though the key’s data will still all go to one partition in a hash shuffle, Spark’s adaptive execution can sometimes split skewed partitions in later versions).

2.2 Salting Technique

When broadcasting isn’t possible (e.g., both tables are large) and skew is severe, a common technique is salting. Salting means adding a random "salt" value to the join key to artificially distribute what would have been one large group into several smaller groups. Both DataFrames must be salted in a coordinated way so that matches can still happen.

How salting works:

Augment keys with a random number: For the skewed key values, append or prepend a random integer (from 0 to N-1) to create a modified key. N could be based on how many splits you want for the largest skewed keys.
Replicate the other side’s data for those keys: The table on the other side needs to have matching salted keys. One way is to do a cross join between the small set of salt values and the portion of the data with the skewed keys, effectively duplicating the skewed key’s rows across N different keys (key_0, key_1, ... key_N).
Join on the salted key: Perform the join using the new salted key. The formerly single heavy key is now treated as N distinct keys, which go to N different partitions, relieving the single-partition bottleneck.
Post-processing: After the join, you may need to drop the salt and aggregate results back on the original key (if you want to get rid of the artificial distinction). For example, if you were joining and then aggregating, you can aggregate by the original key after the join.

Example scenario: Suppose dept_id 101 was extremely common in both DataFrames. We could salt it by, say, splitting into 3 keys: treat them as 101_0, 101_1, 101_2. For each employee with dept 101, assign a random salt 0-2 and form a composite key like 101_0. For each department 101 entry (there’s just one in our case), replicate it 3 times as 101_0, 101_1, 101_2. Then join on this composite key. Now the load of dept 101 is spread across 3 partitions. This is a simplistic illustration — in practice you'd only salt the problematic keys and choose the number of salts based on how skewed it is.

Pitfalls of salting:

It complicates your pipeline: you have to add extra columns and later possibly remove them.
It may over-distribute: You might introduce overhead for keys that weren’t actually problematic, or create too many partitions for the skewed key (some partitions could still be small, resulting in some inefficiency but usually not as bad as one huge one).
It requires knowing or identifying the skewed keys in advance. Sometimes you might not know which key is skewed without analysis.
If the right side of the join is huge and also skewed, replicating its skewed records (via cross join with salts) can increase data size proportionally to the number of salts, which might be expensive.

Despite these pitfalls, salting is an effective last resort when you cannot avoid a skewed shuffle. It ensures the job makes progress (avoiding one straggler task or out-of-memory error) by distributing the work more evenly, even if some tasks are still a bit heavier than others.

2.3 Other Approaches and Best Practices for Skew

Partition Pruning / Filtering: Sometimes you can reduce skew by pre-filtering extremely common keys if they are not needed or handling them separately. For instance, if 50% of data is a particular key that you can process in a special-case manner (outside the main join), that might be easier.
Skew Join Optimization in Spark: Modern Spark (3.x with Adaptive Query Execution) has some built-in skew join handling. Spark can detect at runtime if a partition is skewed (much larger than others) and split it automatically into multiple tasks. Ensure you have adaptive execution enabled (spark.sql.adaptive.enabled=true) so the engine can help with skew mitigation.
Combine with Aggregation: If the reason you’re joining is to then aggregate (say, joining a big fact table with a dimension to filter or group by something from the dimension), consider aggregating first. For example, instead of joining and then grouping, maybe group the fact table by the join key first to reduce its size, then join the smaller aggregated result with the other table.
Monitoring: Keep an eye on the Spark UI for uneven task times or data sizes. Skew often manifests as one task running much longer. If you see that, investigate the data distribution on the join keys.

Summary: Handling skew often requires a combination of techniques. The key is to redistribute data more evenly. Broadcasting small tables, salting large keys, and using Spark's adaptive optimizations are all tools to address the challenge of skewed joins. By applying these, you can prevent a single hot partition from dragging down your job.

Broadcast and Skew Joins