When to use AQE skew handling vs manual salting | Data Engineering

Q: How does two-stage aggregation handle groupBy skew?

For groupBy skew, salting uses a twostage approach. In stage one, you aggregate per salted key so the hot key's rows are spread across multiple partitions. In stage two, you reaggregate across the salts to get the true total. package com.bank.spark; import org.apache.spark.sql.Dataset; import org.ap

Q: What is isolate-and-broadcast?

Isolateandbroadcast splits the join into two branches. You filter the large table to just the hot keys and join them with a broadcast of the matching smallside rows, which avoids shuffling the huge hot partition. The remaining cold keys go through a normal join. You then union the two results. packa

Definition

Try AQE skew handling first because it is automatic and code-free; use salting or isolate-and-broadcast when AQE cannot cope with both-sides skew, groupBy skew, or when you need manual control.

AQE skew join – Adaptive Query Execution detects oversized shuffle partitions at runtime and splits them into smaller sub-partitions automatically, with no code change required.
Manual salting – You append a random salt to the hot key and replicate the other side, giving precise control when AQE thresholds are insufficient.
Two-stage aggregation – For groupBy skew, you aggregate per salted key first, then re-aggregate across salts to get the true total.
Isolate-and-broadcast – You filter out the few hot keys, join them separately with a broadcast join, and union the result with the cold-key join.
Decision order – Enable AQE first, check the Spark UI for stragglers, tune thresholds, and only then add intrusive manual code.

At the National Bank of India, AQE skew handling is like an automatic supervisor who notices one clerk is buried under a pile of transaction slips and quietly splits that pile among several helpers with no instruction change needed. You only step in manually, with salting or by pulling the mega-account aside for special handling, when the automatic supervisor cannot cope: when the overload exists on both lists at once, or when the overload is in a totalling task rather than a matching task.

Understanding the Concepts

What is AQE skew join handling?

Adaptive Query Execution is a Spark 3.x feature that optimizes query plans at runtime using real shuffle statistics. When its skew-join setting is enabled, Spark measures the actual size of each shuffle partition at the stage boundary. If a partition exceeds a configurable factor times the median partition size and an absolute byte threshold, AQE splits that partition into smaller sub-partitions and replicates the matching partition on the other side so the split pieces still find their matches.

JAVA

// Enable AQE skew join in a Spark banking pipeline
package com.bank.spark;

import org.apache.spark.sql.SparkSession;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;

public class AqeSkewJoinDemo {
    public static void main(String[] args) {
        SparkSession spark = SparkSession.builder()
            .appName("NBI-AQE-Skew-Join")
            .config("spark.sql.adaptive.enabled", "true")
            .config("spark.sql.adaptive.skewJoin.enabled", "true")
            .config("spark.sql.adaptive.skewJoin.skewedPartitionFactor", "5")
            .config("spark.sql.adaptive.skewJoin.skewedPartitionThresholdInBytes", "256m")
            .getOrCreate();

        Dataset<Row> transactions = spark.read().parquet("/data/bank/transactions");
        Dataset<Row> accounts = spark.read().parquet("/data/bank/accounts");

        // AQE will automatically split the skewed partition
        Dataset<Row> enriched = transactions.join(accounts, "account_id");
        enriched.write().parquet("/data/bank/enriched_transactions");
    }
}

TEXT

Expected behavior:
AQE detects that account_id = 'POOLING_ACCOUNT_001' has 50x more rows
than the median partition. It splits that partition into 10 sub-partitions
and replicates the matching accounts rows so each sub-partition finds its match.

Why is AQE the first thing to try?

AQE is the first choice for three reasons. It requires no code change, so you do not introduce maintenance burden. It uses real runtime statistics rather than catalog estimates, so it catches skew that only appears as data grows. It handles the common case of a single skewed join side automatically.

