Advanced Join Techniques in Spark SQL#

Why Worry About Advanced Joins?#

While the syntax for a basic join may look simple, real-world scenarios frequently involve many complexities:

• Large fact tables joined to multiple dimension tables.
• Data skew caused by non-uniform distribution of key values.
• Potential duplication of data if partitioning and bucketing strategies are suboptimal.
• Highly selective joins requiring caution to avoid scanning massive amounts of data unnecessarily.

By mastering advanced join techniques, you can ensure that your Spark SQL pipelines are more stable, faster, and make better use of cluster resources.

Join Types Overview Table#

Maintaining a clear understanding of these join types is critical as you optimize queries; different scenarios might require different join semantics.

Example 1: Inner Join#

1
SELECT
2
  o.orderID,
3
  o.customerID,
4
  c.name,
5
  c.country
6
FROM orders o
7
JOIN customers c
8
ON o.customerID = c.customerID

This query returns all the rows for which there is a match in both orders and customers.

Example 2: Left Outer Join#

1
SELECT
2
  o.orderID,
3
  o.customerID,
4
  c.name,
5
  c.country
6
FROM orders o
7
LEFT JOIN customers c
8
ON o.customerID = c.customerID

This query returns all rows from orders plus the matching rows from customers. If a row in orders does not have a matching customerID in customers, the columns from customers will be NULL.

Example 3: Full Outer Join#

1
SELECT
2
  o.orderID,
3
  o.customerID,
4
  c.name,
5
  c.country
6
FROM orders o
7
FULL JOIN customers c
8
ON o.customerID = c.customerID

A full outer join includes all results from both tables, matching rows where possible.

Viewing the Execution Plan#

You can view the execution plan (aka the “explain plan”) by running:

1
df.explain(true)

Or in Spark SQL:

1
EXPLAIN EXTENDED
2
SELECT ...

This will give detailed insights into how Spark decides to handle the join. Whether the optimizer will pick a broadcast join or a shuffle-based join usually depends on the relative sizes of the tables and your Spark configuration settings (spark.sql.autoBroadcastJoinThreshold being a key parameter).

How Broadcast Joins Work#

In a broadcast join, the smaller dataset is “broadcast” across all worker nodes. This allows each executor to have a local copy of the smaller dataset, eliminating the need for a shuffle to match keys across partitions. The process looks like this:

1. Spark identifies that one side of the join is small (by default, smaller than spark.sql.autoBroadcastJoinThreshold which is often set around 10MB).
2. Spark collects that smaller RDD/DataFrame and “broadcasts” (ships) it to every executor node.
3. The larger dataset is scanned in parallel, leveraging local copies of the smaller dataset to perform join operations. This avoids shuffle overhead.

When to Use Broadcast Joins#

• You have one smaller dimension table (often called a lookup table) and a much larger fact table.
• The smaller table can fit entirely in memory on each worker node.
• You want to minimize the overhead of shuffling a large dataset on a join key.

Example: Forcing a Broadcast Join#

In some cases, Spark may not automatically broadcast your smaller DataFrame because it isn’t confident that the dataset is below the threshold or because you have a different threshold set. You can force a broadcast in Spark SQL and DataFrames:

1
SELECT /*+ BROADCAST(c) */
2
  o.orderID,
3
  o.customerID,
4
  c.name,
5
  c.country
6
FROM orders o
7
JOIN customers c
8
ON o.customerID = c.customerID

1
import org.apache.spark.sql.functions.broadcast
2

3
val joinedDF = orders.join(broadcast(customers), Seq("customerID"))

Memory Considerations#

When broadcasting a table, Spark needs enough memory to hold that table in memory on each worker node. If the table is too large, you can run into out-of-memory errors. Hence, be mindful of the cluster resources.

SortMergeJoin#

With SortMergeJoin, Spark sorts each side of the data by the join key. Then, it merges the sorted rows without scanning all combinations (like a nested loop). This is typically more efficient for larger datasets, but it does require sorting, which can be expensive.

ShuffledHashJoin#

The classic ShuffledHashJoin builds a hash table on one of the shuffled RDDs and then streams the other RDD to find matching keys. While faster in some specific circumstances, SortMergeJoin is often preferred for large datasets because it scales more gracefully and can handle partial data that doesn’t fit in memory by utilizing external sort.

Example: Shuffle-Based Join#

1
SELECT
2
  o.orderID,
