Joins in Spark

A join puts two tables side by side. Each row is matched with the rows that have the same key. You know this from SQL: inner, left and anti joins, NULL keys that never match, and rows that multiply.

In Spark, a join is also a big cost. The chapter "The Shuffle: Spark's Performance Killer" showed that a shuffle moves rows between machines. A join can need that move for both tables.

Interviewers often ask why a Spark job is slow. The answer is often in the join. Look at the join strategy, the partition count, or one key that has far more rows than the others.

You find the cause in the query plan.

This chapter has three parts. Each one has its own sections.

1. What each join type returns, on the shop tables.
2. The three strategies Spark uses to run a join, and how to read them in the plan.
3. The mistakes that make a join slow or wrong.

We use the shop again: 6 orders and 5 customers. Customer 5, Omar, has no orders. Customer 4, Dev, has a NULL country.

An inner join keeps only the rows that have a match on both sides. Here the key is customer_id. Look at the code and the result.

orders.join(customers, "customer_id") \
      .select("order_id", "name", "country", "amount").show()

The second argument "customer_id" is the join key. The default type is inner. Dev's NULL is in a normal column, so it does not stop the match.

A left join keeps every row of the left table. Where the right table has no match, Spark fills NULL. Now put customers on the left, so Omar stays.

customers.join(orders, "customer_id", "left") \
         .select("name", "order_id", "amount").show()

The other types are mirrors. A right join keeps every right row. A full join keeps every row from both sides.

Sometimes you only ask a yes or no question: does this customer have an order? A semi join and an anti join answer it. Both return only the left columns.

A left semi join keeps the left rows that have a match. Look at the customers who ordered.

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

A left anti join keeps the left rows that have no match. It finds the missing ones.

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

The semi rows and the anti rows together are all 5 customers: 4 + 1.

The join type says which rows you get. The join strategy says how Spark gets them. The strategy decides how much data moves.

Spark has three main strategies. The next two sections explain the first two in detail.

Spark uses sort-merge by default for two big tables. It uses a broadcast when one table is small. Shuffle hash join is rare, because Spark prefers sort-merge unless you change the setting spark.sql.join.preferSortMergeJoin.

Usually Spark chooses the strategy from the table sizes. You can also give a hint, as the broadcast section shows. You see the choice in the plan.

A sort-merge join must put equal keys in the same partition. So it shuffles both tables by the join key. Both use the same hash and the same partition count.

We use the model from the shuffle chapter: 3 partitions, and customer_id % 3 as the hash. Here is the new partition for each customer_id. It is not the old partition the order came from.

Orders and customers both follow this table. So order 104 and customer 3 land in new partition 0. Look at what each new partition receives, and what it joins.

spark.conf.set("spark.sql.autoBroadcastJoinThreshold", -1)   # turn auto broadcast off

orders.join(customers, "customer_id")

Inside each new partition, Spark sorts both sides by customer_id. Then it walks down both lists once, in step.

In new partition 1, the orders have customer_id 1, 1, 4. The customers have 1, 4. Spark reads each list once, from top to bottom, and never searches back.

Here is the plan for this join. It is simplified: real output has more detail.

== Physical Plan (simplified) ==
SortMergeJoin [customer_id], [customer_id], Inner
:- Sort [customer_id ASC]
:  +- Exchange hashpartitioning(customer_id, 200)
:     +- FileScan parquet orders
+- Sort [customer_id ASC]
   +- Exchange hashpartitioning(customer_id, 200)
      +- FileScan parquet customers

A broadcast hash join does not shuffle the big table. Spark sends a full copy of the small table to every executor. Each task then joins its own orders against the local copy.

A hash table finds a row by its key in one step. Spark builds one from the small table and sends it to every executor. Look at the same 3 tasks, with orders staying where they are.

from pyspark.sql import functions as F

orders.join(F.broadcast(customers), "customer_id")

Spark also broadcasts on its own. It does so when a table is smaller than spark.sql.autoBroadcastJoinThreshold.

The default is 10 MB. A value of -1 turns this off.

Spark needs a size estimate for that. If it has none, it may not broadcast. Then the hint helps.

Here is the plan for the broadcast join.

== Physical Plan (simplified) ==
BroadcastHashJoin [customer_id], [customer_id], Inner, BuildRight
:- FileScan parquet orders
+- BroadcastExchange
   +- FileScan parquet customers

One more rule. Spark cannot broadcast the side that a left join keeps in full. In orders.join(customers, "customer_id", "left"), only customers can be broadcast.

Why? Each task sees only some of the right table, so it cannot know if a left row matched in another task.

Before a job runs, Spark only guesses table sizes. A filter can make a table much smaller than the guess.

The chapter "The Shuffle: Spark's Performance Killer" introduced AQE, adaptive query execution. It looks at real sizes while the job runs.

AQE can change a sort-merge join to a broadcast join when one side turns out small. It is on by default since Spark 3.2.

Imagine customers has millions of rows, and the broadcast setting is back at its default. Look at a filtered customers table.

in_customers = customers.filter(F.col("country") == "IN")
orders.join(in_customers, "customer_id") \
      .select("order_id", "name", "amount").show()

The switch happens after a stage has finished. The big table may already be shuffled by then. So the saving is smaller than with a planned broadcast.

You can write the key in two ways. With a column name, "customer_id", Spark keeps one customer_id column.

With a column expression, Spark keeps both columns. Look at the expression form.

joined = orders.join(customers, orders.customer_id == customers.customer_id)

joined.select("customer_id", "order_id", "name").show()

You have two fixes. Join on the name, or drop one key column. Here is the second fix.

joined.drop(customers.customer_id) \
      .select("customer_id", "order_id", "name").show()

The second key problem is types. Suppose customers was read from a CSV file with no schema.

Then every column is a string. Look at the two types.

orders.schema["customer_id"].dataType      # IntegerType
customers_csv.schema["customer_id"].dataType   # StringType

Spark casts one side for you, and shows no warning. Suppose one row has customer_id "" (empty text).

The cast turns it into NULL, and a NULL key never matches. That customer drops out of the join.

With ANSI mode on (the default in Spark 4), the bad value stops the job with an error instead.

Make the types the same before the join. Cast the string column to int.

customers_ok = customers_csv.withColumn("customer_id", F.col("customer_id").cast("int"))
orders.join(customers_ok, "customer_id")

Fan-out means a join makes more rows than you expect. It happens when a key repeats on a side you thought had each key once. Suppose a bad load put Asha in the customers table twice.

orders.join(customers_loaded_twice, "customer_id") \
      .groupBy("name").sum("amount").show()

A hot key is a key value with far more rows than the others. A shuffle sends all rows with the same key to one partition. So one task gets most of the work, and the job waits for it.

Rows with a NULL key never match. An inner join drops them early. A left or full join keeps them, and they all go to one partition.

So many NULL keys also make a hot key.

The chapter "Data Skew & Salting" shows how to fix hot keys.

Four points to keep.

• The join type decides which rows you get. The join strategy decides how much data moves.
• A sort-merge join shuffles both tables. A broadcast hash join shuffles neither. You see the difference in the plan.
• Broadcast only a small table. A hint does not check the size.
• Check the key types, the key columns, and the row count after every join.

Answer these without looking back.

1. Which join type returns Omar, and which join type returns only Omar?
2. In a plan, which lines show that a sort-merge join shuffled both tables?
3. Why can a left join not broadcast its left table?
4. You join orders to a table with a repeated customer_id. The sums double. Why?