Spark join returned far more rows than expected: find and fix many-to-many key explosions

#Symptom

One of these has happened:

• A dashboard's revenue doubled overnight, and nothing failed.
• A job that took 10 minutes now runs for hours, spills to disk or dies with OutOfMemoryError.
• In the Spark UI, one join outputs far more rows than either input.

A join returns one output row for each matching pair. If a key appears 1,000 times on the left and 1,000 times on the right, that key alone makes 1,000,000 rows. Spark is doing what it was told.

This is separate from data skew, though the two often appear together.

#Step 1: confirm it

Compare the row counts:

from pyspark.sql import functions as F

left_n = orders.count()
right_n = customers.count()
joined = orders.join(customers, "customer_id", "left")
joined_n = joined.count()

print(f"left={left_n:,} right={right_n:,} joined={joined_n:,} ratio={joined_n/left_n:.2f}")

For a left join to a lookup table, the ratio should be exactly 1.00. Anything above that means that some keys match more than once.

If the job won't finish, don't run the join. Use Step 2, which calculates the result without it.

In the Spark UI, open the SQL/DataFrame tab and look at the join node's number of output rows. See the Spark UI guide for help.

#Step 2: predict the size without running the join

Count the keys on each side, and then multiply:

l = orders.groupBy("customer_id").count().withColumnRenamed("count", "l_n")
r = customers.groupBy("customer_id").count().withColumnRenamed("count", "r_n")

estimate = (l.join(r, "customer_id")
              .withColumn("pairs", F.col("l_n") * F.col("r_n")))

estimate.agg(F.sum("pairs").alias("inner_join_rows")).show()
estimate.orderBy(F.desc("pairs")).show(20, truncate=False)

This is cheap, because each side is reduced to a single row per key. The second query gives you the worst keys.

#Step 3: find out why the keys repeat

Check each side for uniqueness:

def key_report(df, keys):
    total = df.count()
    distinct = df.select(*keys).distinct().count()
    nulls = df.filter(" OR ".join(f"{k} IS NULL" for k in keys)).count()
    print(f"rows={total:,} distinct_keys={distinct:,} null_keys={nulls:,} dup_rows={total-distinct:,}")

key_report(customers, ["customer_id"])

(customers.groupBy("customer_id").count()
   .filter("count > 1")
   .orderBy(F.desc("count"))
   .show(20))

Then look at a sample of the duplicates, because what differs tells you the cause:

#Fix A: filter SCD2 dimensions

This is the most common cause. A history table has one row per version of a customer, not one row per customer.

# Current version only
current = customers.filter(F.col("is_current") == True)
joined = orders.join(current, "customer_id", "left")

For historical accuracy, join on the version that was valid at the time. The date range must not overlap between versions:

o = orders.alias("o")
c = customers.alias("c")

joined = o.join(
    c,
    (F.col("o.customer_id") == F.col("c.customer_id")) &
    (F.col("o.order_ts") >= F.col("c.valid_from")) &
    (F.col("o.order_ts") < F.col("c.valid_to")),
    "left",
)

Note the < on valid_to. Using <= matches two versions at the exact moment of a change. Check for overlapping ranges too, because they'll cause the same symptom. See dbt snapshots if that's how you build them.

#Fix B: join on the full key

# Wrong if customer_id is only unique per tenant
orders.join(customers, "customer_id")

# Right
orders.join(customers, ["tenant_id", "customer_id"])

Confirm that the full key is unique with key_report.

#Fix C: handle placeholders and NULLs

With a normal == join, NULL keys don't match each other. But <=> (eqNullSafe) treats them as equal, so a lot of NULLs on both sides multiply quickly.

Placeholder values do the same thing. Look for them:

orders.groupBy("customer_id").count().orderBy(F.desc("count")).show(10)

Separate these rows out and join them differently, or skip them:

BAD = ["UNKNOWN", "", "0", "-1"]
key = F.col("customer_id").cast("string")

good = orders.filter(key.isNotNull() & ~key.isin(BAD))
bad = orders.filter(key.isNull() | key.isin(BAD))

joined = good.join(customers, "customer_id", "left").unionByName(bad, allowMissingColumns=True)

Keep those rows in the output, or quarantine them, and count them. Don't lose them without a trace.

#Fix D: aggregate to the right grain first

If you're joining orders to order lines and summing the order total, the total counts once per line. Aggregate the many side before you join:

line_totals = (order_lines.groupBy("order_id")
               .agg(F.sum("line_amount").alias("lines_total"),
                    F.count("*").alias("line_count")))

result = orders.join(line_totals, "order_id", "left")

If you only need to know whether a match exists, use a semi or anti join. They never add rows:

with_refund = orders.join(refunds, "order_id", "left_semi")
no_refund = orders.join(refunds, "order_id", "left_anti")

#Dedupe with a rule, not at random

dropDuplicates(["customer_id"]) keeps an arbitrary row for each key, and the row it picks can change between runs. Use a window to choose the one you want:

from pyspark.sql.window import Window

w = Window.partitionBy("customer_id").orderBy(F.desc("updated_at"), F.desc("load_ts"))

latest = (customers
          .withColumn("rn", F.row_number().over(w))
          .filter("rn = 1")
          .drop("rn"))

Warning

Don't add .distinct() after the join as your fix. It may hide the symptom only when the duplicate rows happen to be identical, it's expensive, and the data may still be wrong. Fix the input to the join.

#Repair what it damaged

If a bad join has already written output, those numbers have to be corrected:

1. Find out when the problem began, from the row count history or Delta DESCRIBE HISTORY.
2. Fix the job.
3. Rewrite the affected partitions only. See overwrite and replaceWhere.
4. Tell the people who used the data, and reconcile the totals against the source.

#Add a guard

Make the job fail before it writes bad data:

def assert_unique(df, keys, name):
    dupes = df.groupBy(*keys).count().filter("count > 1").limit(5).collect()
    if dupes:
        raise ValueError(f"{name}: duplicate keys {keys}, for example {dupes}")

def assert_no_growth(before, after, name, tolerance=0.0):
    if after > before * (1 + tolerance):
        raise ValueError(f"{name}: join grew rows from {before:,} to {after:,}")

assert_unique(customers, ["customer_id"], "customers")

#Prevent it next time

• State each table's grain and key in its documentation, and test it. See data quality checks.
• Check uniqueness on the lookup side of every left join in a critical pipeline.
• Compare row counts and totals between runs, and alert on big changes.
• Agree who owns upstream key changes in a data contract.