Skew Scenario: One Merchant = 60% of All Transactions

One merchant generates 60% of all transactions. Your join is collapsing on a single executor. Where does the fix actually live — in the data, the query, or the cluster?

The Setup

You're building a daily aggregation pipeline that joins transaction events with a merchant details lookup table:

# Transaction events: 100 million rows per day
transactions_df = spark.read.parquet("s3://datalake/transactions/2026-05-30/")

# Merchant details: 50,000 rows (small lookup table)
merchants_df = spark.read.parquet("s3://datalake/dim_merchants/")

# The join that kills your cluster
result = transactions_df.join(merchants_df, on="merchant_id", how="inner")

The Data Distribution Problem:

+------------------+-----------------+----------+
|   merchant_id    |  transaction_ct | share(%) |
+------------------+-----------------+----------+
|   mch_amazon     |    60,000,000   |   60.0%  |  ← ⚠️ MEGA MERCHANT
|   mch_flipkart   |     5,000,000   |    5.0%  |
|   mch_swiggy     |     3,000,000   |    3.0%  |
|   mch_zomato     |     2,500,000   |    2.5%  |
|   ... (49,996    |    29,500,000   |   29.5%  |
|    other merch.) |                 |          |
+------------------+-----------------+----------+
|   TOTAL          |   100,000,000   |  100.0%  |
+------------------+-----------------+----------+

When Spark executes the join, it computes hash("mch_amazon") % num_partitions and sends ALL 60 million Amazon transactions to a single partition on a single executor core. That executor chokes while 199 other cores finish in seconds and sit idle.

Diagnosing the Skew: Spark UI Evidence

Open Spark UI → Stages Tab → Click on the join stage:

What you'll see in the Task Metrics:

Summary Metrics for 200 Completed Tasks:
+----------+-----------+-------------+---------------+---------+
| Metric   | Min       | 25th Pct    | Median        | Max     |
+----------+-----------+-------------+---------------+---------+
| Duration | 1.2 s     | 2.1 s       | 3.5 s         | 47 min  |  ← 800x skew!
| Input    | 25 MB     | 50 MB       | 75 MB         | 18 GB   |  ← Massive!
| Shuffle  | 12 MB     | 30 MB       | 45 MB         | 15 GB   |
| GC Time  | 50 ms     | 120 ms      | 200 ms        | 12 min  |
+----------+-----------+-------------+---------------+---------+

Diagnosis confirmed: The Max task has 800x the duration and 240x the data of the median task. One partition holds 60% of the data.

Programmatic Verification:

from pyspark.sql import functions as F

# Audit partition distribution
transactions_df.withColumn("partition_id", F.spark_partition_id()) \
    .groupBy("partition_id") \
    .agg(
        F.count("*").alias("row_count"),
        F.countDistinct("merchant_id").alias("unique_merchants")
    ) \
    .orderBy(F.desc("row_count")) \
    .show(10)

+------------+-----------+------------------+
|partition_id|  row_count|unique_merchants   |
+------------+-----------+------------------+
|          47| 60,000,000|                 1 |  ← ALL Amazon here!
|         132|    520,000|               312 |
|          89|    510,000|               298 |
|          ...    ...          ...           |
+------------+-----------+------------------+

Where Does the Fix Live?

This is the critical question. There are three layers where you can address skew, and the answer is: it depends on your constraints.

graph TD
    A["Data Skew Detected"] --> B{"Is the small table<br/>small enough<br/>to broadcast?"}
    B -->|"Yes (< 1 GB)"| C["FIX IN QUERY:<br/>Broadcast Join"]
    B -->|"No (> 1 GB)"| D{"Is the skew<br/>caused by one<br/>dominant key?"}
    D -->|"Yes"| E["FIX IN DATA:<br/>Salting"]
    D -->|"No (many hot keys)"| F{"Can you add<br/>more resources?"}
    F -->|"Yes"| G["FIX IN CLUSTER:<br/>AQE + More Partitions"]
    F -->|"No"| H["FIX IN DATA:<br/>Pre-aggregate +<br/>Two-Pass Join"]

