Joins e pushdown em runtime

O formato de um join

{
  "type": "join",
  "with": { "format": "delta", "path": "sales.customers" },
  "with_transformations": [
    { "type": "filter", "condition": "active = true" },
    { "type": "select", "columns": ["customer_id", "segment"] },
    { "type": "distinct" }
  ],
  "on": "customer_id",
  "how": "left"
}

O lado direito é lido inline e remodelado por with_transformations antes do join. Três hábitos compensam imediatamente:

• select apenas o que você precisa. Um join carrega toda coluna de ambos os lados para dentro do shuffle.
• distinct quando o lado direito pode repetir a chave, ou um left join silenciosamente multiplica linhas.
• Filtre cedo, no lado direito, não depois do join.

Formas do on

"on": "customer_id"                                  // same name on both sides
"on": ["customer_id", "order_date"]                  // composite key
"on": "l.customer_id = r.id AND l.dt = r.date"       // different names, or an inequality

O DataFrame esquerdo recebe o alias l e o direito, r. Esses aliases existem apenas na expressão on e após o join — dentro de with_transformations o lado direito ainda não tem alias, então use nomes de coluna puros ali.

how

Valor	Mantém
inner (padrão)	linhas que casam em ambos os lados
left	toda linha esquerda; colunas direitas nulas quando não há correspondência
right	toda linha direita
full	tudo de ambos
leftsemi	linhas esquerdas que têm correspondência — sem colunas direitas
leftanti	linhas esquerdas sem correspondência — a consulta “o que está faltando?”
cross	produto cartesiano

leftsemi é o que se deve buscar quando você só precisa filtrar por existência: ele nunca duplica linhas e nunca adiciona colunas.

Pushdown em runtime

O problema: seu conjunto de trabalho tem 40 000 pedidos, e a tabela contra a qual você precisa fazer join tem 900 milhões de linhas. Um join comum lê a tabela grande.

A solução é computar a lista de chaves primeiro e empurrá-la para dentro da leitura como um literal:

[
  { "type": "filter", "condition": "status = 'PENDING'" },
  { "type": "checkpoint" },
  { "type": "collect", "column": "customer_id", "as": "pending_customers" },
  {
    "type": "join",
    "with": { "format": "delta", "path": "sales.bronze_events" },
    "with_transformations": [
      { "type": "filter", "condition": "customer_id IN ({{pending_customers}})" },
      { "type": "select", "columns": ["customer_id", "last_event_at", "channel"] }
    ],
    "on": "customer_id",
    "how": "left"
  }
]

{{pending_customers}} é substituído dentro do filtro que roda no lado direito, então Delta e Parquet podem usar suas estatísticas para pular arquivos em vez de escanear a tabela. Esta é a forma declarativa do padrão collect() + isin() que os jobs Spark escritos à mão usam.

Regras que fazem isso funcionar

Cuidado

checkpoint antes do collect. collect é uma ação no lado do driver; em um plano não materializado o Spark recomputa toda a linhagem para produzir a lista, e depois de novo para o join.

• A formatação é automática: strings viram 'a', 'b', números 1, 2.
• Uma coleção vazia é renderizada como NULL, então IN (NULL) não casa com nada — a resposta correta quando não há nada a enriquecer.
• O store de variáveis é compartilhado com os with_transformations aninhados, e é por isso que o lado direito do join consegue enxergá-lo.
• Ele é zerado no início de cada execução, então nada vaza entre pipelines.

Quando não usá-lo

A lista de chaves viaja para o driver e para dentro de uma string SQL. Dezenas de milhares de chaves são aceitáveis; milhões não são — nesse tamanho, um join comum com uma dica de broadcast ou uma tabela bucketed é a ferramenta certa. Se o conjunto de trabalho for ilimitado, pule o pushdown.

Union não é join

{ "type": "union", "with": { "format": "parquet", "path": "/data/orders_archive" }, "allow_missing_columns": true }

Perigo

Com o padrão allow_missing_columns: false, o Spark casa colunas por posição. Duas fontes cujos schemas divergiram vão alegremente concatenar em lixo. Defina como true para casar por nome e preencher colunas ausentes com nulo.

union não tem with_transformations — o lado direito é lido como está. Modele-o a montante, ou use um join se precisar.

Conferindo o trabalho

Solte um nó debug após o join enquanto desenvolve:

{ "type": "debug", "label": "after enrichment", "actions": ["count", "print_schema"] }

Uma contagem de linhas que cresceu significa que o lado direito não era único na chave de join. Esse é, de longe, o bug de join mais comum, e uma validação unique na chave o captura permanentemente.