Definition
- Isolate-and-broadcast – Filter out the few hot keys, join them with a broadcast of the small-side rows, and union with the normal cold-key join.
- Increase parallelism – Raise
spark.sql.shuffle.partitionsso hashing spreads moderately-hot keys across more buckets, reducing collisions. - Pre-aggregation – Aggregate on the map side before the shuffle so each partition sends one partial result per key instead of millions of raw rows.
- Different join strategy – Use a broadcast hash join when the small side fits in memory to avoid shuffling the large skewed side entirely.
- Data model redesign – Split a structurally hot key into sub-keys at the source so no single key dominates, which is the most durable fix.
At the National Bank of India one giant pooling account overwhelms a single clerk. You can pull that one account aside and handle it specially (isolate and broadcast), hire more clerks so the pile is cut into more sub-piles (increase parallelism), have each branch total its own slips before sending them in (pre-aggregate), route the work so the clerk never touches the big ledger (different join strategy), or restructure the books so the one giant account becomes many smaller sub-accounts (redesign the data model).
Understanding the Concepts
What is isolate-and-broadcast?
Isolate-and-broadcast is the most surgical technique when the hot keys are few and known. You split the join into two branches. First you filter the large table to just the hot keys and join that slice with a broadcast of the matching small-side rows. Then you filter to the cold keys and run a normal shuffle join. Finally you union the two results.
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 IsolateBroadcastSkew {
public static void main(String[] args) {
SparkSession spark = SparkSession.builder()
.appName("NBI-Isolate-Broadcast").getOrCreate();
List<String> hotKeys = Arrays.asList("POOLING_001", "SUSPENSE_002");
Dataset<Row> transactions = spark.read().parquet("/data/bank/transactions");
Dataset<Row> accounts = spark.read().parquet("/data/bank/accounts");
// Hot path: broadcast the small side for hot keys
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");
// Cold path: normal shuffle join
Dataset<Row> coldTxns = transactions.filter(not(col("account_id").isin(hotKeys.toArray())));
Dataset<Row> coldResult = coldTxns.join(accounts, "account_id");
// Combine
hotResult.unionByName(coldResult)
.write().parquet("/data/bank/enriched");
}
}
Before: straggler task on POOLING_001 = 45 minutes
After: hot path broadcast = 30 seconds, cold path median = 30 seconds
Why does increasing parallelism help and when does it not?
Raising spark.sql.shuffle.partitions increases the number of hash buckets, so moderately-hot keys that collide into the same partition are spread across more partitions. This helps when many keys are somewhat hot but no single key is a true mega-key.
It does not help a single mega-key because one key always hashes to exactly one partition regardless of partition count. If POOLING_001 has 50 million rows, raising partitions from 200 to 2000 still sends all 50 million to one partition.
package com.bank.spark;
import org.apache.spark.sql.SparkSession;
public class IncreaseParallelismDemo {
public static void main(String[] args) {
SparkSession spark = SparkSession.builder()
.appName("NBI-Parallelism")
.config("spark.sql.shuffle.partitions", "2000")
.getOrCreate();
// With 2000 partitions, moderately-hot merchant keys
// that collided at 200 partitions are now spread out.
// But POOLING_001 still goes to exactly one partition.
spark.read().parquet("/data/bank/merchant_transactions")
.groupBy("merchant_id")
.count()
.write().parquet("/data/bank/merchant_counts");
}
}
What is pre-aggregation and how does it reduce skew?
Pre-aggregation shrinks the data before it crosses the shuffle boundary. If you are joining transactions to accounts only to count per account, do the count first so each partition sends one partial count per key instead of millions of raw rows.
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 PreAggregationDemo {
public static void main(String[] args) {
SparkSession spark = SparkSession.builder()
.appName("NBI-PreAgg").getOrCreate();
Dataset<Row> transactions = spark.read().parquet("/data/bank/transactions");
// Pre-aggregate: reduce 50M rows for POOLING_001 to one row
Dataset<Row> preAgg = transactions
.groupBy("account_id")
.agg(sum("amount").alias("total_amount"),
count("*").alias("txn_count"));
// Now join with accounts: the hot key has only 1 row
Dataset<Row> accounts = spark.read().parquet("/data/bank/accounts");
Dataset<Row> result = preAgg.join(accounts, "account_id");
result.write().parquet("/data/bank/account_summary");
}
}
Before pre-aggregation:
POOLING_001 sends 50,000,000 rows across the shuffle
After pre-aggregation:
POOLING_001 sends 1 row (total_amount, txn_count) across the shuffle
How does changing the join strategy sidestep skew?
Only shuffle-based joins expose skew, because they hash keys to partitions. A broadcast hash join avoids shuffling the large side entirely by broadcasting the small side to all executors, so no partition becomes a straggler.
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 BroadcastJoinDemo {
public static void main(String[] args) {
SparkSession spark = SparkSession.builder()
.appName("NBI-Broadcast-Join")
.config("spark.sql.autoBroadcastJoinThreshold", "100m")
.getOrCreate();
Dataset<Row> transactions = spark.read().parquet("/data/bank/transactions");
Dataset<Row> branches = spark.read().parquet("/data/bank/branches"); // small table
// Broadcast the small branches table to avoid shuffle
Dataset<Row> enriched = transactions.join(
broadcast(branches), "branch_id");
enriched.write().parquet("/data/bank/enriched_with_branch");
}
}
What does redesigning the data model look like?
Redesigning the data model is the deepest fix. If a single pooling or suspense account is structurally hot, you split it into several sub-accounts at the source so no single key dominates.
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 DataModelRedesignDemo {
public static void main(String[] args) {
SparkSession spark = SparkSession.builder()
.appName("NBI-Model-Redesign").getOrCreate();
// Instead of one POOLING_001 account, the source system
// now writes to POOLING_001_00 through POOLING_001_09
Dataset<Row> transactions = spark.read().parquet("/data/bank/transactions");
// The hot key is now 10 separate keys, each with 1/10 the data
Dataset<Row> accounts = spark.read().parquet("/data/bank/accounts");
Dataset<Row> result = transactions.join(accounts, "sub_account_id");
result.write().parquet("/data/bank/result");
}
}
Before redesign: POOLING_001 = 50,000,000 rows (one partition)
After redesign: POOLING_001_00..09 = 5,000,000 rows each (10 partitions)
How do you compare these techniques?
| Technique | Best For | Fixes Mega-Key | Code Change | Cost |
|---|---|---|---|---|
| Isolate-and-broadcast | Few known hot keys | Yes | Moderate | Low |
| Increase parallelism | Many moderately-hot keys | No | Config only | Low |
| Pre-aggregation | Downstream needs aggregate | Yes | Moderate | Low |
| Broadcast join | Small side fits in memory | Yes | Minimal | Low |
| Data model redesign | Structural hot key | Yes | High | High |
What is the practical decision order?
Start with the cheapest and most surgical technique that fits the symptom.
- Check if a broadcast join is possible. If the small side fits in memory, it eliminates skew with no shuffle.
- Try pre-aggregation. If you only need an aggregate downstream, shrink the data before it crosses the shuffle.
- Try increasing parallelism. If many keys are moderately hot, more partitions spread collisions.
- Use isolate-and-broadcast for a few known hot keys that do not respond to the above.
- Reserve data model redesign for a hot key that keeps causing incidents across many jobs.
Decision order: 1. Broadcast join possible? → Use it (cheapest) 2. Only need aggregate? → Pre-aggregate 3. Many moderately-hot keys? → Increase parallelism 4. Few known mega-keys? → Isolate-and-broadcast 5. Structural recurring problem? → Redesign data model
What should a senior engineer know?
A senior engineer combines these techniques deliberately. At the National Bank of India, a single job might use a broadcast join for the branch dimension, pre-aggregation for the merchant summary, and isolate-and-broadcast for the one corporate pooling account.
- Model redesign is the most durable fix but the most expensive to implement, so you reach for it when the same hot key causes incidents across many jobs.
- Monitor for new hot keys that appear over time. A key that was cold last quarter may become hot after a business change.
- Always verify in the Spark UI that the straggler task duration drops to roughly the median after applying the fix.
- Combine techniques rather than relying on a single silver bullet.
// Production monitoring: check for skew in task duration
// Spark UI -> Stages -> Task Duration
// Goal: max_task_duration / median_task_duration < 3x
Visual Overview
Skew Technique Decision Tree
┌───────────────────────────┐
│ Detected skew in stage │
└─────────────┬─────────────┘
│
┌───────▼───────┐
│ Small side │
│ fits in memory?│
└───┬───────┬───┘
Yes No
│ │
▼ ▼
Broadcast ┌────────────┐
Join │ Only need │
│ aggregate? │
└──┬─────┬───┘
Yes No
│ │
▼ ▼
Pre-Agg ┌──────────┐
│ Many mod │
│ hot keys?│
└──┬───┬──┘
Yes No
│ │
▼ ▼
Increase Isolate &
Parallel Broadcast
Isolate-and-Broadcast Flow
┌──────────────────┐ ┌──────────────┐
│ Transactions │ │ Accounts │
└────────┬─────────┘ └──────┬───────┘
│ │
┌─────┴─────┐ │
▼ ▼ │
Hot Keys Cold Keys │
│ │ │
▼ ▼ │
broadcast() shuffle join ◄───┘
│ │
└─────┬─────┘
▼
UNION ALL
│
▼
┌─────────────┐
│ Result │
└─────────────┘
Key Takeaway
Beyond salting and Adaptive Query Execution, there is a toolbox of skew techniques and you pick based on the symptom. Isolate-and-broadcast removes hot keys from the shuffle path by joining them separately with a broadcast join, which works well when the hot keys are few and known. Increasing parallelism spreads moderately-hot key collisions across more partitions but cannot fix a single mega-key. Pre-aggregation shrinks the hot partition at the source by sending partial results instead of raw rows. A broadcast hash join sidesteps skew entirely when the small side fits in memory.
The deepest fix is redesigning the data model so the hot key no longer exists, for example splitting a pooling account into numbered sub-accounts. This is the most durable solution but the most expensive to implement, so you reach for it when the same hot key keeps causing incidents across many jobs rather than patching each job.
At the National Bank of India these were real decisions. A transactions join skewed on one corporate pooling account was cleanest with isolate-and-broadcast. A merchant job with dozens of colliding keys was fixed by raising shuffle partitions. A per-account counting job was fixed by pre-aggregating before the join. The rule is to start with the cheapest technique that fits the symptom, combine two or three when needed, and reserve model redesign for structural recurring problems.
Remember one sentence: start with the cheapest and most surgical skew technique that fits the symptom, combine techniques when needed, and reserve data model redesign for hot keys that keep causing incidents across many jobs.
---