Joins in Spark: Spark & Data Eng lessons from production

Joins in Spark: Spark & Data Eng lessons from production

When you're joining datasets in production, the difference between a job that completes in seconds and one that times out often comes down to understanding how Spark executes joins. We've seen this play out across e-commerce pipelines, CRM systems, and real-time analytics platforms—a poorly chosen join strategy can cascade into cascading failures across the entire batch window. This article walks through join types and strategies we've learned to rely on, with practical patterns to avoid the duplicates and performance traps that catch teams in production.

Join types: inner, left, right, full, semi and anti

Spark supports six join types, each suited to different business logic. Let's ground these in a concrete scenario: you're matching orders (left table) with customers (right table) to enrich order records.

Inner join keeps only rows where keys match in both tables. If a customer never placed an order or an order has no matching customer, it's dropped. This is your strictest join and often the fastest because Spark can drop unmatched rows early.

orders_df.join(
    customers_df,
    orders_df.customer_id == customers_df.id,
    "inner"
)

Left join (left outer) keeps all rows from the left table. Unmatched rows from the right table become NULL. Use this when you never want to lose orders—even if a customer record isn't found, the order row survives with NULLs in customer columns.

orders_df.join(
    customers_df,
    orders_df.customer_id == customers_df.id,
    "left"
)

Right join keeps all rows from the right table. This is less common in practice but useful if you're validating that every customer in the reference table has at least one associated order. Mentally, flip the tables and think of it as a left join from the right side.

Full outer join keeps all rows from both tables, filling in NULLs on both sides where keys don't match. This is expensive—Spark must track both matched and unmatched rows—but invaluable for reconciliation jobs where you need to see what's in each table that isn't in the other.

Semi join returns only rows from the left table where the join key exists in the right table. Critically, it does not include any columns from the right table—just the left table rows. This is a filtering operation, not an enrichment. Use it to find "which orders have a customer ID in our active customers table."

orders_df.join(
    customers_df,
    orders_df.customer_id == customers_df.id,
    "leftsemi"
)

Anti join does the opposite: returns rows from the left table where the join key does not exist in the right table. This is how you find missing references—orders with no corresponding customer record. We'll expand on this later because anti joins are powerful for data quality checks.

orders_df.join(
    customers_df,
    orders_df.customer_id == customers_df.id,
    "leftanti"
)

Strategies: Broadcast, sort-merge and shuffle hash

The join type tells Spark which rows to keep. The join strategy tells Spark how to physically execute the join across the cluster. This is where performance lives or dies.

Broadcast join copies the entire smaller table to every executor node, then does a local hash lookup on each partition of the larger table. If the right table is under 100 MB (configurable via spark.sql.autoBroadcastJoinThreshold), Spark automatically broadcasts it. This is fast because there's no shuffle—no data movement across the network.

from pyspark.sql.functions import broadcast

# Force a broadcast join
orders_df.join(
    broadcast(customers_df),
    orders_df.customer_id == customers_df.id,
    "left"
)

The catch: if the broadcasted table is too large to fit in executor memory, Spark will spill to disk or fail. We've seen this happen when someone accidentally broadcasts a table that's been union'd with itself or contains many duplicates. Always check your DataFrame size with df.count() and a rough per-row estimate before forcing a broadcast.

Shuffle hash join partitions both tables by the join key, then does a hash join on each partition pair. Each executor gets a partition from the left and a matching partition from the right. This involves a shuffle—data movement across the network—but is simpler than sort-merge and often faster when tables aren't already sorted.

Sort-merge join sorts both tables by the join key, then merges them partition by partition. This is Spark's default for large table-to-large table joins. It's predictable and memory-efficient because Spark processes sorted streams without holding entire partitions in memory. However, if your tables are already sorted by the join key (e.g., a fact table partitioned by date and a dimension table sorted by ID), you can hint Spark to skip the sort.

spark.sql("""
    SELECT /*+ BROADCAST(customers_df) */
        orders.*, customers.name
    FROM orders
    JOIN customers
    ON orders.customer_id = customers.id
""")

In practice, let Spark choose the strategy most of the time. Use broadcast hints when you're certain the right table fits in memory and the join is a bottleneck. Tune spark.sql.shuffle.partitions (default 200) to match your cluster size if shuffles are slow.

