DataFrame Joins in Databricks: Inner, Left, Multi-Key, Cross and Broadcast

1.The join() call: other, on and how

In PySpark you combine two DataFrames side by side with df.join(other, on, how). other is the right side of the join. on says how to match rows. It can be a single column name, a list of column names, a Column expression such as df.name == df2.name, or a list of Columns. how names the join type. When on is a name or a list of names, each of those columns must exist in both DataFrames, and Spark runs an equi-join, meaning rows match when the values are equal.

df = spark.createDataFrame([Row(name="Alice", age=2), Row(name="Bob", age=5)])
df2 = spark.createDataFrame([Row(name="Tom", height=80), Row(name="Bob", height=85)])

df.join(df2, "name").show()
# +----+---+------+
# |name|age|height|
# +----+---+------+
# | Bob|  5|    85|
# +----+---+------+

How you write on changes the shape of the result. The same reference page joins these DataFrames with the expression df.name == df2.name and an outer join, and that output has two name columns, one from each side. If you pass the string "name", the result has a single name column, as shown above. Alice and Tom are missing from the inner result because neither has a partner on the other side.

Sources1

2.Inner join vs left join: what happens to rows without a match

Inner and left joins handle matched rows the same way. They differ on the left rows that have no partner. An inner join returns only rows whose key values match in both tables, and it is the default join type in SQL as well as in PySpark. A left join (also called a left outer join) returns every row from the left table. Where the right table has no match, its columns are filled with NULL.

> SELECT id, name, employee.deptno, deptname
   FROM employee
   LEFT JOIN department ON employee.deptno = department.deptno;

Sources2

3.Joining on multiple keys

Sometimes one column cannot identify a match on its own. An order line may only be unique per (order_id, line_no), for example. In PySpark you pass a list of column names to on, and each name must exist in both DataFrames. In SQL you write USING (a, b), which matches rows by checking equality on every listed column. If a column is missing from one of the tables, Databricks raises UNRESOLVED_USING_COLUMN_FOR_JOIN. The SQL reference spells out what a multi-column USING join means.

PySpark has a second route to the same result. Instead of names, on can be a join expression (a Column) or a list of Columns, so you can write one equality per key yourself. That is the PySpark counterpart of the SQL ON first.a = second.a AND first.b = second.b form shown below. Like a single-key expression join, it keeps the key columns from both sides rather than merging them, while a list of names merges them.

SELECT * FROM first JOIN second USING (a, b)

is equivalent to

SELECT first.a, first.b,
       first.* EXCEPT(a, b),
       second.* EXCEPT(a, b)
  FROM first JOIN second ON first.a = second.a AND first.b = second.b

The equivalent query also shows how the columns are laid out. With USING (or NATURAL), SELECT * returns each matching column once, at the front. Next come the other columns from the left table, then the other columns from the right table. An ON condition does not merge the key columns this way. That is why the PySpark expression join above returned two name columns.

Sources21

4.Cross joins: every row against every row

A cross join has no match condition. It returns the Cartesian product, pairing every row on the left with every row on the right. In PySpark you call crossJoin(other) or pass how="cross". In SQL you write CROSS JOIN. The output row count is the left row count multiplied by the right row count.

df = spark.createDataFrame(
    [(14, "Tom"), (23, "Alice"), (16, "Bob")], ["age", "name"])
df2 = spark.createDataFrame(
    [Row(height=80, name="Tom"), Row(height=85, name="Bob")])
df.crossJoin(df2.select("height")).select("age", "name", "height"
    ).orderBy("age", "name", "height").show()
# +---+-----+------+
# |age| name|height|
# +---+-----+------+
# | 14|  Tom|    80|
# | 14|  Tom|    85|
# | 16|  Bob|    80|
# | 16|  Bob|    85|
# | 23|Alice|    80|
# | 23|Alice|    85|
# +---+-----+------+

This is why a Cartesian product can show up without anyone asking for one. The size matters too. In the SQL reference example, 6 employees cross-joined with 3 departments return 18 rows. Databricks' join optimization guidance is blunt about cross joins: take them out of any workload that needs low latency or is recomputed often. Keep them for cases where you really want every combination.

Sources34

5.Broadcast joins: marking the small side

Many joins pair a large fact table with a small lookup table. PySpark gives you broadcast(df) in pyspark.sql.functions for this. It marks a DataFrame as small enough to use in a broadcast join. You wrap the small side in broadcast() and join as usual. The join condition and join type do not change.

Why bother? The broadcast hint tells Spark to send the smaller table to all executors instead of performing a shuffle join, so the large side does not have to be shuffled. Spark can also choose a broadcast on its own: spark.sql.autoBroadcastJoinThreshold sets the maximum size of a table that will be broadcast (10 MB by default, and -1 turns it off). An explicit hint overrides that limit. The hinted side is broadcast regardless of the threshold, so you can broadcast a table that Spark's size estimate would otherwise leave out.

from pyspark.sql import functions as dbf
df = spark.createDataFrame([1, 2, 3, 3, 4], "int")
df_small = spark.range(3)
df_b = dbf. ? (df_small)
df.join(df_b, df.value == df_small.id).show()

In SQL you do the same thing with a join hint inside /*+ ... */. BROADCAST, BROADCASTJOIN and MAPJOIN are interchangeable names, and the hint can go on either table in the join. If a table has an alias, the hint has to use the alias name. When conflicting strategy hints are given, BROADCAST wins over MERGE, SHUFFLE_HASH and SHUFFLE_REPLICATE_NL.

-- Join Hints for broadcast join
> SELECT /*+ BROADCAST(t1) */ * FROM t1 INNER JOIN t2 ON t1.key = t2.key;
> SELECT /*+ BROADCASTJOIN (t1) */ * FROM t1 left JOIN t2 ON t1.key = t2.key;
> SELECT /*+ MAPJOIN(t2) */ * FROM t1 right JOIN t2 ON t1.key = t2.key;

Hints are only part of the story. On compute with Photon enabled, Databricks picks the join type for you. When a query has several joins, join the smallest tables first, then join that result to the larger tables.

Sources56783

Exam traps

Each one states something that sounds right. Open it to see what is actually true.

1. 1.If you leave out how, df.join(df2, "key") keeps every row of df, like a left join.Why is that wrong?

   The default join type is inner. Rows of df with no match in df2 are dropped unless you pass how="left".

   Covered in The join() call: other, on and how

2. 2.A SQL join with no ON or USING clause fails, so you cannot get a Cartesian product by accident.Why is that wrong?

   With no join criteria, any join type behaves as a CROSS JOIN and returns every pairing of rows.

   Covered in Cross joins: every row against every row