When does Spark broadcast automatically?

When it can tell one side is smaller than spark.sql.autoBroadcastJoinThreshold, 10 MB by default. It relies on table statistics, so a DataFrame built from a transformation Spark cannot size — after a UDF, or from a source with no stats — will not be broadcast even if it is tiny. That is when the explicit broadcast hint earns its place.

Why do I get duplicate column names after a join?

Joining on a condition rather than a column name keeps both sides' columns, so a shared name appears twice and referencing it raises an ambiguity error. Join with the string form, join(other, 'id'), which collapses the key into one column, or alias the DataFrames and select explicitly.

What is a salted join?

A technique for skewed keys: append a random suffix to the hot key on the large side and explode the small side to match every suffix. That splits one enormous partition into several, at the cost of duplicating the small side. Adaptive Query Execution handles many skew cases automatically in Spark 3.

Is a left anti join the same as NOT IN?

Almost, and the difference matters. A left anti join keeps rows from the left with no match on the right, and handles nulls sensibly. SQL's NOT IN returns no rows at all if the subquery contains a single null, which is a classic silent bug. Prefer the anti join.

Joins and Broadcast Joins in PySpark

Joins are where Spark jobs most often go from minutes to hours. The API is straightforward; the performance is entirely about whether Spark has to shuffle both sides.

The data

from pyspark.sql import SparkSession, functions as F

spark = SparkSession.builder.appName("joins").master("local[*]").getOrCreate()

orders = spark.createDataFrame(
    [(1, "alice", 250.00), (2, "bob", 180.50), (3, "carol", 320.75),
     (4, "dave", 95.00),   (5, "zoe", 410.20)],
    ["order_id", "customer", "amount"],
)

customers = spark.createDataFrame(
    [("alice", "UK", "gold"), ("bob", "US", "silver"),
     ("carol", "UK", "gold"), ("erin", "DE", "bronze")],
    ["customer", "country", "tier"],
)

orders.show()
customers.show()

+--------+--------+------+
|order_id|customer|amount|
+--------+--------+------+
|       1|   alice| 250.0|
|       2|     bob| 180.5|
|       3|   carol|320.75|
|       4|    dave|  95.0|
|       5|     zoe| 410.2|
+--------+--------+------+

+--------+-------+------+
|customer|country|  tier|
+--------+-------+------+
|   alice|     UK|  gold|
|     bob|     US|silver|
|   carol|     UK|  gold|
|    erin|     DE|bronze|
+--------+-------+------+

dave and zoe have orders but no customer record. erin has a record but no orders. Those mismatches are what distinguish the join types.

Inner join

orders.join(customers, "customer").show()

+--------+--------+------+-------+------+
|customer|order_id|amount|country|  tier|
+--------+--------+------+-------+------+
|   alice|       1| 250.0|     UK|  gold|
|     bob|       2| 180.5|     US|silver|
|   carol|       3|320.75|     UK|  gold|
+--------+--------+------+-------+------+

Three rows out of five. dave and zoe vanished, silently. An inner join is a filter as much as it is a join — losing rows to a missing lookup is the most common data bug in a pipeline, and nothing in the output tells you it happened.

Guard against it:

joined = orders.join(customers, "customer")
print(f"before: {orders.count()}  after: {joined.count()}  dropped: {orders.count() - joined.count()}")

before: 5  after: 3  dropped: 2

Left join

orders.join(customers, "customer", "left").show()

+--------+--------+------+-------+------+
|customer|order_id|amount|country|  tier|
+--------+--------+------+-------+------+
|   alice|       1| 250.0|     UK|  gold|
|     bob|       2| 180.5|     US|silver|
|   carol|       3|320.75|     UK|  gold|
|    dave|       4|  95.0|   NULL|  NULL|
|     zoe|       5| 410.2|   NULL|  NULL|
+--------+--------+------+-------+------+

All five orders kept, unmatched columns filled with null. This is usually what you want for enrichment — keep the facts, accept incomplete dimensions.

Anti and semi joins

The cleanest way to ask “which rows have no match”:

orders.join(customers, "customer", "left_anti").show()

