Spark DataFrame Joins and Unions: Inner, Left, Broadcast, Cross, Union

1.The anatomy of DataFrame.join

Every join in this lesson goes through DataFrame.join(other, on=None, how=None), apart from crossJoin, which has its own method. other is the right side of the join. on can take four forms: a single column name as a string, a list of column names, a join expression built from Columns, or a list of Column expressions. how names the join type. If you leave it out, you get an inner join.

The form of on matters more than it looks. If you pass a string or a list of strings, the named columns must exist in both DataFrames, and Spark runs an equi-join on them. If you pass a Column expression such as df.name == df2.name, you can write any boolean condition, but both sides' columns stay in the result.

import pyspark.sql.functions as sf
from pyspark.sql import Row
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|
# +----+---+------+

The how parameter accepts a fixed set of strings, and several are synonyms. left, leftouter and left_outer all mean the same join, and so do outer, full, fullouter and full_outer. The list also includes cross, semi/leftsemi/left_semi and anti/leftanti/left_anti. On the exam, a spelling like left_outer is a valid choice, not a typo.

Sources1

2.Inner join vs left join: which rows survive

Inner and left joins differ in what happens to a row that has no match. Take an employee table with six people in departments 1–6 and a department table that lists only departments 1, 2 and 3.

-- Use employee and department tables to demonstrate left join.
> SELECT id, name, employee.deptno, deptname
   FROM employee
   LEFT JOIN department ON employee.deptno = department.deptno;
 105 Chloe      5        NULL
 103  Paul      3 Engineering
 101  John      1   Marketing
 102  Lisa      2       Sales
 104  Evan      4        NULL
 106   Amy      6        NULL

Change LEFT JOIN to INNER JOIN and only Paul, John and Lisa come back, because an inner join returns only rows with matching values in both tables. The DataFrame API follows the same rules. how="inner" (or no how) is the inner join, and how="left" is the left outer join. The left side is the DataFrame you call .join on, and the right side is other.

Sources2

3.Joining on multiple keys and avoiding duplicate key columns

To join on several keys, pass a list. With a list of column names such as ["a", "b"], every named column must exist on both sides, and Spark runs an equi-join on all of them. A row matches only when every key is equal. SQL does the same thing with USING (a, b), which the SQL reference defines as an ON clause that joins the equality tests with AND, as in ON first.a = second.a AND first.b = second.b. For conditions that aren't plain equality, pass Column expressions instead, either as one expression combined with & or as a list of Columns.

joined = df.join(df2, df.name == df2.name, "outer").sort(sf.desc(df.name))
joined.show()
# +-----+----+----+------+
# | name| age|name|height|
# +-----+----+----+------+
# |  Bob|   5| Bob|    85|
# |Alice|   2|NULL|  NULL|
# | NULL|NULL| Tom|    80|
# +-----+----+----+------+

Joining by name (a string or list of strings in the DataFrame API, or USING/NATURAL in SQL) collapses each key into a single output column, placed first. An expression join keeps the key from each side. That is usually why a later select("name") fails as ambiguous. When you can't avoid an expression join, for example in a self-join, give each side an alias with df.alias("a") and refer to columns as "a.name" and "b.name".

Sources12

4.Broadcast joins: sending the small side to every worker

Joins often pair a large fact table with a small lookup table. A broadcast join copies the whole small table to all worker nodes, so each worker can join its part of the large table locally. A Databricks engineering post puts it this way: broadcast hash join is usually the fastest strategy when one side fits comfortably in memory.

You can trigger it in two ways. The explicit way is pyspark.sql.functions.broadcast(df), which marks a DataFrame as small enough to broadcast. You then join as usual. The automatic way is the setting spark.sql.autoBroadcastJoinThreshold, which defaults to 10485760 bytes (10 MB). Spark plans a broadcast hash join on its own when a relation's estimated size is below this threshold. Setting it to -1 turns automatic broadcasting off. A BROADCAST join hint tells Spark to use that strategy for the relation you mark.

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()

Sources345

5.Cross joins: the Cartesian product

A cross join has no join condition. It pairs every row on the left with every row on the right, so the result has left × right rows. In the DataFrame API you write it as df.crossJoin(other). In SQL you write CROSS JOIN. The same employee/department tables (6 rows and 3 rows) produce 18 rows.

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|
# +---+-----+------+

You can also get a cross join by accident. The SQL reference warns that if you leave out the join criteria, any join type behaves as a CROSS JOIN. Because the output grows with the product of the two inputs, Databricks' join-performance guidance says cross joins are expensive and should be removed from workloads that need low latency or frequent recomputation.

Sources627

6.Stacking rows: union, UNION ALL and unionByName

Joins add columns side by side. Unions stack rows, and the main question is what happens to duplicate rows.

df1 = spark.createDataFrame([(1, 'A'), (2, 'B'), (3, 'C')], ['id', 'value'])
df2 = spark.createDataFrame([(3, 'C'), (4, 'D')], ['id', 'value'])
df3 = df1.union(df2).distinct().sort("id")
df3.show()
# +---+-----+
# | id|value|
# +---+-----+
# |  1|    A|
# |  2|    B|
# |  3|    C|
# |  4|    D|
# +---+-----+

SQL works the other way round. In SQL, UNION without a qualifier means UNION DISTINCT and removes duplicates. You write UNION ALL to keep them. Both sides must have the same number of columns with compatible types, otherwise you get NUM_COLUMNS_MISMATCH or INCOMPATIBLE_COLUMN_TYPE. So the DataFrame method union() gives you SQL's UNION ALL result. A note on scope: the sources for this lesson don't document a separate DataFrame unionAll method. They list only union and unionByName. What this lesson can teach is the union-all behaviour, which keeps duplicates.

union() matches columns by position, not by name. If one DataFrame has id, value and the other has value, id, union() silently puts each value under the wrong column. unionByName matches columns by name instead. With allowMissingColumns=True, it also accepts columns that exist on only one side and fills the gaps with NULL.

df1 = spark.createDataFrame([[1, 2, 3]], ["col0", "col1", "col2"])
df2 = spark.createDataFrame([[4, 5, 6]], ["col1", "col2", "col0"])
df1.unionByName(df2).show()
# +----+----+----+
# |col0|col1|col2|
# +----+----+----+
# |   1|   2|   3|
# |   6|   4|   5|
# +----+----+----+

Sources8910

Exam traps

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

1. 1.DataFrame.union() removes duplicate rows the way SQL UNION does.Why is that wrong?

   union() keeps duplicates, so it behaves like SQL UNION ALL. Add .distinct() if you want them removed.

   Covered in Stacking rows: union, UNION ALL and unionByName

2. 2.union() lines columns up by name, so the column order of the two DataFrames doesn't matter.Why is that wrong?

   union() matches columns by position. When the column orders differ, use unionByName.

   Covered in Stacking rows: union, UNION ALL and unionByName

3. 3.A SQL join with no ON or USING clause fails to run.Why is that wrong?

   Without join criteria, any join type behaves as a cross join and returns the Cartesian product, which can be very large.

   Covered in Cross joins: the Cartesian product

4. 4.Spark only runs a broadcast join when you call broadcast().Why is that wrong?

   Spark also broadcasts on its own when a table's estimated size is below spark.sql.autoBroadcastJoinThreshold, which defaults to 10 MB. Setting the threshold to -1 turns this off.

   Covered in Broadcast joins: sending the small side to every worker