Cómo detectar y corregir skew en joins en Apache Spark: paso a paso

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

Aprende a detectar y corregir skew (desequilibrio) en joins en Apache Spark para evitar tareas lentas y fallos por falta de memoria. Saber identificar y aplicar técnicas como broadcast, salting y repartition mejora el rendimiento en ETL y pipelines de datos.

Requisitos previos

Apache Spark instalado (Spark 3.x) y acceso a un entorno PySpark o notebook.
Conocimientos básicos de la DataFrame API (select, join, groupBy).
Conjunto de datos de ejemplo con claves desequilibradas (o capacidad para simularlos).

Paso 1: Detectar skew en las claves del join

Antes de optimizar, es necesario confirmar que existe skew. Cuenta la frecuencia de las claves de join para ver valores extremadamente populares que causan particiones muy grandes.

# 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()

Paso 2: Verificar tiempo y fallos del join (investigar)

Ejecuta un join simple y observa el plan físico y las métricas de las tareas (stages). Busca tareas con mucha mayor duración o ejecutores con GC/memoria elevados.

# 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

Paso 3: Usar broadcast cuando una tabla es pequeña

Si una de las tablas cabe en la memoria del executor, usa broadcast para evitar shuffle. Esta es la técnica más simple y eficaz cuando es aplicable.

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

Paso 4: Salting — distribuir valores calientes

Para claves muy populares (hot keys), el salting crea subclaves añadiendo un valor aleatorio, distribuyendo la carga por varias particiones. Aplica salt a ambas tablas correspondientes.

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')

Paso 5: Repartition por clave antes del join

Reparticionar explícitamente por clave ayuda a equilibrar el shuffle y garantiza que las particiones contengan la misma clave, reduciendo movimientos innecesarios.

# 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()

Paso 6: Combinar técnicas para casos difíciles

En escenarios reales, las combinaciones son útiles: broadcast para tablas pequeñas, salting para hot keys y repartition para balanceo general. Prueba cada enfoque con muestras antes de ponerlo en producción.

# ejemplo combinado: broadcast + salting cuando 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 con salting y el resto con join normal/broadcast conforme al tamaño

Verificar el resultado

Confirma la mejora comparando tiempos de ejecución, shuffle write size y duración de las tareas en el Spark UI. Valida que los resultados coinciden con el join original (mismo número de filas y agregados equivalentes).

# 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 datos muy 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()

Conclusión

Detectar y corregir skew en joins en Apache Spark acelera pipelines y evita fallos. Empieza por medir frecuencias y analizar el plan; aplica broadcast cuando sea posible; usa salting para hot keys y repartition para equilibrar. Próximos pasos: automatizar la detección de hot keys y crear pruebas de rendimiento. Consejo: empieza siempre probando con muestras antes de aplicar a toda la base.