No artigo anterior, vimos uma das operações mais importantes para entender performance em Apache Spark:
Shuffle.
Chegamos a um princípio que vale para boa parte dos sistemas distribuídos:
Mover dados entre máquinas custa caro.
Agora vamos aplicar esse princípio a uma operação presente em praticamente qualquer pipeline de Engenharia de Dados:
Join
Em SQL, um join parece simples:
```sql id="fyy18i"
SELECT *
FROM pedidos p
JOIN clientes c
ON p.customer_id = c.customer_id
Em PySpark, também:
```python id="nhg3yr"
resultado = pedidos.join(
clientes,
"customer_id"
)
Mas existe uma pergunta muito mais interessante:
Como esse join será executado fisicamente em um cluster distribuído?
Essa pergunta muda tudo.
Dependendo do tamanho das tabelas, distribuição das chaves, estatísticas disponíveis e estratégia escolhida pelo Spark, duas operações join() visualmente semelhantes podem produzir comportamentos completamente diferentes.
Podemos ter:
```text id="7lczd6"
Shuffle
ou:
```text id="2xg2di"
Broadcast
Podemos encontrar:
```text id="2a8l5m"
SortMergeJoin
ou:
```text id="uk2e5x"
BroadcastHashJoin
E podemos transformar um join aparentemente simples em um dos maiores gargalos do pipeline.
Vamos entender o que realmente acontece.
1. Por que Join é diferente em um sistema distribuído?
Imagine duas tabelas.
Pedidos
```text id="0q1h4k"
order_id | customer_id | value
1001 | 10 | 250
1002 | 20 | 100
1003 | 10 | 500
1004 | 30 | 150
## Clientes
```text id="nqdk1c"
customer_id | name
-------------------
10 | Ana
20 | Carlos
30 | Maria
Em um banco tradicional, podemos imaginar os dados disponíveis para o mecanismo responsável pela consulta.
Mas em Spark, pedidos pode estar distribuído assim:
```text id="02nkpq"
Executor A
├── customer 10
├── customer 20
└── customer 10
Executor B
├── customer 30
├── customer 50
└── customer 70
Enquanto `clientes` pode estar:
```text id="hnq8xk"
Executor A
├── customer 30
└── customer 40
Executor B
├── customer 10
└── customer 20
Para realizar:
```python id="ffty06"
pedidos.join(
clientes,
"customer_id"
)
o Spark precisa encontrar uma maneira de colocar registros compatíveis em condições de serem comparados.
E é aí que começa a estratégia de execução.
---
# 2. A pergunta fundamental de um Join distribuído
Para juntar:
```text id="8i71ne"
pedido.customer_id = 10
com:
```text id="lhv1yi"
cliente.customer_id = 10
esses registros precisam estar disponíveis para a mesma operação de join.
Uma possibilidade seria redistribuir os dois datasets pela chave:
```text id="g7e79d"
PEDIDOS CLIENTES
P1 ────────┐ ┌──────── P1
P2 ────────┼──► SHUFFLE ◄─┼──────── P2
P3 ────────┘ └──────── P3
↓
Partition customer hash 1
Partition customer hash 2
Partition customer hash 3
↓
JOIN
Agora registros com chaves compatíveis podem chegar à mesma partição.
Funciona.
Mas talvez seja caro.
Se temos:
```text id="49jbyc"
Pedidos = 2 TB
Clientes = 500 GB
podemos acabar movimentando volumes enormes pela rede.
Por isso Spark possui diferentes estratégias.
---
# 3. O plano lógico não é o plano físico
Esse é um conceito muito importante.
Quando escrevemos:
```python id="tvlc3g"
resultado = pedidos.join(
clientes,
"customer_id"
)
estamos descrevendo o que queremos.
Não necessariamente como executar.
Podemos pensar em camadas:
```text id="ukjxlk"
Código
↓
Logical Plan
↓
Optimized Logical Plan
↓
Physical Plan
↓
Execution
Essa separação é poderosa.
Nós declaramos:
> Junte pedidos e clientes por `customer_id`.
O Spark pode analisar diferentes maneiras de executar essa operação.
---
# 4. Catalyst Optimizer
Uma peça fundamental dessa arquitetura é o:
# Catalyst Optimizer
O Catalyst é o framework de otimização utilizado pelo Spark SQL.
De maneira simplificada, podemos pensar:
```text id="61pwwd"
SQL / DataFrame API
↓
Logical Plan
↓
Analysis
↓
Optimization Rules
↓
Optimized Logical Plan
↓
Physical Planning
↓
Physical Plan
Isso significa que escrever:
```python id="o8z96c"
df.filter("value > 100")
ou uma expressão equivalente em SQL não implica simplesmente executar cada chamada literalmente na ordem em que aparece no código.
Spark constrói e otimiza um plano.
Esse modelo é uma das razões pelas quais DataFrames e Spark SQL se tornaram abstrações tão importantes.
---
# 5. Como enxergar o plano: `explain()`
Uma das ferramentas mais importantes para entender Spark é:
```python id="c0xpn1"
df.explain()
Por exemplo:
```python id="7i1e8z"
resultado = pedidos.join(
clientes,
"customer_id"
)
resultado.explain()
Também podemos pedir mais detalhes:
```python id="95nrdw"
resultado.explain("formatted")
ou:
```python id="m3pspj"
resultado.explain("extended")
Dependendo do plano, podemos encontrar termos como:
```text id="2o21jc"
BroadcastHashJoin
ou:
```text id="5o65zv"
SortMergeJoin
Além de operadores como:
```text id="7aoxfm"
Exchange
HashAggregate
Filter
Project
Scan
Sort
A partir desse momento, explain() deixa de ser apenas uma ferramenta de curiosidade.
Ele passa a responder:
O que o Spark pretende fazer com meu código?
6. Primeiro cenário: duas tabelas grandes
Imagine:
```text id="zh5drk"
pedidos = 2 TB
clientes = 800 GB
Nenhuma das duas é pequena.
Distribuí-las integralmente para cada Executor seria inviável.
Uma estratégia possível envolve reorganizar os dois lados pela chave de join.
Conceitualmente:
```text id="c03bti"
PEDIDOS
│
▼
Shuffle by customer_id
│
└──────────────┐
│
▼
JOIN
▲
│
┌──────────────┘
│
Shuffle by customer_id
▲
│
CLIENTES
Esse tipo de operação pode movimentar enormes quantidades de dados.
É aqui que encontramos estratégias como o:
Sort Merge Join
7. Sort Merge Join
O Sort Merge Join é uma estratégia importante para joins entre datasets grandes quando as condições permitem sua utilização.
Uma visão simplificada:
```text id="1ru96w"
TABLE A
│
Shuffle
│
Sort
│
└──────────┐
▼
Merge
▲
┌──────────┘
│
Sort
│
Shuffle
│
TABLE B
Os dados são organizados de forma compatível pela chave de join e posteriormente combinados.
Imagine:
```text id="kl6s69"
TABLE A
1
1
2
3
5
7
e:
```text id="h8y3ik"
TABLE B
1
2
2
3
6
7
Como ambos os lados estão ordenados pela chave, o algoritmo pode avançar pelas estruturas realizando o merge.
Em grandes datasets distribuídos, essa estratégia pode ser muito mais apropriada do que tentar colocar uma das tabelas inteira em memória em todos os Executors.
Mas existe um custo importante:
```text id="dky8x5"
Shuffle
+
Sort
Portanto, novamente:
O join não é apenas uma comparação entre duas colunas.
Existe toda uma arquitetura distribuída por trás dele.
8. Segundo cenário: uma tabela enorme e outra pequena
Agora imagine:
```text id="gl6khe"
EVENTOS
2 TB
e:
```text id="vvz0t7"
PAISES
5 MB
Queremos:
```python id="i5ukm8"
eventos.join(
paises,
"country_id"
)
Será que faz sentido redistribuir:
```text id="lggczd"
2 TB
+
5 MB
pela rede?
Talvez exista uma estratégia muito melhor.
Em vez de reorganizar a tabela gigante, podemos enviar a tabela pequena para os Executors.
É aqui que aparece:
Broadcast Join
9. Broadcast Hash Join
A ideia é simples e extremamente poderosa.
Temos:
```text id="d30q2m"
SMALL TABLE
│
┌─────────┼─────────┐
▼ ▼ ▼
Executor 1 Executor 2 Executor 3
│ │ │
▼ ▼ ▼
local local local
join join join
▲ ▲ ▲
│ │ │
Large Dataset Partitions
Em vez de movimentar a tabela grande para reorganizá-la completamente, distribuímos uma cópia da tabela pequena para os Executors.
Agora cada Executor consegue realizar o join localmente sobre suas partições da tabela grande.
Conceitualmente:
```text id="u11cf9"
Large Table
P1 ───► Executor 1 + Small Table
P2 ───► Executor 2 + Small Table
P3 ───► Executor 3 + Small Table
P4 ───► Executor 4 + Small Table
Isso pode evitar um grande shuffle do lado maior.
10. Por que Broadcast pode ser tão rápido?
Compare dois cenários.
Shuffle Join
```text id="9rmkqr"
Large Table A
│
▼
SHUFFLE
│
└──────────┐
▼
JOIN
▲
┌──────────┘
│
SHUFFLE
▲
│
Large Table B
Agora:
## Broadcast Join
```text id="v6c9kp"
Small Table
│
Broadcast
┌───┼───┐
▼ ▼ ▼
E1 E2 E3
▲ ▲ ▲
│ │ │
Large Table partitions
Não precisamos reorganizar toda a tabela grande apenas para executar o join.
Se a diferença de tamanho for significativa, o ganho pode ser enorme.
11. Fazendo Broadcast explicitamente
Em PySpark podemos indicar:
```python id="h83p3u"
from pyspark.sql.functions import broadcast
resultado = pedidos.join(
broadcast(clientes),
"customer_id"
)
Isso informa ao Spark que queremos tratar `clientes` como lado broadcast.
Depois podemos verificar:
```python id="f1ks36"
resultado.explain("formatted")
E procurar no plano físico algo relacionado a:
```text id="09wcti"
BroadcastHashJoin
Mas existe uma regra importante:
> **Não force broadcast apenas porque parece mais rápido.**
Precisamos entender o tamanho real dos dados.
---
# 12. Broadcast usa memória
Se transmitimos uma tabela para os Executors, cada Executor precisa conseguir utilizá-la.
Imagine:
```text id="hssq4n"
Tabela broadcast = 5 MB
Provavelmente temos um cenário muito diferente de:
```text id="bwy8p7"
Tabela broadcast = 8 GB
Agora imagine dezenas de Executors.
O conceito de:
```text id="du4bcu"
"tabela pequena"
é relativo ao ambiente.
Precisamos considerar:
- memória disponível;
- representação dos dados;
- concorrência;
- número de Executors;
- tamanho real após materialização;
- workload executado simultaneamente.
Portanto:
```text id="5a7hly"
Broadcast
≠
mágica
Ele troca um tipo de custo por outro.
---
# 13. Broadcast automático
Spark pode escolher automaticamente estratégias de broadcast quando possui informações suficientes e o dataset atende às condições necessárias.
Conceitualmente:
```text id="dkx76e"
Join
│
├── Table A: grande
│
└── Table B: pequena
│
▼
Optimizer
│
▼
BroadcastHashJoin
Isso é importante porque mostra novamente a diferença entre:
```text id="0il50n"
código que escrevemos
e:
```text id="3ukcrf"
plano que Spark executa
O otimizador pode tomar decisões baseado em estatísticas e configurações.
Mas isso nos leva a outro ponto.
14. Estatísticas importam
Como o otimizador pode escolher uma estratégia se não conhece os dados?
Imagine:
```text id="wql0qb"
Table A
size = ?
Table B
size = ?
Sem informações adequadas, algumas decisões ficam mais difíceis.
Agora:
```text id="hs72hc"
Table A
≈ 2 TB
Table B
≈ 8 MB
Existe uma informação extremamente relevante para planejamento.
Por isso estatísticas e metadados são importantes em engines modernas.
O otimizador precisa responder perguntas como:
```text id="0sgh2p"
Qual dataset é menor?
Quantas linhas existem?
Qual a cardinalidade?
Como os dados estão distribuídos?
Existe uma estratégia mais barata?
Essa é uma diferença importante entre simplesmente executar código e trabalhar com um **query optimizer**.
---
# 15. O que significa `Exchange` no plano?
Ao analisar:
```python id="dflp99"
df.explain("formatted")
podemos encontrar operadores chamados:
```text id="dkf3oh"
Exchange
Esse termo merece atenção.
Em muitos planos, `Exchange` indica que Spark precisa redistribuir dados.
Em outras palavras:
```text id="2zdpkx"
Exchange
↓
possível movimentação
↓
Shuffle / repartitioning
Por exemplo, podemos encontrar algo conceitualmente semelhante a:
```text id="xykwpj"
SortMergeJoin
:- Sort
: +- Exchange hashpartitioning(customer_id)
:
+- Sort
+- Exchange hashpartitioning(customer_id)
Isso já conta uma história:
```text id="x7lf6m"
Left
↓
Exchange
↓
Sort
┐
├── Join
┘
Sort
↑
Exchange
↑
Right
Agora conseguimos enxergar onde existe movimentação de dados.
16. Compare dois planos
Imagine o primeiro:
```text id="65uh56"
SortMergeJoin
├── Exchange
│ └── Table A
│
└── Exchange
└── Table B
Temos movimentação dos dois lados.
Agora outro:
```text id="edct6z"
BroadcastHashJoin
├── Large Table
│
└── BroadcastExchange
└── Small Table
A história mudou.
Em vez de redistribuir toda a tabela grande, Spark transmite a menor.
Só de olhar o plano físico já começamos a formular hipóteses de performance.
17. Nem todo Join é igualdade
Até agora usamos:
```python id="f32oqr"
df1.join(
df2,
"customer_id"
)
que representa um join por igualdade.
Mas podemos ter condições diferentes:
```python id="0ad0tq"
df1.join(
df2,
df1.valor > df2.limite
)
A estratégia física disponível depende também da condição do join.
Algoritmos baseados em hash, por exemplo, funcionam particularmente bem para determinados joins de igualdade.
Isso significa que:
O tipo da condição também influencia as estratégias que o Spark pode considerar.
18. Tipos de Join continuam importantes
Do ponto de vista lógico, continuamos tendo:
```text id="p9z0z4"
INNER
LEFT
RIGHT
FULL
LEFT SEMI
LEFT ANTI
CROSS
Por exemplo:
```python id="tw5sfr"
pedidos.join(
clientes,
"customer_id",
"left"
)
Mas existe uma distinção importante:
```text id="7ujx51"
TIPO LÓGICO
INNER
LEFT
RIGHT
...
não é a mesma coisa que:
```text id="ml4t0r"
ESTRATÉGIA FÍSICA
BroadcastHashJoin
SortMergeJoin
ShuffleHashJoin
...
Uma coisa responde:
Qual resultado quero?
A outra:
Como Spark vai produzi-lo?
Essa distinção é fundamental.
19. LEFT SEMI JOIN: um recurso muito útil
Imagine que queremos apenas pedidos de clientes ativos.
Podemos pensar:
```python id="p02s5c"
pedidos.join(
clientes_ativos,
"customer_id",
"left_semi"
)
O resultado contém registros do lado esquerdo que possuem correspondência no lado direito.
Não precisamos trazer todas as colunas de `clientes_ativos`.
Conceitualmente:
```text id="zt5quf"
PEDIDOS
1
2
3
4
5
CLIENTES ATIVOS
1
3
5
RESULTADO
1
3
5
Isso pode expressar melhor a intenção quando queremos apenas verificar existência.
20. LEFT ANTI JOIN
Agora imagine que queremos encontrar pedidos cujo cliente não existe em determinada referência.
Podemos utilizar:
```python id="ic8f00"
pedidos.join(
clientes,
"customer_id",
"left_anti"
)
Conceitualmente:
```text id="c3fl7n"
PEDIDOS
1
2
3
4
5
CLIENTES
1
2
4
LEFT ANTI
3
5
Esse tipo de operação é muito útil em:
- validação;
- qualidade de dados;
- reconciliação;
- identificação de registros órfãos;
- pipelines de ingestão.
21. O inimigo volta: Data Skew
Agora imagine um join por:
```text id="hts9yz"
customer_id
com esta distribuição:
```text id="qsz9qk"
customer_id = 1 → 500 milhões
customer_id = 2 → 1 milhão
customer_id = 3 → 900 mil
customer_id = 4 → 800 mil
Depois de particionar pela chave:
```text id="xqovnp"
Partition customer=1
████████████████████████████████████████
Outras partitions
█
█
█
█
Temos novamente:
```text id="c5tjlk"
Task 1 → 30 minutos
Task 2 → 25 segundos
Task 3 → 24 segundos
Task 4 → 27 segundos
O cluster inteiro pode ficar esperando a Task associada à chave dominante.
Joins e skew são uma combinação especialmente perigosa.
22. Um caso clássico: valores NULL
Existe outro detalhe que merece investigação em datasets reais.
Imagine milhões de registros com:
```text id="8atwub"
customer_id = NULL
Dependendo da operação, semântica e tratamento aplicado, valores ausentes podem gerar comportamento indesejado ou simplesmente representar dados que nem deveriam participar daquele processamento.
Talvez possamos filtrar previamente:
```python id="p7oxb5"
pedidos_validos = pedidos.filter(
"customer_id IS NOT NULL"
)
se isso fizer sentido para a regra de negócio.
A regra principal continua:
Não mova dados que não precisam participar da operação.
23. Salting: uma estratégia contra Skew
Em alguns cenários de skew severo, podemos utilizar uma técnica conhecida como:
Salting
Imagine uma chave dominante:
```text id="2m69vg"
customer_id = 123
responsável por uma quantidade enorme de registros.
Em vez de enviar tudo para uma única chave física:
```text id="i15k8p"
123
123
123
123
123
123
podemos criar variações:
```text id="8t2gzf"
123_0
123_1
123_2
123_3
distribuindo parte do trabalho.
Conceitualmente:
```text id="qtdc9k"
ANTES
123 ███████████████████████████████
DEPOIS
123_0 ███████
123_1 ███████
123_2 ███████
123_3 ███████
Depois precisamos tratar corretamente o outro lado do join e eventualmente recombinar resultados.
É uma técnica poderosa.
Mas aumenta complexidade.
Portanto:
Salting é uma ferramenta para um problema específico, não uma configuração padrão.
24. AQE e Skew Join
Aqui o Spark moderno oferece uma evolução importante.
Com:
Adaptive Query Execution
o Spark pode observar estatísticas durante a execução e adaptar determinadas decisões.
Isso inclui mecanismos relacionados a skew em determinados tipos de join.
Conceitualmente:
```text id="cb1d88"
Plano inicial
↓
Shuffle acontece
↓
Spark observa tamanho real
↓
Detecta partition muito grande
↓
AQE
↓
Adapta execução
Isso pode reduzir a necessidade de algumas otimizações manuais.
Mas novamente:
> O AQE ajuda muito, mas não transforma modelagem ruim em arquitetura boa.
Precisamos continuar entendendo nossos dados.
---
# 25. Join Order importa
Imagine três tabelas:
```text id="fw9hwh"
A = 2 TB
B = 500 GB
C = 5 MB
Queremos:
```text id="97eblm"
A JOIN B JOIN C
Existem diferentes maneiras de executar essas operações.
Por exemplo:
```text id="28grjw"
(A JOIN B) JOIN C
ou determinadas reorganizações equivalentes permitidas pela semântica da consulta.
O custo pode variar muito dependendo de:
- seletividade;
- cardinalidade;
- filtros;
- tamanho intermediário;
- estratégias disponíveis.
Engines com otimização baseada em custo podem usar estatísticas para avaliar alternativas.
Esse é outro motivo pelo qual metadados e estatísticas são tão importantes.
26. Filter Before Join
Considere:
```text id="27f3fk"
pedidos = 2 TB
Mas só precisamos:
```text id="0v4owq"
ano = 2026
Se isso representa apenas:
```text id="dw31bq"
200 GB
é muito melhor reduzir o dataset antes da operação cara.
Por exemplo:
```python id="b6qvxw"
pedidos_2026 = pedidos.filter(
"ano = 2026"
)
resultado = pedidos_2026.join(
clientes,
"customer_id"
)
Conceitualmente:
```text id="mdpt6m"
2 TB
↓
FILTER
↓
200 GB
↓
JOIN
em vez de:
```text id="fckdd6"
2 TB
↓
JOIN
↓
FILTER
O otimizador pode aplicar determinadas otimizações automaticamente, mas o princípio continua importante para nosso raciocínio.
27. Select Before Join
Imagine que pedidos possui:
```text id="krvza0"
150 colunas
mas precisamos apenas:
```text id="b7b6f4"
order_id
customer_id
value
Podemos trabalhar explicitamente com:
```python id="mru7mx"
pedidos_reduzidos = pedidos.select(
"order_id",
"customer_id",
"value"
)
e depois:
```python id="0sbrt3"
resultado = pedidos_reduzidos.join(
clientes,
"customer_id"
)
Novamente:
Evite carregar e movimentar dados que não serão utilizados.
Esse princípio simples aparece repetidamente porque funciona.
28. Evite duplicação acidental
Joins também podem causar um problema lógico extremamente sério.
Imagine:
```text id="1odv0g"
LEFT
customer_id=10
1 registro
e:
```text id="3k0a4h"
RIGHT
customer_id=10
5 registros
Depois do join:
```text id="8a3nkr"
1 × 5 = 5 registros
Agora imagine:
```text id="xclds1"
LEFT
customer_id=10
100 registros
e:
```text id="1ivxyy"
RIGHT
customer_id=10
100 registros
Dependendo da semântica do join:
```text id="8fn6kn"
100 × 100
=
10.000 combinações
Isso pode gerar explosão de cardinalidade.
Às vezes o problema atribuído ao Spark não é Spark.
É o modelo de dados.
29. Sempre conheça a cardinalidade da chave
Antes de um join importante, gosto de pensar:
```text id="7gg0hc"
A chave é única no lado direito?
É única no lado esquerdo?
É 1:1?
1:N?
N:N?
Compare:
```text id="yqokny"
1 : 1
com:
```text id="o01zdg"
N : N
O segundo pode multiplicar drasticamente a quantidade de registros.
Essa pergunta é tanto de performance quanto de qualidade de dados.
---
# 30. Um Join pode produzir mais dados do que recebe
Imagine:
```text id="2ud5v5"
Table A
100 milhões de rows
Table B
50 milhões de rows
Não podemos assumir que o resultado terá:
```text id="vz5qmu"
≤ 150 milhões
Um many-to-many pode produzir muito mais.
Por exemplo:
```text id="5xcm4h"
Key X
A → 10.000 rows
B → 20.000 rows
Apenas essa chave pode produzir:
```text id="dhg7jk"
10.000 × 20.000
= 200.000.000 combinações
Esse tipo de explosão pode destruir um pipeline.
Por isso conhecer os dados é tão importante quanto conhecer a API.
---
# 31. Cache antes do Join?
Outra otimização aplicada muitas vezes automaticamente por desenvolvedores é:
```python id="yutl80"
df.cache()
antes de um join.
Mas cache não elimina automaticamente:
```text id="3i9qbt"
Shuffle
nem:
```text id="tcnm7h"
Skew
nem:
```text id="i89a3b"
Join explosion
Cache pode ser extremamente útil quando um dataset caro será reutilizado.
Mas:
```text id="4b2vge"
cache()
não significa:
```text id="l0fp1u"
makeEverythingFast()
Ele também consome recursos.
Como sempre:
> Precisamos saber qual problema estamos tentando resolver.
---
# 32. Um workflow de investigação
Se um join estiver lento, eu começaria assim:
```text id="sff5zz"
JOIN LENTO
↓
explain()
↓
Qual estratégia física?
Depois:
```text id="g51bzq"
SortMergeJoin?
BroadcastHashJoin?
Outra estratégia?
Então:
```text id="ahzcy6"
Existe Exchange?
↓
Quanto Shuffle?
Depois:
```text id="d8ceg9"
As partitions estão equilibradas?
↓
Existe skew?
Depois:
```text id="1fzt0n"
Qual tamanho dos datasets?
E:
```text id="9ccetm"
Qual cardinalidade da chave?
Só depois começaria a alterar configurações.
---
# 33. Um exemplo completo
Imagine:
```python id="9vhlm2"
from pyspark.sql import SparkSession
from pyspark.sql.functions import broadcast
spark = (
SparkSession.builder
.appName("join-example")
.getOrCreate()
)
pedidos = spark.read.parquet(
"/data/pedidos"
)
clientes = spark.read.parquet(
"/data/clientes"
)
pedidos_filtrados = (
pedidos
.filter("ano = 2026")
.select(
"order_id",
"customer_id",
"value"
)
)
resultado = pedidos_filtrados.join(
broadcast(clientes),
"customer_id",
"left"
)
resultado.explain("formatted")
Não quero que a leitura desse código seja apenas:
```text id="phoyxp"
read
filter
select
join
Quero que a leitura mental seja:
```text id="s3ldgr"
READ PEDIDOS
↓
Predicate / Filter
↓
Projection
↓
Menos dados
↓
Large side
│
├───────────────┐
│ │
│ ▼
│ BroadcastHashJoin
│ ▲
│ │
└───── Small CLIENTES
↓
Broadcast
Essa é uma forma muito mais poderosa de pensar Spark.
34. O que eu procuro no explain()
Quando analiso um join, quero responder:
```text id="35nhu6"
Qual operador físico foi escolhido?
Existe Broadcast?
Existe Exchange?
Existe Sort?
Quantos Shuffles aparecem?
Existem filtros antes do Join?
As projeções foram aplicadas?
Qual lado está sendo Broadcast?
O plano parece coerente com o tamanho dos dados?
Não precisamos decorar cada linha do plano.
Precisamos aprender a reconhecer padrões.
---
# 35. API simples, execução complexa
Esse é um dos aspectos que mais gosto no Spark.
Podemos escrever:
```python id="3a48w7"
df1.join(
df2,
"id"
)
Uma linha.
Mas por trás dela podemos ter:
```text id="dspvhm"
Logical Plan
↓
Analyzer
↓
Catalyst
↓
Optimization
↓
Statistics
↓
Physical Planning
↓
Exchange
↓
Shuffle
↓
Sort
↓
Join
↓
Tasks
↓
Executors
Uma API simples esconde um sistema distribuído bastante sofisticado.
E é justamente por isso que entender apenas a sintaxe não é suficiente.
---
# 36. A principal lição deste capítulo
Quando vejo:
```python id="p0zx0k"
A.join(B, "id")
minha primeira pergunta não é:
Qual é a sintaxe do join?
É:
Qual estratégia física o Spark vai usar para juntar esses dados?
Depois:
```text id="t8pvdl"
Qual tamanho de A?
Qual tamanho de B?
Qual cardinalidade?
Existe skew?
Existe filtro possível?
Existe broadcast?
Haverá shuffle?
Quanto dado será movimentado?
Qual plano físico foi escolhido?
Essa mudança de mentalidade é fundamental para trabalhar com Spark em escala.
---
# 37. O modelo mental final
Podemos resumir a decisão assim:
```text id="ozngmm"
JOIN
│
▼
Qual o tamanho?
│
┌───────────┴───────────┐
│ │
Um lado pequeno Ambos grandes
│ │
▼ ▼
Broadcast? Shuffle
│ │
▼ ▼
BroadcastHashJoin SortMergeJoin
│ │
└───────────┬───────────┘
▼
Verificar Skew
│
▼
Verificar Plano
│
▼
Spark UI
│
▼
Medir resultado
Não é uma árvore de decisão absoluta.
É um modelo para raciocinar.
E essa diferença é importante.
Próximo artigo
Até agora falamos várias vezes sobre:
```text id="2fkcb5"
Spark decide...
Spark otimiza...
Spark escolhe...
Mas quem está tomando essas decisões?
No próximo artigo vamos abrir a caixa-preta do:
# Spark SQL
e entender a evolução:
```text id="j1ry29"
RDD
↓
DataFrame
↓
Spark SQL
↓
Logical Plan
↓
Catalyst Optimizer
↓
Physical Plan
↓
Tungsten
↓
Execution
Vamos explorar:
- por que DataFrames mudaram a forma de trabalhar com Spark;
- SparkSession;
- schemas;
- SQL vs. DataFrame API;
- Catalyst Optimizer;
- logical plans;
- physical plans;
- predicate pushdown;
- projection pruning;
- Tungsten;
- Whole-Stage Code Generation;
- AQE;
- e por que APIs declarativas permitem que o Spark faça otimizações que seriam muito mais difíceis com código arbitrário.
É a ponte entre o Spark que eu comecei estudando com RDDs e o Spark moderno usado em plataformas de dados.
← Parte 4: Shuffle — onde a performance começa a ficar cara
Próximo: Parte 6 — Spark SQL e DataFrames: como o Catalyst transforma código em execução distribuída
Top comments (0)