Enable it with two configuration lines and Spark handles the rest.
It works for sort-merge joins and shuffled hash joins.
It adapts to changing data distributions without any redeployment.

JAVA

// Minimal configuration to enable AQE skew join
SparkSession spark = SparkSession.builder()
    .config("spark.sql.adaptive.enabled", "true")
    .config("spark.sql.adaptive.skewJoin.enabled", "true")
    .getOrCreate();
// No code change to the join itself

When does AQE skew handling fall short?

AQE has three clear limitations.

First, when both sides of the join are heavily skewed on the same key, AQE splits one side and replicates the other, but if the other side is also huge for that key, the replication itself can blow up memory.

Second, AQE skew-join logic only applies to joins. If the skew is in a groupBy or other aggregation, AQE does not split the hot partition because it is not a join optimization.

Third, AQE decisions are driven by fixed thresholds. An unusual distribution may fall outside what the thresholds catch, leaving residual skew that still causes stragglers.

TEXT

Scenario: groupBy skew at National Bank of India

account_id    | transaction_count
POOLING_001   | 50,000,000    <-- AQE cannot help here
ACCT_002      | 1,200
ACCT_003      | 980
...

The groupBy aggregation sends all 50M rows for POOLING_001
to one partition. AQE skew-join logic does not apply.

What is manual salting and how does it work?

Salting appends a random number to the hot key to spread it across multiple partitions. On the other side of the join, you replicate each row with every possible salt value so the salted keys still find matches.

JAVA

package com.bank.spark;

import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SparkSession;
import static org.apache.spark.sql.functions.*;

public class SaltingJoinDemo {
    public static void main(String[] args) {
        SparkSession spark = SparkSession.builder()
            .appName("NBI-Salting-Join").getOrCreate();

        int saltBuckets = 10;
        Dataset<Row> transactions = spark.read().parquet("/data/bank/transactions")
            .withColumn("salt", expr("CAST(FLOOR(RAND() * " + saltBuckets + ") AS INT)"))
            .withColumn("salted_key", concat(col("account_id"), lit("_"), col("salt")));

        // Explode the small side to match every salt
        Dataset<Row> accounts = spark.read().parquet("/data/bank/accounts");
        Dataset<Row> accountsExploded = accounts
            .crossJoin(spark.range(saltBuckets).withColumnRenamed("id", "salt"))
            .withColumn("salted_key", concat(col("account_id"), lit("_"), col("salt")));

        Dataset<Row> joined = transactions.join(accountsExploded, "salted_key");
        joined.drop("salt", "salted_key").write().parquet("/data/bank/salted_result");
    }
}

How does two-stage aggregation handle groupBy skew?

For groupBy skew, salting uses a two-stage approach. In stage one, you aggregate per salted key so the hot key's rows are spread across multiple partitions. In stage two, you re-aggregate across the salts to get the true total.

JAVA

package com.bank.spark;

import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SparkSession;
import static org.apache.spark.sql.functions.*;

public class TwoStageAggregationDemo {
    public static void main(String[] args) {
        SparkSession spark = SparkSession.builder()
            .appName("NBI-TwoStage-Agg").getOrCreate();

        int saltBuckets = 10;
        Dataset<Row> transactions = spark.read().parquet("/data/bank/transactions")
            .withColumn("salt", expr("CAST(FLOOR(RAND() * " + saltBuckets + ") AS INT)"));

        // Stage 1: partial aggregate per salted key
        Dataset<Row> partial = transactions
            .groupBy("account_id", "salt")
            .agg(sum("amount").alias("partial_sum"),
                 count("*").alias("partial_count"));

        // Stage 2: final aggregate across salts
        Dataset<Row> result = partial
            .groupBy("account_id")
            .agg(sum("partial_sum").alias("total_amount"),
                 sum("partial_count").alias("total_count"));

        result.write().parquet("/data/bank/account_totals");
    }
}

TEXT