Avoid duplicates: Check key uniqueness before joining

One of the costliest mistakes in production is joining on a non-unique key and not realizing it until your metrics explode. Imagine joining orders with customer records on customer_id, but your customers table has two rows for the same customer (a data quality issue). Your order counts just doubled—silently.

Always check key uniqueness before a join:

# Check uniqueness on the join key
left_duplicates = orders_df.groupBy("customer_id").count().filter(col("count") > 1)
right_duplicates = customers_df.groupBy("id").count().filter(col("count") > 1)

if left_duplicates.count() > 0:
    print(f"WARNING: {left_duplicates.count()} duplicate customer_ids in orders")
if right_duplicates.count() > 0:
    print(f"ERROR: {right_duplicates.count()} duplicate ids in customers (should be unique)")

For the left table, duplicates might be acceptable (multiple orders per customer), but for the right table (reference data like customers or products), duplicates are usually a red flag. If you find duplicates on the right, investigate the source. If you must proceed, decide: take the first row, aggregate, or fail the job. Silently multiplying rows is never the answer.

In our CRM pipelines, we've added a pre-join validation step that's become standard:

def validate_join_keys(left_df, right_df, left_key, right_key):
    left_count = left_df.count()
    left_distinct = left_df.select(left_key).distinct().count()
    right_count = right_df.count()
    right_distinct = right_df.select(right_key).distinct().count()
    
    if right_count != right_distinct:
        raise ValueError(f"Right table has {right_count - right_distinct} duplicates on key {right_key}")
    
    print(f"Left: {left_count} rows, {left_distinct} distinct keys")
    print(f"Right: {right_count} rows, {right_distinct} distinct keys (all unique ✓)")

validate_join_keys(orders_df, customers_df, "customer_id", "id")

Anti joins: Find records missing from another table

An anti join returns rows from the left table where the join key does not exist in the right table. This is powerful for data quality and reconciliation. Suppose you want to find orders that have no matching customer in your active customer table—a sign of orphaned or stale data.

orphaned_orders = orders_df.join(
    customers_df,
    orders_df.customer_id == customers_df.id,
    "leftanti"
)

print(f"Found {orphaned_orders.count()} orders with no active customer")
orphaned_orders.show()

Anti joins are also useful for identifying missing reference data. In a travel platform, you might want to see which flights in your booking table don't have a corresponding flight record in your flight schedule master—a data consistency check.

One key insight: anti joins are not the same as filtering where the right table is empty. An anti join is optimized by Spark to not pull any columns from the right table, making it faster than a left join followed by a where clause checking for NULLs. Use anti joins when possible.

Common mistakes

• Joining on non-unique keys without checking: A duplicate customer_id in the reference table silently multiplies your order counts. Validate first.
• Broadcasting too-large tables: Forcing a broadcast on a multi-GB table exhausts executor memory. Let Spark decide, or explicitly check size first.
• Using full outer joins everywhere: Full outer joins are expensive and often unnecessary. Use inner or left joins when you know the reference table is complete.
• Forgetting to repartition: If your join key has high cardinality and uneven distribution, the shuffle can be slow. Repartition on the join key before joining: df.repartition(500, "key").
• Joining on nullable columns: NULL values in join keys are typically ignored by Spark, leading to unexpected row loss. Handle NULLs explicitly before the join.
• Ignoring shuffle partition count: The default 200 partitions might be too many (small partitions, high overhead) or too few (skew, memory pressure). Match it to your cluster.

Key takeaways

• Choose the join type based on business logic: inner (strict match), left (keep all left rows), full (reconciliation), semi (filtering), anti (find missing references).
• Let Spark choose the join strategy in most cases, but understand broadcast (small table), shuffle hash (medium), and sort-merge (large table) so you can diagnose performance issues.
• Always validate join key uniqueness before joining, especially on the right (reference) table, to prevent silent data multiplication.
• Use anti joins for data quality checks and to find orphaned or missing references in your data.
• Monitor shuffle metrics and tune spark.sql.shuffle.partitions if joins are slow. A shuffle is often the bottleneck.
• Test joins on production data volumes in staging. A join that works on a sample might fail at scale due to memory or skew.