Fix 1: In the Query — Broadcast Join (Fastest, Simplest)

If the merchant details table is small (< 1 GB), you can eliminate the shuffle entirely by broadcasting the small table to all executors:

from pyspark.sql.functions import broadcast

# Force broadcast of the small lookup table
result = transactions_df.join(
    broadcast(merchants_df),   # ← This changes EVERYTHING
    on="merchant_id",
    how="inner"
)

What happens internally:

WITHOUT Broadcast (Shuffle Join):
    Transaction records for "mch_amazon" → hash → Partition 47 → 1 executor chokes

WITH Broadcast:
    Merchant table (50K rows, ~5 MB) → copied to EVERY executor's memory
    Transaction records stay in their original partitions → NO shuffle!
    Each executor joins its local partition with its local copy of merchants
    → ALL executors work in parallel, evenly

When to use: The lookup table must fit in executor memory. Default broadcast threshold is 10 MB (spark.sql.autoBroadcastJoinThreshold). Increase it:

spark.conf.set("spark.sql.autoBroadcastJoinThreshold", "1073741824")  # 1 GB

When NOT to use: If BOTH tables are large (e.g., joining 100M transactions with 500M user events), broadcasting is impossible. You need salting.

Fix 2: In the Data — Salting (The Universal Fix)

When both sides are large and broadcast isn't an option, salting breaks the hot key into multiple virtual keys:

from pyspark.sql import functions as F

salt_buckets = 10  # Split the hot key across 10 virtual partitions

# ============================================================
# STEP 1: Salt the large (skewed) DataFrame
# ============================================================
transactions_salted = transactions_df.withColumn(
    "salt", F.floor(F.rand() * salt_buckets).cast("int")
).withColumn(
    "merchant_id_salted",
    F.concat(F.col("merchant_id"), F.lit("_"), F.col("salt"))
)

# ============================================================
# STEP 2: Explode (replicate) the small DataFrame
# ============================================================
salt_array = F.array([F.lit(i) for i in range(salt_buckets)])

merchants_exploded = merchants_df.withColumn(
    "salt", F.explode(salt_array)
).withColumn(
    "merchant_id_salted",
    F.concat(F.col("merchant_id"), F.lit("_"), F.col("salt"))
)

# ============================================================
# STEP 3: Join on the salted key
# ============================================================
result_salted = transactions_salted.join(
    merchants_exploded,
    on="merchant_id_salted",
    how="inner"
)

# ============================================================
# STEP 4: Clean up — drop salt columns
# ============================================================
result_clean = result_salted.drop("salt", "merchant_id_salted")

What salting does to the partition distribution:

BEFORE SALTING:
Partition 47: [mch_amazon × 60,000,000]  ← ONE executor, 47 min

AFTER SALTING (salt_buckets=10):
Partition 12: [mch_amazon_0 × 6,000,000]  ← 6M per partition
Partition 34: [mch_amazon_1 × 6,000,000]
Partition 56: [mch_amazon_2 × 6,000,000]
Partition 78: [mch_amazon_3 × 6,000,000]
Partition 91: [mch_amazon_4 × 6,000,000]
...
Partition 198: [mch_amazon_9 × 6,000,000]  ← 10 executors share the load!

Trade-off: Salting replicates the smaller DataFrame by salt_buckets times. If the smaller table has 50K rows × 10 buckets = 500K rows — trivial. But if it's 100M rows × 10 = 1B rows, the overhead is significant.

Fix 3: In the Cluster — Adaptive Query Execution (AQE)

Starting from Spark 3.0, AQE can automatically detect and fix skew at runtime:

# Enable AQE with skew join optimization
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")
spark.conf.set("spark.sql.adaptive.coalescePartitions.enabled", "true")