Expected output:
account_id       | total_amount    | total_count
POOLING_001      | 5,000,000,000   | 50,000,000
ACCT_002         | 1,440,000       | 1,200

What is isolate-and-broadcast?

Isolate-and-broadcast splits the join into two branches. You filter the large table to just the hot keys and join them with a broadcast of the matching small-side rows, which avoids shuffling the huge hot partition. The remaining cold keys go through a normal join. You then union the two results.

JAVA

package com.bank.spark;

import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SparkSession;
import static org.apache.spark.sql.functions.*;
import java.util.Arrays;
import java.util.List;

public class IsolateBroadcastDemo {
    public static void main(String[] args) {
        SparkSession spark = SparkSession.builder()
            .appName("NBI-Isolate-Broadcast").getOrCreate();

        List<String> hotKeys = Arrays.asList("POOLING_001", "CORPORATE_002");

        Dataset<Row> transactions = spark.read().parquet("/data/bank/transactions");
        Dataset<Row> accounts = spark.read().parquet("/data/bank/accounts");

        // Branch 1: hot keys with broadcast
        Dataset<Row> hotTxns = transactions.filter(col("account_id").isin(hotKeys.toArray()));
        Dataset<Row> hotAccounts = accounts.filter(col("account_id").isin(hotKeys.toArray()));
        Dataset<Row> hotResult = hotTxns.join(broadcast(hotAccounts), "account_id");

        // Branch 2: cold keys with normal join
        Dataset<Row> coldTxns = transactions.filter(not(col("account_id").isin(hotKeys.toArray())));
        Dataset<Row> coldResult = coldTxns.join(accounts, "account_id");

        // Union both
        hotResult.unionByName(coldResult)
            .write().parquet("/data/bank/enriched_transactions");
    }
}

How do you compare AQE vs salting vs isolate-and-broadcast?

| Feature | AQE Skew Join | Manual Salting | Isolate-and-Broadcast |
|---|---|---|---|
| Code change required | No | Yes | Yes |
| Handles join skew | Yes | Yes | Yes |
| Handles groupBy skew | No | Yes (two-stage) | No |
| Both-sides skewed | Limited | Yes | Partial |
| Hot keys must be known | No | No | Yes |
| Runtime adaptive | Yes | No | No |
| Maintenance cost | None | Medium | Medium |

What is the practical decision workflow?

The workflow at the National Bank of India follows a clear escalation path.

Enable AQE and its skew-join setting. Let it handle the skew automatically.
Run the job and inspect the Spark UI. Check whether stragglers remain on any stage.
If stragglers persist, profile the data to identify the hot keys and their distribution.
Tune AQE thresholds before resorting to code. Lower the skewed-partition factor or the byte threshold.
If AQE still cannot cope, choose the manual technique that fits the symptom: salting for groupBy skew or both-sides skew, isolate-and-broadcast when the hot keys are few and known.
Combine techniques when needed. For example, use AQE for most joins and add salting for one specific groupBy.

DIAGRAM

Decision tree:

  Is skew in a join?
  ├── Yes
  │   ├── Enable AQE skew join
  │   ├── Stragglers remain?
  │   │   ├── No  → Done
  │   │   └── Yes
  │   │       ├── Both sides skewed? → Salting
  │   │       ├── Few known hot keys? → Isolate-and-broadcast
  │   │       └── Otherwise → Tune AQE thresholds, then salt
  └── No (groupBy / aggregation)
      └── Two-stage salted aggregation

What should a senior engineer know?

A senior engineer should know the decision order and be able to justify each step. The goal is to prefer the zero-code automatic path and only add intrusive, maintained code when the data genuinely requires it.

AQE skew join works together with other AQE features like coalescing post-shuffle partitions and switching join strategies at runtime.
Salting introduces data replication on the small side, which costs memory and shuffle bytes, so the salt bucket count must be tuned to the skew ratio.
Isolate-and-broadcast assumes the hot keys are few and the small-side data for those keys fits in memory for a broadcast.
In production, monitor for new hot keys that appear over time. AQE handles this automatically, but manual techniques need updating.
Always verify the fix in the Spark UI: the straggler task duration should drop to roughly the median after the skew is resolved.

