Joins no Spark: broadcast e shuffle

Para juntar duas tabelas distribuídas, as linhas com a mesma chave precisam estar na mesma máquina. Há dois jeitos de conseguir isso.

Shuffle (sort-merge join)

Os dois lados são redistribuídos pela chave (shuffle), ordenados e casados. Funciona para qualquer tamanho, mas move os dois lados pela rede e grava arquivos de shuffle em disco. É o padrão para dois lados grandes.

Broadcast (broadcast hash join)

O lado pequeno é copiado inteiro para cada executor, e o lado grande fica onde está: sem shuffle dele. O Spark faz isso sozinho quando estima que um lado tem menos que spark.sql.autoBroadcastJoinThreshold (padrão 10 MB); a AQE pode decidir de novo durante a execução, com o tamanho real.

from pyspark.sql import functions as F

vendas = spark.table("silver.vendas")       # bilhões de linhas
lojas = spark.table("silver.lojas")         # algumas centenas

juntas = vendas.join(F.broadcast(lojas), "loja_id")
juntas.explain()

No plano, procure BroadcastHashJoin (ou BroadcastExchange) em vez de SortMergeJoin com dois Exchange. Em SQL, a mesma dica é SELECT /*+ BROADCAST(lojas) */ ....

Quando não fazer broadcast

• O lado "pequeno" não cabe com folga na memória de cada executor e do driver (o driver monta a cópia).
• O lado pequeno é o da esquerda de um LEFT JOIN: o Spark não consegue fazer broadcast do lado que precisa ser preservado inteiro.

Join com skew

Se uma chave aparece em milhões de linhas, a partição dela no sort-merge join fica gigante. A AQE divide partições enviesadas (spark.sql.adaptive.skewJoin.enabled, ligado por padrão); quando não basta, filtrar a chave quente e tratá-la à parte, ou o *salting*, resolve.

• Chave NULL nunca casa: muitos NULL numa chave de join são linhas que vão para a mesma partição e não servem para nada. Filtre antes de juntar.
• Juntar por colunas de tipos diferentes (texto com número) força conversão e pode estragar o plano.