A GroupBy Aggregation Hangs at 799/800 Tasks

1. Root cause

This is single-key skew, and the specific mechanism matters: NULL is being treated as a valid device_id for grouping purposes, so every event that failed device attribution — 15% of a 120GB table, which is a very large absolute number of rows — collapses into one group. The task handling the device_id = NULL partition has to shuffle-read roughly 340M records, ~570x the median task's 600K, which exactly matches the observed task-duration outlier. This is a different flavor of skew than a single "hot" real key (like a popular product) — it's a sentinel/missing-value key absorbing everything that didn't get a real value, which is extremely common in production event pipelines and worth naming as its own category in an interview answer.

2. Is salting appropriate, and the fix

Salting is appropriate here if the NULL group's aggregate value is actually needed downstream (e.g., "unattributed watch time" is a real metric someone consumes). sum is associative and decomposable, so a two-pass salted aggregation is a correct fix:

SALT_BUCKETS = 50  # NULL key needs more buckets than a typical hot key given its size

salted = events.withColumn(
    "salt",
    F.when(F.col("device_id").isNull(), (F.rand() * SALT_BUCKETS).cast("int"))
     .otherwise(F.lit(0))  # non-null keys don't need salting, keep them as one bucket
)

partial = (
    salted
    .groupBy("device_id", "salt")
    .agg(F.sum("watch_seconds").alias("partial_sum"))
)

result = (
    partial
    .groupBy("device_id")
    .agg(F.sum("partial_sum").alias("total_watch_seconds"))
)

Only salting the NULL key (rather than every key) keeps the fix targeted — salting every key would add unnecessary shuffle overhead to the 799 already-fast tasks for no benefit, since they weren't skewed to begin with. The second-stage shuffle here is cheap: it's shuffling num_real_keys + SALT_BUCKETS rows, not the original 120GB.

3. A simpler fix given the NULL-specific context

Yes — if unattributed watch time doesn't need to be broken out by any dimension (it's just "total unattributed," a single number), the simplest fix is to filter device_id IS NOT NULL before the aggregation and compute the NULL-group total separately with a trivial filter().agg(F.sum(...)) (a full-table scan and a simple sum, no per-key grouping, so no skew is possible — there's only one group). This avoids salting's complexity entirely for this specific case:

real_device_totals = (
    events.filter(F.col("device_id").isNotNull())
    .groupBy("device_id")
    .agg(F.sum("watch_seconds").alias("total_watch_seconds"))
)
unattributed_total = (
    events.filter(F.col("device_id").isNull())
    .agg(F.sum("watch_seconds").alias("total_watch_seconds"))
    .withColumn("device_id", F.lit(None))
)
result = real_device_totals.unionByName(unattributed_total)

I'd use this instead of salting for this specific case — it's simpler and directly exploits the fact that the skewed group doesn't need further breakdown. I'd still keep AQE's automatic skew-join handling enabled as a general safety net for other skew that might show up in joins elsewhere in the pipeline (a real hot device, for instance), since that's a zero-cost default that doesn't require knowing about every possible skewed key in advance the way this targeted fix does.