+--------+--------+------+
|customer|order_id|amount|
+--------+--------+------+
|    dave|       4|  95.0|
|     zoe|       5| 410.2|
+--------+--------+------+

Exactly the orphaned orders, with no null columns to filter on. A semi join is the mirror — rows from the left that do match, without bringing the right side’s columns:

orders.join(customers, "customer", "left_semi").show()

+--------+--------+------+
|customer|order_id|amount|
+--------+--------+------+
|   alice|       1| 250.0|
|     bob|       2| 180.5|
|   carol|       3|320.75|
+--------+--------+------+

Semi joins are cheaper than inner joins when you only need existence, because Spark can stop scanning the right side once it finds one match per key.

Full outer

orders.join(customers, "customer", "full_outer").orderBy("customer").show()

+--------+--------+------+-------+------+
|customer|order_id|amount|country|  tier|
+--------+--------+------+-------+------+
|   alice|       1| 250.0|     UK|  gold|
|     bob|       2| 180.5|     US|silver|
|   carol|       3|320.75|     UK|  gold|
|    dave|       4|  95.0|   NULL|  NULL|
|    erin|    NULL|  NULL|     DE|bronze|
|     zoe|       5| 410.2|   NULL|  NULL|
+--------+--------+------+-------+------+

Everything from both sides. Useful for reconciliation — the null columns tell you which side each mismatch came from.

Duplicate columns

Joining on a condition instead of a name keeps both key columns:

bad = orders.join(customers, orders.customer == customers.customer, "inner")
bad.select("customer").show()

pyspark.errors.exceptions.captured.AnalysisException:
[AMBIGUOUS_REFERENCE] Reference `customer` is ambiguous, could be: [`customer`, `customer`].

Two columns named customer and no way to say which you meant. Either use the string form, which collapses the key:

orders.join(customers, "customer").select("customer").show(2)

+--------+
|customer|
+--------+
|   alice|
|     bob|
+--------+

or alias the DataFrames and qualify:

o, c = orders.alias("o"), customers.alias("c")
o.join(c, F.col("o.customer") == F.col("c.customer")) \
 .select("o.customer", "o.amount", "c.tier").show(2)

+--------+------+------+
|customer|amount|  tier|
+--------+------+------+
|   alice| 250.0|  gold|
|     bob| 180.5|silver|
+--------+------+------+

Aliasing is the habit worth building — it survives refactoring, and the qualified names make the intent obvious to a reader.

Shuffle join vs broadcast join

Build a large fact table and a small dimension:

import time

big = spark.range(20_000_000).select(
    (F.col("id") % 1000).alias("product_id"),
    (F.rand() * 100).alias("amount"),
)
small = spark.range(1000).select(
    F.col("id").alias("product_id"),
    F.concat(F.lit("product-"), F.col("id")).alias("product_name"),
)

Force a shuffle join by disabling auto-broadcast:

spark.conf.set("spark.sql.autoBroadcastJoinThreshold", -1)

start = time.perf_counter()
big.join(small, "product_id").agg(F.sum("amount")).collect()
print(f"shuffle join: {time.perf_counter() - start:.2f}s")
big.join(small, "product_id").explain()

shuffle join: 14.82s

