How AQE handles data skew (and when you still need salting)

AQE's skew join optimization works in two steps.

• Detection – A shuffle partition is flagged as skewed if its size exceeds both skewedPartitionFactor times the median partition size (default 5x) and skewedPartitionThresholdInBytes (default 256 MB). Requiring both prevents false positives.
• Resolution – The skewed partition is split into several sub-partitions, and the matching rows from the other join side are replicated into each, so they join independently in parallel.
• Limitation – AQE targets joins. Heavily skewed aggregations, window operations, or a single key so massive that sub-partitions are still too large still need manual salting.

Salting appends a random suffix to the key on the fact side and expands the dimension side to match.

What is data skew?

Data skew is when the values of a join or grouping key are distributed very unevenly. After the shuffle a join requires, all rows for a key land in one partition, so a hot key produces one giant partition while others are small. The task on that giant partition runs long after all others complete: a straggler.

# Diagnose skew: check partition sizes after a shuffle
from pyspark.sql import functions as F

txns = spark.read.parquet("/data/bank/transactions")

# Count rows per account to identify hot keys
txns.groupBy("account_id").count() \
    .orderBy(F.col("count").desc()) \
    .show(10)

account_id     count
CORP-MEGA-001  5000000   ← hot key (50x more than median)
CORP-MEGA-002  3000000
ACC-1001       100000
ACC-1002       95000
...            ~100000 (median)

How does AQE detect skew?

AQE detects skew at the shuffle boundary using two thresholds that must both be met. This avoids false positives on small datasets.

# AQE skew detection thresholds
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")

Detection logic:
  Partition is skewed if:
    size > skewedPartitionFactor * median_partition_size
    AND
    size > skewedPartitionThresholdInBytes

Example:
  Median partition: 50 MB
  Partition for CORP-MEGA-001: 2 GB

  2 GB > 5 * 50 MB (250 MB)?  YES
  2 GB > 256 MB?               YES
  → Flagged as skewed

How does AQE resolve skew?

Once flagged, AQE splits the skewed partition into several sub-partitions along the skewed side. It then replicates the matching rows from the other join side into each sub-partition, so each joins independently as its own task.

Before AQE skew handling:
  Partition 0: 50 MB   → Task 0 (fast)
  Partition 1: 2 GB    → Task 1 (STRAGGLER - 40x slower)
  Partition 2: 45 MB   → Task 2 (fast)

After AQE skew handling:
  Partition 0:    50 MB  → Task 0
  Partition 1a: 500 MB  → Task 1a  (split from 2 GB)
  Partition 1b: 500 MB  → Task 1b  (other side replicated)
  Partition 1c: 500 MB  → Task 1c
  Partition 1d: 500 MB  → Task 1d
  Partition 2:    45 MB  → Task 2

Result: no straggler, all tasks finish around the same time

# AQE handles this automatically
result = txns.join(
    spark.read.parquet("/data/bank/accounts"),
    "account_id"
)
result.show()
# Check Spark UI: AQE shows "SkewJoin" in the executed plan

When is AQE not enough and you need salting?

Manual salting is required in three cases: when a single key is so dominant that even split sub-partitions are too large, when the skewed operation is not a join, and when you need to control skew before the shuffle.

from pyspark.sql import functions as F
import random

NUM_SALT_BUCKETS = 10

# Salt the fact side: append random suffix to hot keys
txns_salted = txns.withColumn(
    "salted_account_id",
    F.concat(
        F.col("account_id"),
        F.lit("_"),
        (F.rand() * NUM_SALT_BUCKETS).cast("int").cast("string")
    )
)

# Explode the dimension side: one copy per salt bucket
accounts = spark.read.parquet("/data/bank/accounts")
salt_df = spark.range(NUM_SALT_BUCKETS).withColumnRenamed("id", "salt")

accounts_exploded = accounts.crossJoin(salt_df).withColumn(
    "salted_account_id",
    F.concat(
        F.col("account_id"),
        F.lit("_"),
        F.col("salt").cast("string")
    )
)

# Join on salted key: hot key is now spread across 10 partitions
result = txns_salted.join(
    accounts_exploded,
    "salted_account_id"
).drop("salted_account_id", "salt")

result.show()

How does salting work for aggregations?

AQE's skew handling targets joins. For heavily skewed aggregations, use a two-stage approach: partial aggregate with salt, then final aggregate without.

# Skewed aggregation: one account has 5M rows
# Two-stage salted aggregation

# Stage 1: partial aggregation with salt
partial = txns.withColumn(
    "salt", (F.rand() * 10).cast("int")
).groupBy("account_id", "salt").agg(
    F.sum("amount").alias("partial_sum"),
    F.count("*").alias("partial_count")
)

# Stage 2: final aggregation without salt
final = partial.groupBy("account_id").agg(
    F.sum("partial_sum").alias("total_amount"),
    F.sum("partial_count").alias("txn_count")
)

final.show()

How do you diagnose skew in the Spark UI?

Look for stages where one task takes much longer than the median. The Spark UI's task metrics show min, median, and max duration.

Spark UI > Stages > Stage 5 (shuffle)

Task Metrics:
  Duration:  Min=5s  Median=8s  Max=320s  ← straggler!
  Input:     Min=40MB  Median=50MB  Max=2GB

This tells you:
  - One partition is 2 GB (40x the median)
  - The task on it takes 320s (40x the median)
  - This is data skew

What are the configuration parameters?

| Parameter | Default | Purpose |
|---|---|---|
| skewJoin.enabled | true | Enable AQE skew join handling |
| skewedPartitionFactor | 5 | Times median to flag as skewed |
| skewedPartitionThresholdInBytes | 256 MB | Minimum absolute size to flag |

# Tune for the bank's data profile
spark.conf.set("spark.sql.adaptive.skewJoin.skewedPartitionFactor", "3")
spark.conf.set("spark.sql.adaptive.skewJoin.skewedPartitionThresholdInBytes", "128m")

What should a senior engineer know?

• AQE's skew handling is join-specific: it splits the skewed side and replicates the other side.
• Salting is the general-purpose tool for any skewed operation: joins, aggregations, or windows.
• Salting has a cost: the dimension side is exploded by the salt bucket count, increasing shuffle volume.
• For the National Bank of India, AQE handled the everyday moderate skew of busy branches automatically, but mega-corporate accounts with extreme volume were salted.
• Always diagnose skew first (Spark UI task metrics) before applying fixes blindly.