How AQE handles skew internally:

1. Spark executes the shuffle as normal.
2. AQE analyzes the actual shuffle partition sizes at runtime.
3. Detects that Partition 47 is 5x larger than the median (skewed).
4. Automatically SPLITS Partition 47 into sub-partitions.
5. Schedules extra tasks to process the sub-partitions in parallel.
6. No code changes needed!

When to use: AQE is a great first line of defense. Enable it ALWAYS. But for extreme skew (60% in one key), AQE alone may not be enough — combine it with salting or broadcast for best results.

Fix 4: In the Data — Two-Pass Join (Advanced)

For extreme skew where neither broadcast nor simple salting works well, use a two-pass approach — process the hot key separately from the rest:

# Define the hot key threshold
hot_key = "mch_amazon"

# ============================================================
# PASS 1: Process hot key separately with broadcast
# ============================================================
txn_hot = transactions_df.filter(F.col("merchant_id") == hot_key)
merchant_hot = merchants_df.filter(F.col("merchant_id") == hot_key)

result_hot = txn_hot.join(broadcast(merchant_hot), on="merchant_id", how="inner")

# ============================================================
# PASS 2: Process all other keys normally (no skew!)
# ============================================================
txn_rest = transactions_df.filter(F.col("merchant_id") != hot_key)
merchant_rest = merchants_df.filter(F.col("merchant_id") != hot_key)

result_rest = txn_rest.join(merchant_rest, on="merchant_id", how="inner")

# ============================================================
# UNION the two results
# ============================================================
final_result = result_hot.unionByName(result_rest)

Why this works: By isolating the hot key, Pass 2 has a perfectly uniform distribution, and Pass 1 uses broadcast which eliminates shuffle entirely.

The Decision Matrix: Which Fix to Use?

Post-Fix Verification

After applying your fix, verify the skew is resolved:

# Re-check partition distribution after the fix
final_result.withColumn("partition_id", F.spark_partition_id()) \
    .groupBy("partition_id") \
    .count() \
    .agg(
        F.min("count").alias("min_partition_size"),
        F.max("count").alias("max_partition_size"),
        F.avg("count").alias("avg_partition_size"),
        (F.max("count") / F.avg("count")).alias("skew_ratio")
    ).show()

+------------------+------------------+------------------+-----------+
|min_partition_size |max_partition_size |avg_partition_size |skew_ratio |
+------------------+------------------+------------------+-----------+
|          450,000 |          650,000 |          500,000 |      1.30 |  ← ✅ Healthy!
+------------------+------------------+------------------+-----------+

A skew_ratio < 3 is healthy. Above 5 needs attention. Above 10 is a crisis.

Summary

The Answer: The fix lives in all three layers — enable AQE as a baseline (cluster), try broadcast first (query), and use salting when broadcast isn't possible (data). The best production pipelines use a combination.

Follow-Up Questions & Answers

Q1: What if there are MULTIPLE hot keys, not just one merchant?

A: If 5-10 merchants each hold 10%+ of the data, salting on the join key alone may not be enough. You need a frequency-aware salting strategy:

# Step 1: Identify all hot keys (keys with > 5% of total data)
key_counts = transactions_df.groupBy("merchant_id").count()
total_count = transactions_df.count()

hot_keys = key_counts.filter(F.col("count") > total_count * 0.05) \
    .select("merchant_id").rdd.flatMap(lambda x: x).collect()

print(f"Hot keys detected: {hot_keys}")
# Output: ['mch amazon', 'mch flipkart', 'mch swiggy', 'mch paytm', 'mch phonepe']

# Step 2: Variable salt factor based on frequency
# Hotter keys get more salt buckets
salt_map = {
    "mch_amazon": 20,     # 60% → split into 20 buckets
    "mch_flipkart": 10,   # 10% → split into 10 buckets
    "mch_swiggy": 5,      #  5% → split into 5 buckets
}