3
  c.name
4
FROM orders o
5
JOIN customers c
6
ON o.customerID = c.customerID

Internally, Spark might decide on a SortMergeJoin. If you check the explain plan, you might see lines like:

1
*(2) SortMergeJoin [customerID], [customerID], Inner
2
:- *(2) Sort [customerID ASC NULLS FIRST], false, 0
3
:  +- Exchange hashpartitioning(customerID, 200)
4
+- *(2) Sort [customerID ASC NULLS FIRST], false, 0
5
   +- Exchange hashpartitioning(customerID, 200)

The Exchange hashpartitioning(customerID, 200) indicates a shuffle of the data based on customerID. Then, the SortMergeJoin operator merges them.

Symptoms of Data Skew#

• Some tasks take much longer than others to complete.
• You see bottlenecks in a few tasks while others finish quickly.
• Certain stages in the Spark UI show extremely large data sizes for a few partitions.

Techniques to Mitigate Skew#

1. Salting (Key-Salting)

   • Introduce a random value or partition number into the join key.
   • This random distribution spreads out the workload for commonly occurring keys.

2. Map-Side Aggregation

   • If possible, aggregate your data before the join to reduce the volume.

3. Broadcast Skewed Data

   • If the skewed side is relatively small, consider broadcasting it.

4. Skew Join Hints (Spark-specific hints)

   • Spark has certain strategies you can leverage to handle skew in advanced ways via hints.

1
import org.apache.spark.sql.functions.{lit, rand}
2

3
// Add a salt column with random integer from 0 to 9
4
val saltedCustomers = customers
5
  .withColumn("salt", (rand() * 10).cast("int"))
6

7
// For orders, replicate across possible salt values
8
val saltedOrders = orders
9
  .crossJoin(
10
    spark.range(0, 10).toDF("salt").alias("s")
11
  )
12
  .withColumn("customerID_salted", $"customerID")
13

14
// Then join on both the original key and the salt
15
val joined = saltedOrders.join(
16
  saltedCustomers,
17
  Seq("customerID_salted", "salt")
18
)

Default Partitioning#

By default, Spark partitions data (DataFrames or RDDs) using a hash of the partition key across spark.sql.shuffle.partitions (default is often 200). This leads to uniform distribution most of the time, but you might have to tweak it for large clusters or for skew.

Bucketing#

Bucketing is an advanced technique often used in combination with partitioning in Spark SQL or Hive. By bucketing a table on a certain column, you ensure that the same bucket contains rows with the same column hash. For joins on that same column, Spark can restrict data to matching buckets, dramatically reducing the amount of data shuffled.

1
CREATE TABLE customers_bucketed (
2
  customerID INT,
3
  name STRING,
4
  country STRING
5
)
6
CLUSTERED BY (customerID)
7
INTO 8 BUCKETS
8
STORED AS PARQUET;

Star Schema Joins#

Common in data warehousing, the star schema has one large fact table joined to multiple small dimension tables. In Spark:

• The dimension tables can often be broadcast.
• The fact table typically is too large for broadcast, so it uses a shuffle-based join when joining with large dimension tables.
• If all the dimension tables are small enough, you can broadcast each dimension, drastically reducing shuffle overhead.

Multi-Way Joins#

Sometimes a query requires joining more than two tables, e.g., a fact table (orders) with two dimension tables (customers, products). Spark can handle multi-way joins under the hood by forming a logical plan that joins them pairwise. The Catalyst optimizer considers multiple broadcast possibilities. If multiple dimension tables are each below the broadcast threshold, Spark can decide which ones to broadcast and in what order the join is best implemented.

Z-Ordering (in Databricks and advanced distributions)#

If you are on Databricks or a distribution that supports Z-Order clustering, you can reorder the data in storage to optimize for certain columns. This can help reduce the scan size when filtering and joining on those columns. Although not a direct join optimization technique, it can complement your Spark joins by making relevant data physically closer on disk.

Using Join Hints#

Spark supports hints in SQL to influence the optimizer’s choice of join strategies. For instance, you can manually specify whether you want a SHUFFLE_MERGE join, SHUFFLE_HASH join, or to broadcast one side:

1
SELECT /*+ SHUFFLE_MERGE(orders) SHUFFLE_HASH(customers) */
2
  ...

Be careful when using hints; if done incorrectly, they can degrade performance. They work best when you have detailed knowledge of your data distribution.