Como detetar e corrigir skew em joins em Apache Spark: passo a passo

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

Aprende a detetar e corrigir skew (desiquilíbrio) em joins em Apache Spark para evitar tarefas lentas e falhas por falta de memória. Saber identificar e aplicar técnicas como broadcast, salting e repartition melhora o desempenho em ETL e pipelines de dados.

Pré-requisitos

Apache Spark instalado (Spark 3.x) e acesso a um ambiente PySpark ou notebook.
Conhecimentos básicos da DataFrame API (select, join, groupBy).
Conjunto de dados de exemplo com chaves desiquilibradas (ou capacidade de simular).

Passo 1: Detectar skew nas chaves do join

Antes de optimizar, é preciso confirmar que existe skew. Conta a frequência das chaves de join para ver valores extremamente populares que causam partições muito 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()

Passo 2: Verificar tempo e falhas do join (investigar)

Executa um join simples e observa o plano físico e as métricas das tarefas (stages). Procura tarefas com muito maior duração ou executores com GC/memória 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

Passo 3: Usar broadcast quando uma tabela é pequena

Se uma das tabelas couber em memória do executor, usa broadcast para evitar shuffle. Esta é a técnica mais simples e eficaz quando aplicável.

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

Passo 4: Salting — distribuir valores quentes

Para chaves muito populares (hot keys), o salting cria subchaves adicionando um valor aleatório, distribuindo a carga por várias partições. Aplica salt a ambas as tabelas correspondentes.

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

Passo 5: Repartition por chave antes do join

Reparticionar explicitamente por chave ajuda a equilibrar o shuffle e garante que as partições contenham a mesma chave, reduzindo movimentos desnecessários.

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

Passo 6: Combinar técnicas para casos difíceis

Em cenários reais, combinações são úteis: broadcast para tabelas pequenas, salting para hot keys e repartition para balanceamento geral. Testa cada abordagem com amostras antes de pôr em produção.

# 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

Verificar o resultado

Confirma a melhoria comparando tempos de execução, shuffle write size e duração das tarefas no Spark UI. Valida que os resultados coincidem com o join original (mesmo número de linhas e 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 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()

Conclusão

Detetar e corrigir skew em joins em Apache Spark acelera pipelines e evita falhas. Começa por medir frequências e analisar o plano; aplica broadcast quando possível; usa salting para hot keys e repartition para equilibrar. Próximos passos: automatizar a deteção de hot keys e criar testes de desempenho. Dica: começa sempre por testar em amostras antes de aplicar a toda a base.