JAVA

// Verify skew resolution by checking task metrics
// In Spark UI: Stages tab -> look at task duration distribution
// Before fix: max task = 45 min, median = 30 sec
// After fix:  max task = 35 sec, median = 30 sec

Visual Overview

DIAGRAM

  AQE Skew Join – Automatic Split at Runtime

  ┌─────────────────────────────────────────────────┐
  │  Shuffle Stage Boundary                          │
  │                                                  │
  │  Partition sizes (before AQE):                   │
  │  [P1: 50MB] [P2: 45MB] [P3: 2GB] [P4: 48MB]    │
  │                          ▲ skewed                │
  │                                                  │
  │  AQE detects: P3 >> median * factor              │
  │                                                  │
  │  After split:                                    │
  │  [P1: 50MB] [P2: 45MB] [P3a: 500MB] [P3b: 500MB]│
  │  [P3c: 500MB] [P3d: 500MB] [P4: 48MB]           │
  │                                                  │
  │  Other side replicated for P3a..P3d              │
  └─────────────────────────────────────────────────┘

DIAGRAM

  Manual Salting – Two-Stage Aggregation

  Stage 1: Partial aggregate per salted key
  ┌──────────────────────────────────────────┐
  │  POOLING_001_0  → partial_sum = 500M     │
  │  POOLING_001_1  → partial_sum = 480M     │
  │  POOLING_001_2  → partial_sum = 520M     │
  │  ...                                     │
  │  POOLING_001_9  → partial_sum = 510M     │
  └──────────────────────────────────────────┘
            │
            ▼
  Stage 2: Final aggregate across salts
  ┌──────────────────────────────────────────┐
  │  POOLING_001    → total_sum = 5,000M     │
  └──────────────────────────────────────────┘

DIAGRAM

  Isolate-and-Broadcast

  ┌──────────────────┐
  │  All Transactions │
  └────────┬─────────┘
           │
     ┌─────┴─────┐
     ▼           ▼
  Hot Keys    Cold Keys
     │           │
     ▼           ▼
  Broadcast   Normal
  Join        Shuffle Join
     │           │
     └─────┬─────┘
           ▼
      UNION ALL

Key Takeaway

When facing data skew in Spark, the first move is always Adaptive Query Execution's skew join optimization. With AQE enabled, Spark measures the actual size of each shuffle partition at runtime and automatically splits oversized partitions into smaller sub-partitions, replicating the matching side so the splits still find their joins. This requires no code change, uses real statistics rather than estimates, and handles the common case of a single skewed join side very well.

You escalate to manual techniques only when AQE is insufficient. For groupBy skew, AQE does not apply because its logic is join-specific, so you use salting with a two-stage aggregation: aggregate per salted key in stage one, then re-aggregate across salts for the true total. When both sides are skewed on the same key, salting with careful control is safer than AQE's split-and-replicate approach. When there are a few known hot keys, isolate-and-broadcast removes them from the shuffle path entirely by joining them separately with a broadcast join.

At the National Bank of India this was the exact playbook. A transactions-to-accounts join skewed on one pooling account was resolved just by enabling AQE, with no code change. A per-account groupBy skewed by the same account needed salting with a two-stage aggregation. A join skewed on a handful of corporate mega-accounts was cleanest with isolate-and-broadcast. The workflow is to enable AQE first, confirm in the Spark UI whether stragglers remain, tune thresholds as a next step, and only then add manual code.

Remember one sentence: try AQE skew join first because it is automatic, and fall back to salting or isolate-and-broadcast only when AQE cannot handle groupBy skew, both-sides skew, or when you need precise manual control.

---