Quando o broadcast join do Spark realmente ajuda, e quando ele se volta contra você?

spark.sql.autoBroadcastJoinThreshold (10MB por padrão) diz ao planejador: se o tamanho estimado de uma tabela está abaixo disso, pule o shuffle e mande uma cópia completa para cada executor.

O que o threshold realmente controla

Abaixo do threshold, o planejador troca um shuffle join (os dois lados reparticionados e embaralhados pela rede) por um broadcast join (o lado pequeno coletado no driver, depois enviado a cada executor como tabela hash). Sem shuffle para o lado grande significa sem stage de shuffle read, sem o skew associado ao shuffle, e normalmente um join visivelmente mais rápido - exatamente por isso o planejador faz isso automaticamente sempre que estima que uma tabela é pequena o suficiente.

O modo de falha: um broadcast que deveria ter ficado como shuffle

A estimativa que o planejador usa vem das estatísticas da tabela, que ficam desatualizadas - uma tabela que tinha 8MB quando as estatísticas foram calculadas pela última vez mas agora tem 800MB ainda vai ser enviada por broadcast se ninguém atualizou essas estatísticas. Resultado: cada executor tenta manter uma tabela hash de 800MB em memória ao mesmo tempo, e muitas vezes é o driver (que coleta os dados do broadcast primeiro) que dá OOM antes de qualquer executor.

Como identificar isso pelo event log

Procure um stage com pouquíssimas tarefas (uma etapa de coleta de broadcast basicamente roda em uma única tarefa) mostrando um uso de memória de pico incomumente alto, seguido imediatamente por uma falha em nível de stage ou um OOM do driver nos logs da aplicação, em vez de uma falha em nível de tarefa. Um OOM de shuffle join, por outro lado, aparece como muitas tarefas falhando num stage de tamanho normal - uma assinatura completamente diferente.