== Physical Plan ==
*(5) Project [product_id#2L, amount#3, product_name#7]
+- *(5) SortMergeJoin [product_id#2L], [product_id#6L], Inner
   :- *(2) Sort [product_id#2L ASC NULLS FIRST], false, 0
   :  +- Exchange hashpartitioning(product_id#2L, 200), ENSURE_REQUIREMENTS
   :     +- ...
   +- *(4) Sort [product_id#6L ASC NULLS FIRST], false, 0
      +- Exchange hashpartitioning(product_id#6L, 200), ENSURE_REQUIREMENTS
         +- ...

Two Exchange nodes — both sides shuffled — then both sorted, then merged. Twenty million rows crossed the network to be co-located with a thousand-row table.

Now broadcast the small side:

start = time.perf_counter()
big.join(F.broadcast(small), "product_id").agg(F.sum("amount")).collect()
print(f"broadcast join: {time.perf_counter() - start:.2f}s")
big.join(F.broadcast(small), "product_id").explain()

broadcast join: 3.41s

== Physical Plan ==
*(2) Project [product_id#2L, amount#3, product_name#7]
+- *(2) BroadcastHashJoin [product_id#2L], [product_id#6L], Inner, BuildRight
   :- ...
   +- BroadcastExchange HashedRelationBroadcastMode(...)
      +- ...

Four times faster. BroadcastHashJoin with one BroadcastExchange: the small table is sent once to every executor and held in memory, so the large side never moves. No sort, no shuffle of 20 million rows.

Broadcast whenever one side fits comfortably in executor memory. The default automatic threshold is 10 MB:

print(spark.conf.get("spark.sql.autoBroadcastJoinThreshold"))

10485760

Raise it if your executors have room, or use the explicit F.broadcast() hint when Spark cannot estimate the size — which happens after UDFs and with sources that carry no statistics.

Broadcasting something too large fails loudly rather than silently degrading:

org.apache.spark.SparkException: Cannot broadcast the table that is larger than 8GB: 12 GB

Skew

When one key dominates, one task does most of the work:

skewed = spark.range(10_000_000).select(
    F.when(F.rand() < 0.9, F.lit(1)).otherwise((F.rand() * 1000).cast("int")).alias("key"),
    F.rand().alias("value"),
)
skewed.groupBy("key").count().orderBy(F.desc("count")).show(3)

+---+-------+
|key|  count|
+---+-------+
|  1|9000432|
|704|   1043|
|218|   1021|
+---+-------+

Ninety percent of rows share one key, so one partition gets 9 million rows and the other 199 get a handful. The job’s runtime is that one task’s runtime, and adding executors changes nothing.

Spark 3’s adaptive execution detects and splits this automatically:

print(spark.conf.get("spark.sql.adaptive.skewJoin.enabled"))

true

When AQE cannot handle it, salt the key — add a random suffix on the large side and explode the small side to match:

SALT = 10
salted_big = skewed.withColumn("salt", (F.rand() * SALT).cast("int")) \
                   .withColumn("join_key", F.concat_ws("_", "key", "salt"))

salted_small = small.withColumn("salt", F.explode(F.array([F.lit(i) for i in range(SALT)]))) \
                    .withColumn("join_key", F.concat_ws("_", "product_id", "salt"))

salted_big.join(salted_small, "join_key").count()

The hot key becomes ten keys, spread across ten tasks. The cost is duplicating the small side tenfold — worth it when one task is holding up the whole job.

Practice

1. Which orders have no matching customer? Use two different methods and compare.

orders.join(customers, "customer", "left_anti").show()
orders.join(customers, "customer", "left").filter(F.col("tier").isNull()).select("order_id", "customer", "amount").show()

Both return dave and zoe. The anti join is cheaper — it stops at the first match per key and never materialises the right side’s columns. It is also clearer about intent.

2. Set autoBroadcastJoinThreshold to 0 and check the plan for a tiny join.

+- SortMergeJoin [customer#1], [customer#5], Inner

SortMergeJoin even on four rows. Auto-broadcast is doing real work by default — disabling it is a diagnostic tool, not a tuning step.

3. Join on two columns at once.

orders.join(customers, ["customer", "country"], "left")

AnalysisException: [UNRESOLVED_USING_COLUMN_FOR_JOIN] USING column `country` cannot be
resolved on the left side of the join.

orders has no country. Multi-column joins with the list form require the column to exist on both sides; otherwise use an explicit condition with &.

4. What does a cross join of a 1000-row and a 1000-row DataFrame produce?

print(small.crossJoin(small.withColumnRenamed("product_id", "pid2")).count())

A million rows — every pair. Spark requires spark.sql.crossJoin.enabled or an explicit crossJoin call precisely because an accidental cross join is catastrophic at scale. A join condition that never matches degenerates into one, which is worth remembering when a job that usually takes minutes suddenly does not finish.

Next: window functions — ranking, running totals, and comparisons across rows.