Salting: spreading a hot key across partitions, and the trade-off | Data Engineering

Q: How does salting work for a skewed groupBy?

For aggregation skew, salting requires a twostage approach. First aggregate per salted key, then reaggregate across all salt values to produce the true total. from pyspark.sql.functions import sum as spark_sum Stage 1: aggregate per salted key salted_txns_agg = txns \ .withColumn("salt", (rand() SAL

Definition

Salting appends a random suffix to the hot key to spread it across N partitions and replicates the other side N times to still match – trading duplicated data for balanced partitions.

Salting – a manual skew-mitigation technique that appends a random number (the salt) to a skewed join key on the large side, splitting one hot key into N composite keys.
Explosion – the small side is replicated N times, once per salt value, so each composite key on the large side finds its match on the small side.
Trade-off – the small side's row count, shuffle, and memory footprint multiply by N, so you balance N carefully.
Targeted salting – apply the salt only to known hot keys to limit duplication rather than salting the entire dataset.
Two-stage aggregation – for skewed groupBy operations, aggregate per salted key first, then re-aggregate across all salt values to get the true total.

At the National Bank of India, one pooling account's millions of transactions all pile onto one clerk. Salting is like splitting that account into ten sub-labels – account plus zero through account plus nine – and sprinkling its transactions across ten clerks. But for the matching to work, the one reference record for that account must be copied to all ten sub-labels so each clerk has what they need. You have spread the heavy work across ten clerks, but you made ten copies of the reference record.

Understanding the Concepts

What is salting in Spark?

Salting is a technique to fix data skew by breaking a single hot key into multiple composite keys, distributing the load across several partitions. You add a random integer between zero and N-1 to the hot key on the large side, creating N distinct composite keys. After the shuffle, the hot key's rows are spread across N partitions instead of piling into one.

PYTHON

from pyspark.sql import SparkSession
from pyspark.sql.functions import col, lit, rand, floor, concat, explode, array

spark = SparkSession.builder.appName("SaltingDemo").getOrCreate()
SALT_BUCKETS = 10

# Large side: transactions at the National Bank of India
txns = spark.read.parquet("/data/transactions")

# Add salt column: random int 0 to N-1
salted_txns = txns.withColumn("salt", (rand() * SALT_BUCKETS).cast("int")) \
    .withColumn("salted_key",
        concat(col("account_id"), lit("_"), col("salt")))

How does the small side get exploded?

The small side must be replicated N times so that each salt variant of the hot key finds a match. You create an array of all salt values, explode it into rows, and build the same composite key.

PYTHON

# Small side: account reference data
accounts = spark.read.parquet("/data/accounts")

# Explode: replicate each account row N times with salt 0..N-1
salt_values = array([lit(i) for i in range(SALT_BUCKETS)])
exploded_accounts = accounts.withColumn("salt", explode(salt_values)) \
    .withColumn("salted_key",
        concat(col("account_id"), lit("_"), col("salt")))

# Join on the salted composite key
result = salted_txns.join(exploded_accounts, "salted_key") \
    .drop("salt", "salted_key")

What is the trade-off of salting?

The cost of salting is replication. Exploding the small side by N multiplies its row count, shuffle footprint, and memory consumption by N. If N is large and you explode the entire small side, a table of one million accounts becomes ten million rows, which can itself become a problem.

| Salt Factor (N) | Small Side Rows | Shuffle Increase | Skew Reduction |
|---|---|---|---|
| 5 | 5x original | 5x | Moderate |
| 10 | 10x original | 10x | Good |
| 50 | 50x original | 50x | Excellent but costly |
| 100 | 100x original | 100x | Overkill for most cases |

PYTHON

# Choosing N: balance between skew reduction and duplication cost
# If the hot key has 10M rows and 200 shuffle partitions,
# each salted partition gets 10M / N rows.
# N=10 -> 1M rows per partition (usually fine)
# N=100 -> 100K rows per partition (great, but 100x small-side cost)

How do you implement targeted salting?

Targeted salting applies the salt only to known hot keys and leaves cold keys unsalted. This limits the duplication cost to just the rows of the small side that match the hot keys.

PYTHON

from pyspark.sql.functions import when

HOT_KEYS = ["CORP_POOL_001", "SUSPENSE_ACCT"]

# Large side: salt only hot keys
salted_txns = txns.withColumn("salt",
    when(col("account_id").isin(HOT_KEYS),
         (rand() * SALT_BUCKETS).cast("int"))
    .otherwise(lit(0))
).withColumn("salted_key",
    concat(col("account_id"), lit("_"), col("salt")))

# Small side: explode only hot key rows
hot_accounts = accounts.filter(col("account_id").isin(HOT_KEYS))
cold_accounts = accounts.filter(~col("account_id").isin(HOT_KEYS))

exploded_hot = hot_accounts.withColumn("salt", explode(salt_values)) \
    .withColumn("salted_key",
        concat(col("account_id"), lit("_"), col("salt")))

cold_with_key = cold_accounts.withColumn("salt", lit(0)) \
    .withColumn("salted_key",
        concat(col("account_id"), lit("_"), col("salt")))

exploded_accounts = exploded_hot.union(cold_with_key)
result = salted_txns.join(exploded_accounts, "salted_key").drop("salt", "salted_key")

How does salting work for a skewed groupBy?

For aggregation skew, salting requires a two-stage approach. First aggregate per salted key, then re-aggregate across all salt values to produce the true total.

PYTHON

from pyspark.sql.functions import sum as spark_sum

# Stage 1: aggregate per salted key
salted_txns_agg = txns \
    .withColumn("salt", (rand() * SALT_BUCKETS).cast("int")) \
    .withColumn("salted_key",
        concat(col("account_id"), lit("_"), col("salt"))) \
    .groupBy("salted_key", "account_id") \
    .agg(spark_sum("amount").alias("partial_total"))

# Stage 2: re-aggregate across salts to get true total
final_totals = salted_txns_agg.groupBy("account_id") \
    .agg(spark_sum("partial_total").alias("total_amount"))

final_totals.show(5)

TEXT

+-------------+------------------+
|   account_id|      total_amount|
+-------------+------------------+
|CORP_POOL_001|4987652341.50     |
|          101|     487532.75    |
|          250|     312847.00    |
+-------------+------------------+

When should you use AQE instead of manual salting?

Adaptive Query Execution in Spark 3.0+ can detect and split skewed partitions automatically at runtime. Try AQE first because it requires no code changes.

PYTHON

# Enable AQE 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",
    "256MB")

# AQE detects skew at runtime and splits the oversized partition
result = txns.join(accounts, "account_id")
result.count()  # AQE handles the skew transparently

Reach for manual salting when:

AQE does not reduce the skew enough (both sides skewed on the same key).
The skew is in a groupBy aggregation, not a join.
You need fine-grained control over the salt factor per hot key.
You are on a Spark version older than 3.0 without AQE.

What are common mistakes with salting?

The most frequent mistakes are salting everything unnecessarily and forgetting the two-stage aggregation.

PYTHON

# WRONG: salting the entire dataset including cold keys
# This explodes the small side for every key, not just the hot ones
bad_salted = txns.withColumn("salt", (rand() * 100).cast("int"))
bad_exploded = accounts.withColumn("salt",
    explode(array([lit(i) for i in range(100)])))
# Result: accounts table goes from 1M to 100M rows unnecessarily

# CORRECT: targeted salting on only the hot keys (shown above)

PYTHON

# WRONG: forgetting re-aggregation for a salted groupBy
bad_totals = txns \
    .withColumn("salt", (rand() * 10).cast("int")) \
    .withColumn("salted_key",
        concat(col("account_id"), lit("_"), col("salt"))) \
    .groupBy("salted_key").agg(spark_sum("amount").alias("total"))
# Result: 10 partial totals per account instead of one true total

# CORRECT: add a second aggregation stage (shown above)

What should a senior engineer know about salting?

Salting is intrusive code that should be documented and ideally gated behind detected hot keys.
The salt factor N should be tuned based on the hot key's row count and the target partition size.
For very large hot keys, combine salting with broadcast join on the hot key subset for maximum performance.
Monitor whether the hot keys change over time and update the salt configuration accordingly.
In production pipelines at the National Bank of India, salting configuration was externalized to a config file listing hot account ids and their salt factors, so the pipeline adapted without code changes.

Visual Overview

DIAGRAM

  Without Salting (skewed)
  ┌──────────────────────┐
  │  Large Side (txns)   │
  │  account_id = CORP   │
  │  10M rows            │
  └──────────┬───────────┘
             │ hash(CORP)
             ▼
  ┌─────────────────┐
  │   Partition 7   │  10M rows  ← STRAGGLER
  └─────────────────┘

  With Salting (balanced)
  ┌──────────────────────┐
  │  Large Side (txns)   │
  │  CORP_0 ... CORP_9   │
  │  10M rows spread     │
  └──┬──┬──┬──┬──┬──┬──┬─┘
     │  │  │  │  │  │  │
     ▼  ▼  ▼  ▼  ▼  ▼  ▼
  ┌────┐┌────┐┌────┐┌────┐ ...
  │ 1M ││ 1M ││ 1M ││ 1M │     10 balanced partitions
  └────┘└────┘└────┘└────┘

  Small Side Explosion
  ┌──────────┐     ┌──────────────────────────────┐
  │ CORP row │ --> │ CORP_0, CORP_1, ... CORP_9   │
  │ (1 row)  │     │ (10 copies)                  │
  └──────────┘     └──────────────────────────────┘

Key Takeaway

Salting is a manual skew-mitigation technique with two coordinated steps. On the large side, you add a salt column with a random integer from zero to N-1 and build a composite join key of the original key combined with the salt. The hot key, which was a single value funneling millions of rows into one partition, becomes N distinct composite keys spread across N partitions. On the small side, you explode each matching row into N copies with the same composite keys so the join still finds matches.

The trade-off is replication. Exploding the small side by N multiplies its row count, shuffle size, and memory footprint by N. The preferred approach is targeted salting: identify the few hot keys and only replicate the small-side rows for those keys, leaving cold keys unsalted. Choosing N is a balance between skew reduction and duplication cost.

At the National Bank of India, the transactions-to-accounts join was skewed by a corporate pooling account whose partition straggled hundreds of times longer than the rest. We first tried AQE skew join optimization, and when it was insufficient we salted that specific hot key, spreading its transactions across ten salted composite keys and replicating just its account row into the same variants. For a skewed groupBy, salting means a two-stage aggregation: sum per salted key first, then re-aggregate across salts to get the true total.

The judgment a senior engineer conveys is to try AQE first, reach for manual salting only when AQE is not enough, salt only the hot keys to control duplication, and document the salt configuration so it can be updated as data patterns change.

Remember one sentence: salting spreads a hot key across N partitions by appending a random suffix and replicating the other side N times, trading controlled duplication for balanced partitions.

---