# Step 3: Apply variable salting
@F.udf
def dynamic_salt(merchant_id):
    import random
    buckets = salt_map.get(merchant_id, 1)  # Non-hot keys: no salting
    return random.randint(0, buckets - 1)

transactions_salted = transactions_df.withColumn("salt", dynamic_salt(F.col("merchant_id")))

Q2: Does salting affect the accuracy of aggregations like COUNT, SUM, AVG?

A: No — salting does NOT affect accuracy for joins. The join result is mathematically identical because:

• Each large-table record gets exactly ONE random salt value.
• The small table is replicated to match ALL possible salt values.
• Every join match that would have occurred without salting still occurs.

However, if you're doing a GroupBy aggregation (not a join), salting splits the group and you need an extra re-aggregation step:

# WRONG: GroupBy with salt gives partial aggregates
partial = salted_df.groupBy("merchant_id", "salt").agg(F.sum("amount"))
# This gives 20 partial sums for Amazon instead of 1 total!

# CORRECT: Two-stage aggregation
stage1 = salted_df.groupBy("merchant_id", "salt").agg(F.sum("amount").alias("partial_sum"))
stage2 = stage1.groupBy("merchant_id").agg(F.sum("partial_sum").alias("total_amount"))

Q3: How does Spark 3.x AQE skew join actually work under the hood?

A: AQE skew join optimization has three internal phases:

Phase 1: EXECUTE shuffle normally (collect shuffle statistics)
    └── Spark now knows the exact size of each shuffle partition

Phase 2: DETECT skew at runtime
    └── Partition 47 = 18 GB, Median = 75 MB
    └── 18 GB > 75 MB × 5 (skewedPartitionFactor) → SKEWED!

Phase 3: SPLIT the skewed partition
    └── Partition 47 (18 GB) → Split into 72 sub-partitions of ~250 MB each
    └── The OTHER side of the join duplicates its corresponding partition
        to match each sub-partition
    └── 72 parallel tasks replace the 1 choking task

Limitation: AQE can only handle skew in SortMergeJoin. If Spark is using ShuffledHashJoin or BroadcastNestedLoopJoin, AQE skew handling doesn't apply. Check the physical plan in the SQL tab.

Q4: What if the skew is caused by NULL keys?

A: NULL join keys are a special case of skew. All NULLs hash to the same partition, creating a massive hot spot.

# Step 1: Separate NULL keys from non-NULL keys
null_keys = transactions_df.filter(F.col("merchant_id").isNull())
non_null_keys = transactions_df.filter(F.col("merchant_id").isNotNull())

# Step 2: Process non-NULL keys normally (join works fine)
result_non_null = non_null_keys.join(merchants_df, on="merchant_id", how="inner")

# Step 3: Handle NULL keys separately
# Option A: Drop them (if NULLs shouldn't join)
# Option B: Replace with a default value
null_filled = null_keys.fillna({"merchant_id": "UNKNOWN_MERCHANT"})
result_null = null_filled.join(
    merchants_df.union(
        spark.createDataFrame([("UNKNOWN_MERCHANT", "Unknown", "N/A")], 
                              merchants_df.columns)
    ),
    on="merchant_id", how="inner"
)

# Step 4: Union the results
final = result_non_null.unionByName(result_null)

Interview insight: Always ask "how many NULLs do you have?" before designing a join. A 5% NULL rate in a 1 billion row table is 50 million records — all landing in one partition.

Q5: How do you choose the right salt_buckets number?

A: The formula is:

salt_buckets = ceil(hot_key_records / target_partition_size)

Where:
  hot_key_records = number of records for the hottest key
  target_partition_size = 500,000 to 2,000,000 records (sweet spot)

Example:

• Hot key has 60,000,000 records
• Target partition size = 2,000,000
• salt_buckets = ceil(60,000,000 / 2,000,000) = 30

