95% of My PySpark Job Finished in 4 Minutes. The Last Task Took 40. Here's Why.

I had a PySpark job joining a large transactions table with a customer dimension table. Nothing exotic — a standard join, then an aggregation. On paper, it looked like it should scale fine across the cluster.

In practice, the job would race through most of its tasks and then stall. The Spark UI told the real story: almost every task finished in a few minutes, but one or two tasks ran for over 40 minutes while their executors sat at 80–90% CPU, and the rest of the cluster sat mostly idle waiting for them to finish.

This post walks through what data skew actually is, how I confirmed it was the cause, and the fix that brought the job back under control.

What data skew actually is

Data skew happens when one or a few keys hold a disproportionate share of the data. When Spark distributes work across partitions — usually via hash partitioning on a join or group-by key — all the rows for a given key land in the same partition. If one key has millions of rows and most others have a few thousand, that one partition (and the single task processing it) ends up doing far more work than every other partition combined.

The result is a job where 95% of tasks look completely healthy, and the remaining 5% become the actual bottleneck. Total job time is dictated by the slowest task, not the average one — so a single skewed key can dominate your entire runtime even if it represents a tiny fraction of your total row count.

How I confirmed it

The Spark UI's stage view was the first clue: a chart of task durations with almost every bar clustered together, and one or two bars stretching far beyond the rest. That pattern — uniform short tasks plus one long outlier — is close to a signature for skew.

To confirm which key was responsible, I ran a simple aggregation on the join key before doing anything else:

from pyspark.sql import functions as F

df.groupBy("customer_id") \
  .count() \
  .orderBy(F.desc("count")) \
  .show(20)

The output made it obvious: a small number of customer IDs had row counts orders of magnitude higher than the rest of the dataset. A handful of high-volume accounts were responsible for a disproportionate share of all transactions, and Spark's default hash partitioning sent every one of those rows into a single partition.

The fix: salting

The technique that actually solved this is called salting. The core idea:

1. Take the skewed join key and append a random "salt" value to it, splitting what was one hot key into several smaller synthetic keys.
2. On the smaller side of the join, explode each row into one copy per possible salt value, so every salted variant of the key still has a matching row to join against.
3. Join on the new salted key instead of the original key.
4. Drop the salt column once the join is done, since it was only needed to spread the data across partitions.

This doesn't change the result of the join — every row still matches correctly — but it changes how the work is distributed. What used to be one massive partition handling all of a hot customer's rows gets split into several smaller partitions, each processed by a different executor in parallel.

Here's a simplified version of what I used:

from pyspark.sql.functions import rand, concat_ws, col

SALT_BUCKETS = 10

large_df = large_df.withColumn(
    "salt", (rand() * SALT_BUCKETS).cast("int")
)
large_df = large_df.withColumn(
    "join_key", concat_ws("_", col("customer_id"), col("salt"))
)

salt_values = spark.range(0, SALT_BUCKETS).withColumnRenamed("id", "salt")

small_df_expanded = small_df.crossJoin(salt_values)
small_df_expanded = small_df_expanded.withColumn(
    "join_key", concat_ws("_", col("customer_id"), col("salt"))
)

result_df = large_df.join(
    small_df_expanded,
    on="join_key",
    how="left"
).drop("salt", "join_key")

With SALT_BUCKETS = 10, each previously-hot key gets split across 10 partitions instead of 1. The slowest task in the job dropped dramatically — no more single straggler holding up the entire stage while everything else finished and sat idle.

Things that surprised me

Spark never warns you about skew. There's no error, no red flag in the logs — just one task that quietly runs far longer than the rest while the cluster utilization graph looks strangely uneven. You only notice it if you're actually looking at task-level timing in the Spark UI, not just whether the job succeeded.

Salting isn't free. Adding salt buckets increases the amount of data being shuffled and joined, since the smaller side of the join gets duplicated once per bucket. It's worth applying selectively to the specific hot keys causing the problem, not blindly salting an entire dataset — over-salting can add more overhead than it saves.

Check broadcast joins first. If one side of your join is small enough to fit comfortably in executor memory (a common threshold cited is under roughly 500MB, though this depends on your cluster's spark.sql.autoBroadcastJoinThreshold setting), a broadcast join avoids the shuffle — and the skew problem — entirely. It's simpler than salting and should usually be tried first if it's feasible for your data size.

Not all skew comes from joins. The same problem shows up in groupBy aggregations, window functions partitioned by a skewed column, and even repartitioning by a skewed key. The detection method (checking row counts per key) and the fix (salting, or in some cases using Spark's adaptive query execution skew join optimization) generalize beyond just joins.

A quick detection checklist

If your Spark job "succeeds" but feels slower than it should be for the data volume involved, this is a fast way to check for skew before assuming the cluster just needs more resources:

• Open the Spark UI and look at the stage's task duration distribution — a long tail with one or two extreme outliers is the classic signature.
• Run a groupBy(key).count().orderBy(desc("count")) on your join or aggregation key and look for one or a few values that dominate the row count.
• Check executor CPU utilization during the slow stage — if most executors are idle while one or two are pegged, that's a strong secondary signal.
• If confirmed, evaluate a broadcast join first if feasible, and fall back to salting for genuinely large-to-large joins where broadcasting isn't an option.

The real lesson

"My job finished" and "my job finished efficiently" are two very different claims. A completed Spark job can still be hiding a massive inefficiency in plain sight, and the only way to catch it is to look past the green checkmark and into task-level timing.

Adding more cluster resources would not have fixed this — the bottleneck wasn't total compute capacity, it was that all the work for the hottest keys was funneled into a single partition no matter how many executors were available. Understanding why a job is slow matters more than just throwing more hardware at it.

Have you run into a skewed join in production? What tipped you off — the Spark UI, a cost spike, or something else?