Data Skew & Salting

In the chapter "The Shuffle: Spark's Performance Killer", rows with the same key meet in one new partition. This chapter is about what goes wrong when one key has too many rows.

Here is an example scenario for an interview. A stage has 200 tasks.

One task runs for 40 minutes. The other 199 tasks run for 1 minute each. The question is: why, and what do you do?

The answer is data skew. This chapter has five parts.

1. What skew is, on the shop data.
2. How to find it: the Spark UI and a count query.
3. Why one slow task hurts the whole job.
4. Three easy fixes to try first.
5. Salting, the fix you write by hand, and what it costs.

Data skew means one key has far more rows than the other keys.

That key is the hot key. After a shuffle, all rows of one key go to one partition.

So a hot key makes one very big partition.

The real shop has only 6 orders. To show skew, we add 12 made-up orders.

Asha (customer 1) starts to buy a lot. Orders 201 to 212 all belong to her. Each one is paid and has the amount 10.00.

We use the same model as before: new partition = customer_id % 3. Look at how the 18 orders spread out after a shuffle by customer_id.

orders.join(customers, "customer_id")   # shuffles orders by customer_id

One task works on each new partition. The task for partition 1 has 15 rows. The other two tasks have 2 rows and 1 row.

Real skew has millions of rows, not 15. The shape is the same.

You find skew in two steps. First, check the Spark UI to see that the stage has skew. Second, run a count query to find the hot key.

In the Spark UI, open the slow stage. Find the table "Summary Metrics" for the completed tasks.

It sorts the tasks by each column and shows five values: Min, 25th percentile, Median, 75th percentile and Max.

The Median is the middle task. Half the tasks are smaller, and half are bigger. Two columns matter: Duration, and Shuffle Read Size / Records.

Skew shows as a Max much larger than the Median. Look at our model.

The Max (15) is much larger than the Median (2). In the interview scenario, the Duration column would show the same: a Median of 1 minute and a Max of 40 minutes.

The UI tells you that skew exists. It does not tell you which key. For that, count the rows for each key.

from pyspark.sql import functions as F

(orders.groupBy("customer_id")
       .count()
       .orderBy(F.desc("count"), "customer_id")
       .show())

The joins chapter showed that NULL keys kept by a left or full join all go to one partition.

So also count the NULL keys. Our orders have no NULL customer_id, but real tables often do.

A stage ends only when its last task ends. In the scenario, 199 tasks finish after 1 minute. The stage still waits 39 more minutes for the last task.

During those 39 minutes, most cores have no work. The job pays for a cluster and uses one core.

The slow task has a second problem. In our model, it holds 15 rows. The other tasks hold 1 or 2.

If the data does not fit, Spark writes the extra to disk. You met this in the shuffle chapter: it is a spill.

The Spark UI shows spill in two columns: Spill (Memory) and Spill (Disk). A spill on only the slowest task is another sign of skew.

Try these three fixes in order. Each one needs less work than salting, which comes next.

Fix 1: handle the NULL keys. In an inner join, Spark already drops NULL keys before the shuffle.

The problem is a left or full join. It keeps the NULL rows, and they all go to one partition. Split them off, join the rest, then add them back.

with_key = orders.filter(F.col("customer_id").isNotNull())
no_key = orders.filter(F.col("customer_id").isNull())
joined = with_key.join(customers, "customer_id", "left")
result = joined.unionByName(no_key, allowMissingColumns=True)

Fix 2: broadcast the small side. The joins chapter showed the broadcast hash join.

Spark sends a copy of the small table to every executor. The big table does not move, so there is no shuffle by key.

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

# Plan lines (simplified):
# BroadcastHashJoin [customer_id]
# +- BroadcastExchange          <- the small table is copied
# (no Exchange hashpartitioning on orders)

This needs the small side to fit in memory on every machine. If both sides are large, go to fix 3.

Fix 3: let AQE split the partition. AQE (Adaptive Query Execution) is on by default since Spark 3.2. Its skew join setting is spark.sql.adaptive.skewJoin.enabled, on by default when AQE is on.

