TopolearnJoins are the most common source of wrong results and slow stages in Spark jobs.
left.join(right, on="uid", how="left") joins on equally named columns; on=["uid", "day"] for several keys; or an expression left.uid == right.user_id for different names. Join types: inner (default), left, right, full, left_semi (left rows with a match, left columns only), left_anti (left rows with no match), and cross.
Ambiguous columns: joining with an expression keeps both key columns, and other same-named columns clash. Joining with on="uid" (a name or list) keeps one copy of the key. Otherwise alias the inputs, a = df1.alias("a"), and refer to F.col("a.amount").
Null keys never match in normal equality joins. Use eqNullSafe if nulls should join.
Duplicates multiply: if a key repeats on both sides, the output has the product of the counts. Check key uniqueness before joining dimension tables.
Performance: joining two big tables normally needs a shuffle of both sides (a sort-merge join). If one side is small (by default under 10 MB, spark.sql.autoBroadcastJoinThreshold), Spark broadcasts it to every executor and skips the shuffle. Force that with big.join(F.broadcast(small), "key") when Spark's size estimate is wrong. Broadcasting something large can exhaust driver and executor memory.
Example
from pyspark.sql import SparkSession, functions as F
spark = SparkSession.builder.master("local[2]").getOrCreate()
users = spark.createDataFrame([(1, "Ana"), (2, "Ben"), (3, "Cy")], "uid INT, name STRING")
orders = spark.createDataFrame([(1, 10), (1, 5), (3, 7), (4, 9)], "uid INT, total INT")
left = users.join(orders, on="uid", how="left").orderBy("uid", "total")
print([tuple(r) for r in left.collect()])
# [(1, 'Ana', 5), (1, 'Ana', 10), (2, 'Ben', None), (3, 'Cy', 7)]
never = users.join(orders, "uid", "left_anti")
print([r.name for r in never.collect()]) # ['Ben']
small = F.broadcast(users) # hint: ship to every executor
plan = orders.join(small, "uid")._jdf.queryExecution().executedPlan().toString()
print("BroadcastHashJoin" in plan) # TrueRemember
- Choose the join type by which rows must survive; semi/anti filter without adding columns
- Join on column names to avoid duplicate key columns
- Broadcast small tables to avoid shuffling the big one
Watch out for
- Joining on keys that repeat on both sides and silently multiplying row counts.