Data skew in distributed joins with salting and broadcast joins: an interactive simulator

Distributed joins and aggregations send every row with the same key to the same task. When one key is far more popular than the rest, a viral ad, a celebrity account, a default value, the task that gets it does a large share of all the work while the others sit idle, and the job takes as long as that one task. Adding tasks does not help, because one key cannot be split by hashing. Salting splits it on purpose; a broadcast join avoids shuffling by key at all.

The playground joins 100 million ad clicks with the ads table, where clicks follow a power law and one viral ad takes 1%, 10% or 30% of them. Each task processes 100,000 rows a second.

What each strategy does

• A plain hash join is as slow as its busiest task. With the viral ad at 10% of clicks, a typical task finishes in under 4 s and the one holding the viral ad takes 104 s. Even at 1% the job is held up by the most popular ads, 38 s against a few seconds typical.
• More tasks make the typical task faster and leave the slowest one where it was: 1,000 tasks still take 101 s.
• Salting adds a random suffix to the hot keys, so their rows spread over many tasks, and copies the matching ad rows to each of them. The job drops to 16 s. It needs to know which keys are hot, from a sample or from statistics.
• A broadcast join copies the whole small table to every machine and joins each slice of clicks where it already is: no shuffle of the clicks, no skew, 5.5 s, and 1 GB over the network instead of 10. It only works when one side fits in memory; a 500 GB table cannot be broadcast.

In Spark

-- Spark SQL: broadcast the small side, no shuffle of the big one
SELECT /*+ BROADCAST(ads) */ c.*, a.campaign
FROM clicks c JOIN ads a ON c.ad_id = a.ad_id;

-- or let adaptive execution split skewed partitions at run time
SET spark.sql.adaptive.enabled = true;
SET spark.sql.adaptive.skewJoin.enabled = true;

-- salting by hand: spread each hot key over 32 sub-keys
clicks_salted = clicks.withColumn("salt", (rand() * 32).cast("int"))
ads_salted    = ads.crossJoin(spark.range(32).withColumnRenamed("id", "salt"))
joined        = clicks_salted.join(ads_salted, ["ad_id", "salt"])

In a system design interview

When a pipeline aggregates or joins by a key with a long tail, say what happens to the head: “Clicks per ad follow a power law, and a viral ad can be a large share of a day. We join with the ads table as a broadcast join, since it is small, and for the aggregation we pre-aggregate on the map side so a hot key sends one partial count per task instead of every row. If a large-to-large join is skewed, we salt the top keys found by sampling.” Expect the follow-ups: how to detect hot keys automatically, and how the same skew shows up in streaming as one hot partition.

Read the guide to MapReduce and Spark or hot keys and skew, see hot keys and parallel transcoding, or watch a strong candidate design an ad click aggregator.

Questions people ask