Warning: Over-salting wastes memory by over-replicating the small table. If you salt with 100 buckets and the small table has 1M rows, you replicate it to 100M rows — which may itself cause OOM.

Sub-Scenarios

Sub-Scenario A: Skew in a Window Aggregation (No Join)

Situation: You're computing a 1-hour rolling sum per merchant. Amazon has 60% of events, so the window computation for Amazon takes 100x longer than others.

Fix: Use partial aggregation + re-aggregation:

# Stage 1: Split Amazon's data across salt buckets, aggregate partially
partial = df.withColumn("salt", F.floor(F.rand() * 10)) \
    .groupBy("merchant_id", "salt", F.window("event_time", "1 hour")) \
    .agg(F.sum("amount").alias("partial_sum"), F.count("*").alias("partial_count"))

# Stage 2: Re-aggregate across salt buckets
final = partial.groupBy("merchant_id", "window") \
    .agg(F.sum("partial_sum").alias("total_amount"), F.sum("partial_count").alias("total_count"))

Sub-Scenario B: Skew Changes Over Time (Dynamic Hot Keys)

Situation: Today Amazon is the hot key, but during the holiday season, Flipkart becomes the hot key. Static salting rules don't adapt.

Fix: Dynamic salt factor based on real-time data profiling:

# Run this profiling query before each job execution
hot_key_profile = transactions_df.groupBy("merchant_id").count() \
    .withColumn("pct", F.col("count") / F.lit(total_count) * 100) \
    .filter(F.col("pct") > 5) \
    .collect()

# Dynamically build salt map
salt_map = {}
for row in hot_key_profile:
    salt_map[row["merchant_id"]] = max(1, int(row["pct"] / 2))  # 1 bucket per 2% share

print(f"Dynamic salt map: {salt_map}")
# Output: {'mch flipkart': 30, 'mch amazon': 10, 'mch meesho': 8}

Sub-Scenario C: Skew in the Write Path (Small File Problem)

Situation: After fixing the join skew, you now have the opposite problem: the salt creates 30 small output files for Amazon and 1 file for each small merchant. Your output directory has thousands of tiny files.

Fix: Repartition before writing + file size optimization:

# After join, repartition by the ORIGINAL key (not the salted key)
result_clean = result_salted.drop("salt", "merchant_id_salted") \
    .repartition(200, "merchant_id") \
    .sortWithinPartitions("merchant_id", "event_timestamp")

# Write with Delta Lake's auto-optimize
result_clean.write \
    .format("delta") \
    .option("optimizeWrite", "true") \      # Delta auto-coalesces small files
    .option("maxRecordsPerFile", 1000000) \  # Target 1M records per file
    .partitionBy("event_date") \
    .save("s3://datalake/transactions/")

Sub-Scenario D: Interview Twist — "Can You Fix Skew Without Salting or Broadcast?"

Answer: Yes, there are alternative approaches:

1. Partial Map-Side Join: Pre-filter the hot key and process it separately using a map-side operation (no shuffle).
2. Range Partitioning: Replace hash partitioning with range partitioning that explicitly distributes the hot key.
3. Pre-aggregation: If the downstream operation is an aggregation, pre-aggregate the hot key's data before the join.
4. Bucketed Tables: Write both tables as bucketed/sorted tables with the same bucket count. Spark can do bucket-to-bucket joins without any shuffle at all.

# Bucketed table approach (eliminates shuffle entirely)
transactions_df.write \
    .bucketBy(200, "merchant_id") \
    .sortBy("merchant_id") \
    .saveAsTable("transactions_bucketed")

merchants_df.write \
    .bucketBy(200, "merchant_id") \
    .sortBy("merchant_id") \
    .saveAsTable("merchants_bucketed")

# This join has ZERO shuffle — each bucket joins with its matching bucket
result = spark.table("transactions_bucketed").join(
    spark.table("merchants_bucketed"), on="merchant_id"
)