How to detect and fix skew in joins in Apache Spark: step by step

João Barros 30 de July de 2026 4 min read

Learn how to detect and fix skew (data imbalance) in joins in Apache Spark to avoid slow tasks and out-of-memory failures. Being able to identify and apply techniques like broadcast, salting and repartition improves performance in ETL and data pipelines.

Prerequisites

Apache Spark installed (Spark 3.x) and access to a PySpark environment or notebook.
Basic knowledge of the DataFrame API (select, join, groupBy).
Example dataset with unbalanced keys (or ability to simulate).

Step 1: Detect skew in the join keys

Before optimizing, you need to confirm that skew exists. Count the frequency of join keys to see extremely popular values that cause very large partitions.

# PySpark example: contar frequência de chave
from pyspark.sql import SparkSession
spark = SparkSession.builder.getOrCreate()
df = spark.read.parquet('/path/to/tableA')
# coluna de join: key
freq = df.groupBy('key').count().orderBy('count', ascending=False).limit(20)
freq.show()

Step 2: Check join time and failures (investigate)

Run a simple join and inspect the physical plan and task metrics (stages). Look for tasks with much higher duration or executors with high GC/memory.

# small join para análise do plano
dfB = spark.read.parquet('/path/to/tableB')
joined = df.join(dfB, on='key', how='inner')
joined.explain(True)  # vê o plano físico
joined.count()       # força execução para ver métricas no UI do Spark

Step 3: Use broadcast when one table is small

If one of the tables fits in executor memory, use broadcast to avoid shuffle. This is the simplest and most effective technique when applicable.

from pyspark.sql.functions import broadcast
# se dfB for pequena:
joined_b = df.join(broadcast(dfB), on='key', how='inner')
joined_b.count()

Step 4: Salting — distribute hot values

For very popular keys (hot keys), salting creates subkeys by adding a random value, distributing the load across multiple partitions. Apply salt to both corresponding tables.

import pyspark.sql.functions as F
from pyspark.sql import functions as sf
# número de buckets para salt (ajusta conforme a carga)
num_salts = 10
# adicionar coluna salt em cada tabela
df_salted = df.withColumn('salt', (F.rand()*num_salts).cast('int'))
dfB_salted = dfB.withColumn('salt', (F.rand()*num_salts).cast('int'))
# criar chave composta
df_salted = df_salted.withColumn('key_salted', F.concat_ws('_', F.col('key'), F.col('salt')))
dfB_salted = dfB_salted.withColumn('key_salted', F.concat_ws('_', F.col('key'), F.col('salt')))
# join pela chave composta
joined_salted = df_salted.join(dfB_salted, on='key_salted', how='inner')

# depois de agregado, remover o salt
result = joined_salted.drop('salt', 'key_salted')

Step 5: Repartition by key before the join

Explicitly repartitioning by key helps balance the shuffle and ensures that partitions contain the same key, reducing unnecessary movement.

# repartition por chave com número de partições adequado
num_parts = 200
df_r = df.repartition(num_parts, 'key')
dfB_r = dfB.repartition(num_parts, 'key')
joined_r = df_r.join(dfB_r, on='key', how='inner')
joined_r.count()

Step 6: Combine techniques for difficult cases

In real scenarios, combinations are useful: broadcast for small tables, salting for hot keys and repartition for general balancing. Test each approach with samples before deploying to production.

# exemplo combinado: broadcast + salting quando dfB pequena exceto hot keys
# identifica hot keys top N
hot_keys = df.groupBy('key').count().orderBy('count', ascending=False).limit(50).select('key')
# trata hot_keys com salting e o restante com join normal/broadcast conforme o tamanho

Verify the result

Confirm the improvement by comparing execution times, shuffle write size and task durations in the Spark UI. Validate that the results match the original join (same number of rows and equivalent aggregates).

# validar correspondência de resultados (exemplo simples)
orig = df.join(dfB, on='key', how='inner').groupBy('key').count()
opt = joined_salted.groupBy('key').count()
# compara contagens por chave (pode ser custoso em dados muito grandes)
diff = orig.join(opt, on='key', how='full_outer').select(
    orig['count'].alias('c1'), opt['count'].alias('c2'))
# filtrar discrepâncias
diff.filter((F.col('c1') != F.col('c2')) | F.col('c1').isNull() | F.col('c2').isNull()).show()

Conclusion

Detecting and fixing skew in joins in Apache Spark speeds up pipelines and prevents failures. Start by measuring frequencies and analyzing the plan; apply broadcast when possible; use salting for hot keys and repartition to balance. Next steps: automate hot key detection and create performance tests. Tip: always start by testing on samples before applying to the entire dataset.