AQE sees the real partition sizes after the shuffle. It splits a skewed partition of a sort-merge join into smaller pieces.

Newer Spark versions also do this for a shuffled hash join. Spark copies the matching rows of the other side for each piece, so the result stays the same.

AQE splits joins only. It does not split a groupBy.

AQE acts only on a partition that is much bigger than the others. It must be over 5 times the median and over 256 MB (the default limits). A mild skew can pass without a split.

Use salting when fixes 1 to 3 do not work. A salt is a random number in a new column, next to the key.

The key is now the pair (customer_id, salt). A key with 3 different salts looks like 3 different keys, so it goes to up to 3 partitions.

Salting has four steps. We use N = 3 salts, 0 to 2.

1. Put a random salt, 0 to N - 1, in a new column on every row of the big side.
2. Copy each row of the small side N times, one copy for each salt. This is called exploding.
3. Join on (customer_id, salt) instead of customer_id.
4. For an aggregate, add up twice: first by (customer_id, salt), then by customer_id. Step 2 puts Asha's 3 pieces back into one row.

We add up every order, in any status. Ben's 60.00 is a refunded order.

This is the code for the job "join each order to its customer, then find the total amount for each customer".

N = 3
salted_orders = orders.withColumn("salt", (F.rand() * N).cast("int"))

salts = spark.range(N).select(F.col("id").cast("int").alias("salt"))
salted_customers = customers.crossJoin(salts)      # 5 x 3 = 15 rows

joined = salted_orders.join(salted_customers, ["customer_id", "salt"])

step1 = joined.groupBy("customer_id", "salt").agg(F.sum("amount").alias("partial"))
step2 = step1.groupBy("customer_id").agg(F.sum("partial").alias("total"))

Here is one possible run. The salts are made up so we can check the numbers.

We still use a model, not Spark's real hash: new partition = (customer_id + salt) % 3. The + only picks a partition. The key stays the pair, so (1, 1) and (2, 0) share partition 2 but never match each other.

Count the rows in each new partition.

Partition 0 has 4 rows. Partition 1 has 5 + 1 + 1 = 7 rows. Partition 2 has 5 + 1 + 1 = 7 rows.

Before salting, the sizes were 2, 15 and 1. Now they are 4, 7 and 7. The largest task is less than half the size.

Every order still finds its customer. Order 201 has salt 0, so it looks for (1, 0). That key is in the exploded customers table.

Now check that the answer did not change. Step 1 adds up each (customer_id, salt) key.

step2.show()

The totals match the totals without salting. Salting changes where rows go, not the answer.

Salting is not free. It has three costs.

• The small side grows N times. In our example customers went from 5 rows to 15 rows. A bigger N spreads the key more, but copies the small side more.
• There is more work. A plain join and groupBy become a salt column, an explode, a join on two columns and two aggregates.
• The second aggregate needs a shuffle too. It is small, because it only reads the partial sums.

You can lower the first cost. Salt only the hot key. Give every other key salt 0, and copy only the hot key's small-side row N times.

Then the sizes become 6, 6 and 6 rows.

A plain groupBy with sum often does not need salting. The shuffle chapter showed this: Asha's orders became one partial row, (1, 65.00), before the shuffle.

A join cannot do this.

Four points to keep.

• Skew is one key with far more rows than the rest. After a shuffle, it makes one big partition and one slow task.
• Find it with max against median in the Spark UI, and with a groupBy count for each key.
• Try in order: handle NULL keys, broadcast the small side, let AQE split the partition.
• Salt when those fail. Add a salt to the big side, explode the small side, join on (key, salt), add up twice. The answer stays the same.

Answer these without looking back.

1. In our model, which new partition gets Asha's orders, and how many rows does it hold?
2. Why does adding executors not fix a skewed stage?
3. Which Spark UI columns show skew and spill?
4. After salting with N = 3, why do we copy the customers table 3 times?
5. Asha's partial sums are 80.00, 65.00 and 40.00. What is her total?