Sev1: Spark job stuck at 199/200 tasks for hours — finding and fixing data skew

#Symptoms

• A stage shows all tasks finished except one or a few, which run far longer than the rest.
• Executors sit idle; the job misses its SLA.
• Sometimes the long task fails with OOM or FetchFailedException and retries.

#1. Confirm skew in the Spark UI

Open the stuck stage and look at Summary Metrics for Completed Tasks:

If durations are similar but all slow, it is not skew; look at cluster size or the source instead.

#2. Find the hot keys

from pyspark.sql import functions as F

(df_orders.groupBy("customer_id")
   .count()
   .orderBy(F.desc("count"))
   .show(20, truncate=False))

Common surprises: NULL keys, a default value like 0 or 'UNKNOWN', or one giant tenant.

#3. Quick win: let AQE handle it

Adaptive Query Execution splits skewed partitions in sort-merge joins:

spark.conf.set("spark.sql.adaptive.enabled", "true")
spark.conf.set("spark.sql.adaptive.skewJoin.enabled", "true")
spark.conf.set("spark.sql.adaptive.skewJoin.skewedPartitionFactor", "5")
spark.conf.set("spark.sql.adaptive.skewJoin.skewedPartitionThresholdInBytes", "256m")

AQE is on by default in recent Spark versions, but the skew thresholds may not catch your case. It only applies to joins, not to groupBy aggregations.

#4. Fix NULL or junk keys

If the hot key is NULL, those rows can never match in an inner join. Filter them out before the join, or handle them separately:

left_valid = df_orders.filter(F.col("customer_id").isNotNull())
left_null  = df_orders.filter(F.col("customer_id").isNull())
joined = left_valid.join(df_customers, "customer_id", "left").unionByName(left_null, allowMissingColumns=True)

#5. Broadcast the small side

If one side is small enough to fit in executor memory, broadcast it and avoid the shuffle entirely:

joined = df_orders.join(F.broadcast(df_customers), "customer_id", "left")

Warning

Broadcasting a table that is larger than expected causes driver or executor OOM. Check the size first and keep it to tens or low hundreds of MB.

#6. Salt the hot key

For a large-to-large join with a few hot keys:

SALT = 16
orders_s = df_orders.withColumn("salt", (F.rand() * SALT).cast("int"))
cust_s = df_customers.crossJoin(spark.range(SALT).withColumnRenamed("id", "salt"))
joined = orders_s.join(cust_s, ["customer_id", "salt"]).drop("salt")

For aggregations, aggregate in two steps: first by key plus salt, then by key.

#7. Get today's run done

• Kill the stuck run, apply the AQE settings or the broadcast fix (smallest code change), and rerun.
• If the job writes with overwrite semantics, check the partial output was not committed before rerunning; with Delta, failed writes don't commit.

#Prevent it next time

• Add a data quality check for key distribution (top key share) on critical inputs.
• Keep AQE on and tune skew thresholds for large jobs.
• Alert on job duration above P95, not just on failure.
• Document known hot keys and their handling in the job's code.