DEV Community: renanpyd

Como aprendi Apache Spark — Parte 7: Memória — Executors, Driver, Cache, Spill e OutOfMemory

renanpyd — Sat, 24 Oct 2026 16:10:33 +0000

Nos artigos anteriores construímos uma visão cada vez mais completa do Apache Spark.

Passamos por:

Data
 ↓
Partitions
 ↓
Tasks
 ↓
Executors
 ↓
Shuffle
 ↓
Joins
 ↓
Spark SQL
 ↓
Catalyst
 ↓
Physical Plan

Mas existe um recurso atravessando praticamente todas essas etapas:

memória.

Quando comecei a estudar problemas de memória no Spark, era tentador resumir tudo a:

OutOfMemory
     ↓
aumentar RAM

Hoje eu vejo esse raciocínio como incompleto.

Um problema de memória pode estar relacionado a:

Driver
Executors
Partitions
Shuffle
Broadcast
Cache
Serialization
Garbage Collection
Python workers
Off-heap/native memory
Memory Overhead

E aumentar memória sem descobrir qual desses componentes está causando o problema pode apenas tornar um Job caro antes de ele falhar novamente.

A pergunta correta não é:

Quanto de memória devo adicionar?

É:

Quem está consumindo memória, por quê, e em qual parte da arquitetura?

1. Memória em uma aplicação Spark

Vamos voltar à arquitetura básica:

                 DRIVER
                    │
                    ▼
             Cluster Manager
                    │
          ┌─────────┼─────────┐
          ▼         ▼         ▼
      Executor 1 Executor 2 Executor 3

Temos dois contextos que precisam ser diferenciados imediatamente:

Driver Memory

Executor Memory

Essa distinção é fundamental.

Porque:

Driver OOM

Executor OOM

podem ter causas completamente diferentes.

2. Driver Memory

O Driver coordena a aplicação.

Ele mantém informações relacionadas a:

aplicação;
Jobs;
Stages;
Tasks;
scheduling;
metadados;
planos de execução;
resultados retornados ao Driver;
estado necessário para coordenar a computação.

Conceitualmente:

                  DRIVER
                     │
      ┌──────────────┼──────────────┐
      ▼              ▼              ▼
   Planning       Scheduling     Metadata
      │              │              │
      └──────────────┼──────────────┘
                     ▼
                  Control

Isso significa que o Driver não deveria ser tratado como destino para todo o dataset distribuído.

Esse ponto parece óbvio.

Até alguém executar:

dados = df.collect()

3. O perigo do `collect()`

Considere:

df.count()

O processamento acontece de maneira distribuída e o resultado retornado é pequeno:

123456789

Agora compare com:

dados = df.collect()

collect() solicita que todos os registros do DataFrame sejam retornados ao Driver.

Conceitualmente:

Executor 1 ──────┐
Executor 2 ──────┤
Executor 3 ──────┼──► DRIVER
Executor 4 ──────┤
Executor 5 ──────┘

Se o dataset possuir:

500 GB

e o Driver tiver:

8 GB

temos um problema bastante previsível.

O Spark é distribuído.

Mas acabamos de pedir:

Traga tudo para uma única máquina.

4. `collect()` não é ruim — o contexto é que importa

Não existe problema em:

df.limit(10).collect()

quando sabemos que o resultado será pequeno.

Também existem cenários legítimos em que precisamos trazer pequenas quantidades de dados ao Driver.

O problema é usar collect() sem compreender:

Distributed Dataset
        ↓
      Driver

A pergunta antes de um collect() deveria ser:

Qual o volume máximo que pode voltar para o Driver?

Essa pergunta evita muitos problemas.

5. Outros caminhos para sobrecarregar o Driver

collect() é o exemplo mais conhecido, mas não é o único.

Problemas de Driver também podem aparecer com:

resultados excessivamente grandes;
estruturas locais gigantes;
excesso de metadados;
quantidade extrema de Tasks;
operações que materializam dados no processo do Driver;
código Python executado localmente;
determinadas operações de visualização ou conversão.

Por exemplo:

pdf = df.toPandas()

merece atenção.

O resultado precisa caber na memória do processo que recebe esse DataFrame local.

Ou seja:

Spark DataFrame distribuído
          ↓
       toPandas()
          ↓
DataFrame local
          ↓
      DRIVER

Se os dados forem grandes, temos novamente uma concentração de memória.

6. Executor Memory

Agora vamos para os Executors.

Eles executam as Tasks.

Conceitualmente:

Executor
   │
   ├── Task 1
   ├── Task 2
   ├── Task 3
   └── Task 4

Durante a execução, memória pode ser necessária para:

processamento;
agregações;
joins;
sorting;
shuffle;
estruturas intermediárias;
cache;
dados serializados;
execução de código;
componentes fora do heap JVM.

É aqui que começam vários problemas clássicos.

7. Uma partição vira trabalho para uma Task

Da Parte 3:

Em uma determinada Stage, cada partição normalmente corresponde ao trabalho de uma Task.

Então existe uma conexão direta:

Partition
    ↓
Task
    ↓
Memory pressure

Imagine:

Partition 1 → 200 MB
Partition 2 → 180 MB
Partition 3 → 220 MB
Partition 4 → 18 GB

A Task responsável pela quarta partição terá um perfil completamente diferente das demais.

Podemos observar:

Task 1 → OK
Task 2 → OK
Task 3 → OK
Task 4 → Memory Pressure

Talvez o problema não seja:

Executor com pouca memória.

Talvez seja:

Uma partição absurdamente maior que as outras.

8. Particionamento e memória são inseparáveis

Imagine um dataset de:

1 TB

distribuído em:

10 partitions

Simplificando muito:

≈ 100 GB por partition

Agora considere:

1000 partitions

Teríamos algo mais próximo de:

≈ 1 GB por partition

Isso não significa que uma Task precise manter toda a partição simultaneamente em memória.

Mas o exemplo mostra uma relação importante:

O tamanho e a distribuição das partições influenciam diretamente a pressão de memória das Tasks.

É por isso que simplesmente aumentar spark.executor.memory pode esconder a causa real.

9. Execution Memory

Durante operações como:

df.groupBy("customer_id").sum("value")

Spark precisa de memória para executar estruturas intermediárias.

Também em:

df.orderBy("timestamp")

ou em determinados joins.

Podemos pensar conceitualmente:

Executor Memory
      │
      ├── Execution
      │      │
      │      ├── Shuffle
      │      ├── Join
      │      ├── Sort
      │      └── Aggregation
      │
      └── Storage
             │
             ├── Cached Data
             └── Persisted Data

No Spark moderno, execution e storage participam de um modelo de gerenciamento de memória integrado.

Isso é importante porque a memória disponível não deve ser imaginada como caixas totalmente independentes e rígidas em todos os momentos.

10. Storage Memory

Agora imagine:

df.cache()

Quando o DataFrame é materializado, Spark pode manter partições para reutilização.

Conceitualmente:

Executor
   │
   ├── Execution Memory
   │
   └── Storage Memory
            │
            ├── Partition A
            ├── Partition B
            └── Partition C

Isso pode ser extremamente útil.

Se o mesmo dataset caro for utilizado várias vezes:

Expensive computation
       ↓
     CACHE
       ↓
 ┌─────┼─────┐
 ▼     ▼     ▼
Use 1 Use 2 Use 3

evitamos recalcular repetidamente a mesma linhagem.

Mas cache também custa memória.

11. `cache()` não significa "deixe rápido"

Esse é outro anti-pattern comum:

df.cache()

ser utilizado como se significasse:

optimize(df)

Não significa.

Imagine um DataFrame de:

800 GB

que será utilizado apenas uma vez.

Fazer cache pode simplesmente:

consumir memória;
provocar eviction;
aumentar pressão sobre o Executor;
competir com execution memory;
não trazer benefício relevante.

Cache faz mais sentido quando existe reuso.

Por exemplo:

base = transformacao_cara(df)

base.cache()

resultado_a = base.groupBy("estado").count()
resultado_b = base.groupBy("produto").sum("valor")
resultado_c = base.filter("status = 'ACTIVE'")

Agora existe um motivo claro:

Expensive Base
     ↓
   CACHE
     ↓
 ┌────┼────┐
 ▼    ▼    ▼
 A    B    C

12. Cache também é Lazy

Considere:

df.cache()

Isso não significa necessariamente:

Neste instante, todas as partições foram computadas e armazenadas.

A materialização ocorre quando uma Action força a computação.

Por exemplo:

df.cache()

df.count()

Agora temos:

cache()
   ↓
mark for persistence
   ↓
count()
   ↓
compute partitions
   ↓
store according to persistence strategy

Lazy evaluation continua presente.

13. `persist()`

Além de:

df.cache()

podemos trabalhar com:

df.persist()

e escolher estratégias de persistência adequadas ao contexto.

Conceitualmente, a persistência pode envolver combinações de:

Memory
Disk
Serialization
Replication

A escolha depende de fatores como:

tamanho;
frequência de reuso;
custo de recomputação;
memória disponível;
tolerância a I/O.

Novamente, não existe uma configuração universalmente melhor.

14. E quando a memória não é suficiente?

Agora chegamos a algo que apareceu na Parte 4:

Spill

Imagine uma Task realizando:

Sort

ou:

Aggregation

ou processamento relacionado a:

Shuffle

Ela utiliza estruturas em memória.

Mas a quantidade necessária ultrapassa o que pode ser mantido eficientemente ali.

Então parte dos dados pode ser enviada para disco.

Conceitualmente:

Task
 ↓
Memory
 ↓
Memory pressure
 ↓
Spill
 ↓
Disk

Depois:

Disk
 ↓
Read
 ↓
Continue processing

Isso permite que determinadas operações continuem mesmo quando tudo não cabe simultaneamente em memória.

Mas existe um preço.

15. Spill troca memória por I/O

Compare:

Memory
  ↓
CPU

com:

Memory
  ↓
Disk Write
  ↓
Disk Read
  ↓
CPU

Disco normalmente é muito mais lento do que memória.

Então grandes volumes de spill podem transformar:

Job

em:

Job lento

Mesmo sem ocorrer um OutOfMemory.

Essa é uma lição importante:

Uma aplicação pode estar sofrendo de pressão de memória sem necessariamente falhar por OOM.

Ela pode simplesmente ficar muito mais lenta.

16. Shuffle + Memory + Spill

Agora conectamos vários artigos.

Considere:

resultado = (
    vendas
    .groupBy("customer_id")
    .sum("valor")
)

Podemos ter:

Partitions
     ↓
Partial Aggregation
     ↓
Shuffle Write
     ↓
Network
     ↓
Shuffle Read
     ↓
Aggregation
     ↓
Memory pressure
     ↓
Spill
     ↓
Disk

Agora um simples:

groupBy()

ganha uma interpretação completamente diferente.

17. Data Skew volta novamente

Imagine:

P1 = 200 MB
P2 = 210 MB
P3 = 190 MB
P4 = 15 GB

Durante uma agregação:

P1 → OK
P2 → OK
P3 → OK
P4 → Spill → Spill → Spill

Ou pior:

P4 → OutOfMemory

A reação inicial pode ser:

Aumentar executor memory.

Talvez funcione.

Mas talvez a pergunta correta seja:

Por que existe uma partição de 15 GB quando as demais possuem aproximadamente 200 MB?

Estamos novamente diante de:

Data Skew

A causa é distribuição.

O sintoma aparece na memória.

18. OutOfMemory não é um único problema

Quando alguém diz:

"Spark deu OOM."

Ainda sabemos muito pouco.

Precisamos perguntar:

Onde?

Por exemplo:

Driver OOM?

ou:

Executor OOM?

ou pressão fora do heap/container?

Também precisamos perguntar:

Em qual Stage?

Em qual operação?

Em todas as Tasks?

Em uma única Task?

Depois de um Shuffle?

Durante Broadcast?

Durante collect()?

Essas perguntas mudam completamente a investigação.

19. Um Driver OOM pode indicar problema de arquitetura

Imagine:

dados = df.collect()

sobre um dataset enorme.

Podemos aumentar:

spark.driver.memory

de:

4 GB

para:

16 GB

Depois:

32 GB

Depois:

64 GB

Talvez o Job finalmente funcione.

Mas a pergunta continua:

Por que estamos trazendo um dataset distribuído gigantesco para um único processo?

Aumentar memória resolveu o sintoma.

Talvez não tenha resolvido o design.

20. Executor OOM também pode esconder outro problema

Agora imagine:

Executor OOM

A reação:

spark.executor.memory
8 GB → 16 GB

Mas talvez exista:

uma partition = 40 GB

Ou:

join skewed

Ou:

broadcast excessivo

Ou várias Tasks concorrentes disputando recursos.

Ou consumo significativo fora do heap principal.

Então:

Executor OOM não significa automaticamente que o Executor precisa apenas de mais heap.

21. `spark.executor.memory`

Uma configuração conhecida é:

spark.executor.memory

Ela define a quantidade de memória heap disponibilizada ao processo Executor JVM.

Por exemplo:

--executor-memory 8g

Mas existe uma armadilha conceitual importante:

A memória total necessária por um Executor não é necessariamente igual ao valor de spark.executor.memory.

Existe memória utilizada fora desse heap.

E isso nos leva diretamente ao tema do material original que estudei:

Memory Overhead

22. O que é Memory Overhead?

Uma aplicação Spark não utiliza memória apenas para o heap principal da JVM.

Podemos pensar de forma simplificada:

Executor Container / Process Environment
│
├── JVM Heap
│
│    └── spark.executor.memory
│
└── Additional / Non-Heap Memory
     │
     ├── JVM overhead
     ├── native memory
     ├── Python processes
     ├── direct buffers
     └── other runtime overhead

É essa segunda região que torna memoryOverhead tão importante.

Dependendo do ambiente e do workload, um Executor pode morrer mesmo quando aparentemente existe heap suficiente.

23. Um exemplo conceitual

Imagine uma configuração:

Executor Heap
8 GB

Isso não significa necessariamente:

Executor total
=
8 GB

O processo também precisa de espaço para outras necessidades.

Podemos pensar:

Container / Pod / Executor allocation

┌───────────────────────────────┐
│                               │
│ JVM Heap                      │
│ spark.executor.memory         │
│                               │
├───────────────────────────────┤
│                               │
│ Memory Overhead               │
│                               │
│ Native / Python / Direct /    │
│ JVM non-heap / other          │
│                               │
└───────────────────────────────┘

Se a região adicional for insuficiente, o processo pode exceder os limites impostos pelo gerenciador de recursos.

24. Por que isso aparece muito com PySpark?

Em PySpark existe uma característica importante.

Nosso código Python não transforma a JVM inteira em Python.

Existe interação entre:

JVM
 ↕
Python worker

Conceitualmente:

Executor JVM
     │
     ├── Spark execution
     │
     └── Python worker
             │
             └── Python code

Esses processos também consomem memória.

Dependendo do workload, bibliotecas e operações utilizadas, isso pode se tornar relevante para o consumo total do Executor.

Portanto, em PySpark precisamos olhar além de:

JVM Heap

25. Memory Overhead no material histórico

Uma das anotações que motivou este capítulo tratava especificamente de problemas de memoryOverhead.

Esse tipo de problema continua conceitualmente relevante.

O que mudou ao longo dos anos foram:

versões do Spark;
nomes e comportamento de determinadas configurações;
resource managers;
defaults;
cálculo automático de overhead;
integração com containers;
Kubernetes;
execução Python;
mecanismos internos de memória.

Por isso não considero uma boa prática pegar uma configuração antiga e simplesmente reproduzi-la hoje.

O conceito continua válido.

A configuração precisa ser analisada no contexto da versão e do ambiente atuais.

26. Heap vs. Off-Heap vs. Overhead

Uma simplificação útil é:

PROCESS MEMORY
     │
     ├── JVM Heap
     │
     ├── JVM Non-Heap
     │
     ├── Direct / Native Memory
     │
     ├── Python Worker Memory
     │
     └── Runtime Overhead

Isso é mais próximo da realidade do que pensar:

memory = spark.executor.memory

Quando um container é encerrado por ultrapassar limite de memória, precisamos descobrir qual parte do consumo cresceu.

27. Garbage Collection

Agora vamos entrar em outro sintoma importante:

GC

A JVM gerencia automaticamente memória de objetos.

Objetos são criados:

Objects
 ↓
Heap

Depois deixam de ser necessários.

O Garbage Collector precisa recuperar esse espaço.

Em um cenário saudável:

Application Work ███████████████████

GC               █

Mas sob forte pressão podemos encontrar algo mais próximo de:

Application ███████

GC          █████████████

Ou seja:

A aplicação passa tempo demais tentando gerenciar memória em vez de processar dados.

28. GC alto é sintoma

Assim como OOM, GC excessivo não deveria gerar automaticamente:

Aumente a heap.

Talvez tenhamos:

objetos demais;
partições grandes;
agregações pesadas;
estruturas intermediárias enormes;
cache excessivo;
serialização inadequada;
concorrência excessiva por Executor.

Precisamos descobrir por que a memória está sob pressão.

29. Mais cores por Executor também afetam memória

Imagine:

Executor
8 cores

Dependendo da configuração, várias Tasks podem executar simultaneamente nesse Executor.

Conceitualmente:

Executor Memory
      │
 ┌────┼────┬────┬────┬────┐
 ▼    ▼    ▼    ▼    ▼    ▼
T1   T2   T3   T4   T5   ...

Cada Task pode precisar de memória.

Então existe uma relação:

mais Tasks concorrentes por Executor
              ↓
mais competição pela memória disponível

Isso mostra por que tuning não deveria analisar:

executor memory

isoladamente.

Precisamos olhar para:

Executor Memory
      +
Executor Cores
      +
Partition Size
      +
Task Behavior

30. Executors gigantes também não são automaticamente melhores

Poderíamos pensar:

1 Executor
64 cores
256 GB RAM

Parece poderoso.

Mas Executors gigantes podem trazer trade-offs relacionados a:

concorrência;
garbage collection;
blast radius de falha;
utilização;
scheduling;
memória compartilhada entre muitas Tasks.

No outro extremo:

64 Executors
1 core
4 GB

também possui trade-offs.

Não existe um tamanho universal.

Precisamos equilibrar:

cores
memory
parallelism
partition size
workload
cluster manager

31. Broadcast também consome memória

Na Parte 5 vimos:

from pyspark.sql.functions import broadcast

resultado = grande.join(
    broadcast(pequena),
    "id"
)

A estratégia pode evitar um shuffle grande.

Mas existe um custo:

Small Table
    ↓
Broadcast
 ┌──┼──┐
 ▼  ▼  ▼
E1 E2 E3

Cada Executor precisa conseguir trabalhar com a estrutura transmitida.

Se tentarmos broadcast de algo grande demais, podemos criar pressão significativa de memória.

Novamente:

Network optimization
        ↕
Memory cost

Sistemas distribuídos são cheios de trade-offs.

32. Cache também compete por recursos

Imagine:

df_a.cache()
df_b.cache()
df_c.cache()
df_d.cache()

Pode parecer:

Quanto mais cache, melhor.

Mas:

Cache A
Cache B
Cache C
Cache D
     ↓
Storage Pressure
     ↓
Eviction / recomputation / resource competition

Cache deveria ser uma decisão baseada em:

reuse frequency
×
recomputation cost
×
available resources

e não um reflexo automático.

33. `unpersist()`

Se um dataset não será mais reutilizado, podemos liberar sua persistência:

df.unpersist()

Isso comunica:

Não preciso mais manter essas partições persistidas.

Em pipelines longos isso pode ser importante.

Por exemplo:

Phase A
 ↓
cache intermediate
 ↓
reuse several times
 ↓
phase completed
 ↓
unpersist
 ↓
free resources for Phase B

Gerenciamento de memória também faz parte do design do pipeline.

34. O Spark UI volta a ser essencial

Se existe problema de memória, não quero começar alterando configurações.

Quero observar.

No Spark UI e métricas relacionadas, procuro sinais como:

Task Duration
GC Time
Shuffle Read
Shuffle Write
Spill
Storage
Executor behavior
Failed Tasks
Skew

Também observo logs.

A pergunta é:

O que aconteceu antes da falha?

35. Um padrão de investigação

Imagine:

Job falhou
   ↓
Executor lost
   ↓
Memory-related error

Meu raciocínio começaria assim:

Qual Stage?
    ↓
Qual operação?
    ↓
Join?
Aggregation?
Sort?
UDF?
Cache?

Depois:

Todas as Tasks falharam?
        ↓
ou apenas algumas?

Se apenas algumas:

Possível skew?

Depois:

Qual tamanho das partitions?
Quanto shuffle?
Quanto spill?
Quanto GC?

Só então:

Heap insuficiente?
Overhead insuficiente?
Particionamento ruim?
Broadcast inadequado?
Concorrência excessiva?

36. Caso 1 — Driver morrendo

Sintoma:

Driver OutOfMemory

Código:

dados = df.collect()

Dataset:

120 GB

Driver:

8 GB

Diagnóstico provável:

Distributed Data
      ↓
collect()
      ↓
Single Process
      ↓
OOM

Solução arquitetural:

Não coletar o dataset inteiro.

Talvez:

df.write.parquet(
    "/data/output"
)

ou continuar o processamento distribuído.

A solução principal não é necessariamente:

8 GB → 128 GB

37. Caso 2 — Uma Task morre repetidamente

Sintoma:

Stage 12

199 Tasks → OK
1 Task    → FAIL

Isso é extremamente informativo.

Talvez tenhamos:

Data Skew

ou uma partição muito maior.

Se aumentarmos memória de todos os Executors:

Cluster cost ↑

para resolver uma única partição anormal, talvez estejamos atacando o lugar errado.

Precisamos investigar a distribuição.

38. Caso 3 — Todas as Tasks sofrem Spill

Imagine:

Task 1 → heavy spill
Task 2 → heavy spill
Task 3 → heavy spill
Task 4 → heavy spill
...

Agora talvez tenhamos um problema sistêmico.

Possíveis causas:

Partitions muito grandes
Memory insuficiente
Join pesado
Sort pesado
Aggregation pesada
Task concurrency

Nesse cenário, ajustar recursos pode realmente fazer parte da solução.

Mas ainda precisamos medir.

39. Caso 4 — OOM depois de Broadcast

Imagine que:

Job funcionava

e alguém adicionou:

broadcast(dim)

Agora Executors começam a morrer.

Pergunta:

Qual o tamanho real de dim?

Talvez:

"small table"

tenha crescido.

Ontem:

50 MB

Hoje:

6 GB

Uma otimização válida no passado pode se tornar um problema quando os dados mudam.

Isso é importante:

Performance depende também da evolução dos dados.

40. Caso 5 — PySpark e memória externa ao heap

Imagine:

spark.executor.memory

aparentemente confortável.

Mas containers continuam sendo encerrados por memória.

Agora precisamos olhar para:

JVM Heap
     +
Python workers
     +
Native memory
     +
Direct buffers
     +
Other overhead

É exatamente nesse tipo de cenário que compreender memoryOverhead deixa de ser detalhe e passa a ser parte do diagnóstico.

41. Não ajuste `memoryOverhead` às cegas

Se um Executor morre por ultrapassar limites externos ao heap, aumentar overhead pode ser apropriado.

Mas ainda precisamos perguntar:

Por que esse consumo existe?

Pode ser:

Python;
bibliotecas nativas;
buffers;
comportamento da JVM;
volume de dados;
operação específica.

A configuração correta depende do workload e do ambiente.

Por isso evito recomendações como:

sempre configure X GB

ou:

sempre use Y%

sem contexto.

Defaults e comportamentos também evoluem entre versões.

42. Memória em Kubernetes

Em ambientes containerizados, a questão fica ainda mais visível.

Podemos pensar:

Kubernetes Pod
      │
      └── Executor
            │
            ├── JVM Heap
            ├── Non-Heap
            ├── Native
            ├── Python
            └── Overhead

O Pod possui limites.

Não basta a JVM achar que possui heap suficiente.

O consumo total do processo e componentes associados precisa respeitar os recursos disponíveis.

Esse é um dos motivos pelos quais problemas modernos de Spark precisam ser analisados também no nível da infraestrutura.

43. Memória em YARN

Em ambientes YARN, existe raciocínio semelhante.

O Executor executa dentro de recursos controlados pelo gerenciador do cluster.

Historicamente, muitos erros de memoryOverhead ficaram conhecidos justamente em ambientes onde o container ultrapassava seu limite mesmo quando a heap configurada parecia aceitável.

O princípio permanece:

JVM Heap
≠
Total Process Memory

44. A configuração depende do ambiente

Hoje Spark pode rodar em diferentes contextos:

Standalone
YARN
Kubernetes
Managed Platforms
Cloud Services

Isso significa que recomendações de memória precisam considerar:

Spark version;
cluster manager;
container limits;
JVM;
Python;
workload;
plataforma gerenciada.

Uma configuração copiada de um artigo antigo pode estar:

correta historicamente

inadequada atualmente

ao mesmo tempo.

Esse é exatamente o tipo de cuidado que estou aplicando ao revisitar meu material antigo nesta série.

45. Um modelo mental melhor para OOM

Em vez de:

OOM
 ↓
More Memory

prefiro:

                    OOM
                     │
          ┌──────────┴──────────┐
          ▼                     ▼
       Driver?               Executor?
          │                     │
          ▼                     ▼
      collect()?          Which Stage?
      toPandas()?              │
      metadata?                ▼
                         Which Operation?
                               │
                    ┌──────────┼──────────┐
                    ▼          ▼          ▼
                  Skew      Shuffle    Broadcast
                    │          │          │
                    └──────────┼──────────┘
                               ▼
                         Partition Size
                               │
                               ▼
                      Heap or Overhead?
                               │
                               ▼
                         Root Cause

Esse modelo evita tratar todos os erros de memória como iguais.

46. Performance tuning começa com observabilidade

Imagine dois engenheiros.

O primeiro vê:

OutOfMemoryError

e imediatamente altera:

spark.executor.memory

O segundo investiga:

Stage
Tasks
Partition distribution
Shuffle
Spill
GC
Executor logs
Physical plan
Data volume

O segundo talvez leve alguns minutos a mais antes de alterar a configuração.

Mas a chance de resolver a causa real é muito maior.

Essa é uma diferença importante entre:

configuration tuning

performance engineering

47. Um checklist para problemas de memória

Quando encontro um problema de memória em Spark, quero responder:

O erro aconteceu no Driver ou Executor?
Qual Stage estava executando?
Qual operação estava acontecendo?
Existe collect()?
Existe toPandas()?
Existe broadcast?
Existe cache excessivo?
As partições estão equilibradas?
Existe data skew?
Qual o tamanho das maiores Tasks?
Existe Shuffle?
Quanto?
Existe Spill?
Quanto?
Quanto tempo está sendo gasto em GC?
Todas as Tasks sofrem ou apenas algumas?
Existem Python workers?
Existe consumo nativo relevante?
O limite do container está sendo ultrapassado?
O problema é heap ou overhead?
A configuração corresponde à versão e ao cluster manager utilizados?

Só depois:

Qual configuração preciso alterar?

48. O que eu evitaria fazer automaticamente

Se o Job estiver lento ou falhando, eu evitaria começar por:

dobrar executor memory

ou:

dobrar driver memory

ou:

aumentar memoryOverhead arbitrariamente

ou:

cachear tudo

ou:

reduzir o número de partitions sem medir

ou:

aumentar o cluster

Todas essas ações podem ser válidas.

Mas nenhuma delas deveria substituir diagnóstico.

49. Memória, performance e custo

Em cloud, existe ainda outra dimensão.

Imagine:

Executor Memory
8 GB

e simplesmente dobramos para:

16 GB

em dezenas ou centenas de Executors.

Talvez o Job fique estável.

Mas talvez tenhamos acabado de aumentar recursos para compensar:

Data Skew

ou:

repartition ruim

ou:

join inadequado

ou:

cache desnecessário

Então temos:

Problema arquitetural
       ↓
Mais infraestrutura
       ↓
Maior custo
       ↓
Problema mascarado

Performance tuning também é FinOps.

50. A conexão entre tudo que vimos até aqui

Agora podemos montar uma visão muito mais completa:

                    DATA
                      │
                      ▼
                 PARTITIONS
                      │
                      ▼
                    TASKS
                      │
                      ▼
                  EXECUTORS
                      │
          ┌───────────┼───────────┐
          ▼           ▼           ▼
       Memory       CPU        Network
          │                       │
          ▼                       ▼
      Aggregation              Shuffle
          │                       │
          └───────────┬───────────┘
                      ▼
                    Spill
                      │
                      ▼
                     Disk
                      │
                      ▼
                 Performance
                      │
                      ▼
                    Cost

Agora adicione:

Data Skew

e podemos gerar:

Uneven Partition
      ↓
Huge Task
      ↓
Memory Pressure
      ↓
Spill / GC
      ↓
Slow Task
      ↓
Straggler
      ↓
Slow Stage

Isso mostra por que os capítulos anteriores estavam conectados.

51. A principal lição deste capítulo

Se eu pudesse resumir memória no Spark em uma frase:

Problemas de memória frequentemente são sintomas de como os dados e a computação foram distribuídos.

Nem sempre.

Às vezes realmente precisamos de mais memória.

Mas antes de aumentar recursos, quero saber:

Quem está consumindo?
        ↓
Onde?
        ↓
Durante qual operação?
        ↓
Por quê?

Só então:

Architecture
     +
Configuration
     +
Resources

entram na solução.

52. Como penso hoje

Meu modelo mental deixou de ser:

Spark OOM
   ↓
Increase Memory

e passou a ser:

Spark OOM
   ↓
Driver or Executor?
   ↓
Heap or Overhead?
   ↓
Which Stage?
   ↓
Which Task?
   ↓
Which Operation?
   ↓
Partition Size?
   ↓
Skew?
   ↓
Shuffle?
   ↓
Spill?
   ↓
Broadcast?
   ↓
Cache?
   ↓
Python / Native Memory?
   ↓
Root Cause
   ↓
Then Tune

Esse raciocínio é muito mais útil do que decorar valores de configuração.

Próximo artigo

Agora entendemos:

Partitions
    ↓
Tasks
    ↓
Memory
    ↓
Shuffle
    ↓
Spill

Mas existe outro conceito fundamental que apareceu desde os primeiros artigos:

Lineage.

Quando uma partição é perdida, como Spark sabe reconstruí-la?

Por que RDD significa:

Resilient Distributed Dataset

Como Spark obtém tolerância a falhas sem simplesmente replicar toda computação intermediária?

E o que acontece quando a linhagem fica grande demais?

No próximo artigo vamos explorar:

RDD Lineage;
DAG;
fault tolerance;
recomputation;
cache vs. recomputation;
checkpoint;
persistência;
falha de Executor;
recuperação de partições;
por que Spark é "resilient";
e como essas ideias evoluíram nas APIs modernas.

Vamos partir deste modelo:

Source
  ↓
Map
  ↓
Filter
  ↓
Join
  ↓
Aggregate
  ↓
Result

para enxergar algo mais profundo:

              LINEAGE
                 │
                 ▼
Source → Transform → Transform → Shuffle → Result
  │          │          │
  └──────────┴──────────┘
             │
      information needed
       for recomputation

Essa é uma das ideias fundamentais que fizeram o Spark ser Spark.

← Parte 6: Spark SQL e DataFrames — do código ao plano de execução

Próximo: Parte 8 — RDD Lineage e tolerância a falhas: como o Spark se recupera quando algo dá errado

Como aprendi Apache Spark — Parte 6: Spark SQL e DataFrames — do código ao plano de execução

renanpyd — Sat, 24 Oct 2026 16:07:14 +0000

Nos primeiros artigos desta série, construímos o Spark de baixo para cima.

Começamos com:

RDD
 ↓
Partitions
 ↓
Tasks
 ↓
Executors

Depois entendemos:

Transformations
      ↓
Lazy Evaluation
      ↓
Jobs
      ↓
Stages
      ↓
Shuffle

E no artigo anterior chegamos aos joins distribuídos e a uma pergunta importante:

Quando escrevemos uma transformação de alto nível, como o Spark decide o que realmente executar?

Essa pergunta nos leva a uma das maiores evoluções da plataforma:

Spark SQL e DataFrames

Foi aqui que minha visão sobre Spark começou a mudar.

Spark deixou de parecer apenas uma API para manipular coleções distribuídas e passou a mostrar algo muito mais interessante:

um engine capaz de compreender a estrutura da computação e otimizar sua execução.

Essa diferença é fundamental.

Com RDDs, dizemos muito sobre como executar uma transformação.

Com DataFrames e SQL, podemos expressar principalmente o que queremos obter.

E quanto mais o Spark entende nossa intenção, mais oportunidades ele tem para otimizar o caminho até o resultado.

1. Antes dos DataFrames: o mundo dos RDDs

Nos primeiros anos do Spark, RDDs eram a principal abstração da plataforma.

Podíamos escrever:

rdd = sc.textFile("/data/vendas.csv")

Depois:

resultado = (
    rdd
    .map(lambda linha: linha.split(","))
    .filter(lambda x: float(x[3]) > 1000)
    .map(lambda x: (x[1], float(x[3])))
)

Para nós, esse código possui significado.

Sabemos que:

x[1] = estado
x[3] = valor

Mas para o Spark, boa parte dessa lógica pode ser simplesmente código arbitrário executado sobre objetos.

Ele não necessariamente possui a mesma compreensão semântica que teria se soubesse:

coluna = valor
tipo = decimal
condição = valor > 1000

Essa diferença é enorme.

2. Estrutura muda tudo

Agora imagine os mesmos dados representados por um DataFrame:

+--------+---------+------+-------+
|order_id|estado   |produto|valor |
+--------+---------+------+-------+
|1001    |SP       |A     |1500.0 |
|1002    |RJ       |B     |800.0  |
|1003    |MG       |C     |2200.0 |
+--------+---------+------+-------+

Agora Spark conhece algo parecido com:

Schema

order_id → integer
estado   → string
produto  → string
valor    → double

E podemos escrever:

resultado = (
    vendas
    .filter("valor > 1000")
    .select(
        "estado",
        "valor"
    )
)

Agora o engine consegue entender muito mais sobre nossa intenção.

Ele sabe que:

valor é uma coluna;
existe uma condição sobre essa coluna;
apenas estado e valor serão necessários posteriormente;
existe um schema;
determinadas regras de otimização podem ser aplicadas.

Essa informação abre uma enorme oportunidade para otimização.

3. De RDD para DataFrame

Uma forma útil de enxergar a evolução é:

RDD
│
├── objetos distribuídos
├── transformations
├── actions
├── partitions
└── controle de baixo nível

DataFrame
│
├── dados estruturados
├── schema
├── expressões
├── plano lógico
└── otimização automática

Isso não significa que DataFrames deixaram de utilizar computação distribuída.

Por baixo deles continuam existindo:

Partitions
Tasks
Stages
Executors
Shuffle
Memory
Network

O que mudou foi a abstração.

Podemos pensar:

DataFrame API
      ↓
Logical Plan
      ↓
Optimizer
      ↓
Physical Plan
      ↓
Distributed Execution
      ↓
Tasks
      ↓
Executors

Essa camada intermediária é extremamente poderosa.

4. Spark SQL

O nome Spark SQL pode causar uma interpretação limitada.

Pode parecer apenas:

"Uma maneira de executar SQL no Spark."

Mas Spark SQL é muito mais do que isso.

Ele representa o módulo de processamento de dados estruturados do Spark.

Podemos escrever:

SELECT
    estado,
    SUM(valor) AS receita
FROM vendas
WHERE ano = 2026
GROUP BY estado

ou utilizar DataFrame API:

resultado = (
    vendas
    .filter("ano = 2026")
    .groupBy("estado")
    .sum("valor")
)

Essas duas formas podem chegar a planos de execução equivalentes ou muito semelhantes.

Isso acontece porque ambas alimentam a mesma infraestrutura de processamento estruturado.

Podemos pensar:

SQL
 ─────────┐
          │
          ▼
       Spark SQL
          ▲
          │
DataFrame API

E depois:

Spark SQL
    ↓
Logical Plan
    ↓
Optimization
    ↓
Physical Plan
    ↓
Execution

5. SparkSession: a entrada moderna

Historicamente, quem estudou versões antigas do Spark encontrou objetos como:

SparkContext
SQLContext
HiveContext

Com a evolução da plataforma, SparkSession tornou-se o principal ponto de entrada para trabalhar com dados estruturados.

Em PySpark:

from pyspark.sql import SparkSession

spark = (
    SparkSession.builder
    .appName("SparkSQLExample")
    .getOrCreate()
)

A partir dela podemos:

df = spark.read.parquet(
    "/data/vendas"
)

executar SQL:

spark.sql("""
    SELECT *
    FROM vendas
""")

criar DataFrames:

df = spark.createDataFrame(
    dados,
    schema
)

e ainda acessar o contexto Spark:

sc = spark.sparkContext

Uma forma de enxergar essa evolução é:

Spark antigo

SparkContext
SQLContext
HiveContext
     ↓

Spark moderno

SparkSession
     ↓
DataFrame / SQL

6. Schema: Spark precisa entender os dados

Imagine:

"1001,SP,1500.00"

Sem estrutura, temos apenas uma sequência de caracteres.

Agora defina:

order_id → integer
estado   → string
valor    → double

Temos semântica.

Em PySpark:

from pyspark.sql.types import (
    StructType,
    StructField,
    IntegerType,
    StringType,
    DoubleType
)

schema = StructType([
    StructField(
        "order_id",
        IntegerType(),
        False
    ),
    StructField(
        "estado",
        StringType(),
        True
    ),
    StructField(
        "valor",
        DoubleType(),
        True
    )
])

Agora Spark possui informações sobre os tipos.

Isso permite:

validação estrutural;
expressões tipadas;
planejamento;
otimizações;
representação mais eficiente.

Schema não é apenas documentação.

Ele faz parte da maneira como o engine entende os dados.

7. Schema inference ou schema explícito?

Podemos pedir para determinadas fontes inferirem o schema.

Por exemplo:

df = (
    spark.read
    .option("header", True)
    .option("inferSchema", True)
    .csv("/data/vendas.csv")
)

Isso é conveniente.

Mas em pipelines de produção, frequentemente existe valor em definir explicitamente o schema.

Por exemplo:

df = (
    spark.read
    .schema(schema)
    .option("header", True)
    .csv("/data/vendas.csv")
)

Agora o contrato esperado está claro.

Isso pode ajudar com:

previsibilidade;
qualidade;
evolução do pipeline;
controle de tipos;
detecção de problemas.

A escolha depende do contexto, mas existe uma lição importante:

Schema é parte do contrato de dados.

8. DataFrame é declarativo

Considere:

resultado = (
    vendas
    .filter("valor > 1000")
    .select(
        "estado",
        "valor"
    )
)

Nós não estamos dizendo:

Pegue exatamente este byte
 ↓
coloque nesta estrutura
 ↓
execute esta instrução
 ↓
mova para esta máquina

Estamos descrevendo uma transformação.

Algo como:

Quero registros onde valor > 1000

e

quero apenas estado e valor

Spark pode então decidir como executar isso.

Essa separação entre:

WHAT

HOW

é uma das ideias centrais dos sistemas declarativos.

9. Lazy Evaluation continua existindo

DataFrames não abandonaram o conceito de lazy evaluation.

Considere:

df = spark.read.parquet(
    "/data/vendas"
)

filtrado = df.filter(
    "valor > 1000"
)

selecionado = filtrado.select(
    "estado",
    "valor"
)

Até aqui estamos principalmente construindo uma descrição da computação.

Então:

selecionado.show()

temos uma Action.

Conceitualmente:

read
 ↓
filter
 ↓
select
 ↓
show()
 ↓
ACTION
 ↓
EXECUTION

Essa execução tardia é importante porque permite que Spark observe o conjunto de transformações antes de executá-las.

E então possa otimizá-las.

10. Do código ao plano

Considere:

resultado = (
    vendas
    .filter("valor > 1000")
    .select(
        "customer_id",
        "valor"
    )
)

Podemos imaginar inicialmente:

DataFrame Operations
        ↓
Unresolved Logical Plan
        ↓
Analyzed Logical Plan
        ↓
Optimized Logical Plan
        ↓
Physical Plans
        ↓
Selected Physical Plan
        ↓
Execution

Vamos abrir cada etapa.

11. Unresolved Logical Plan

Quando escrevemos:

vendas.select(
    "customer_id",
    "valor"
)

Spark inicialmente recebe referências a nomes.

Algo conceitualmente semelhante a:

Project
├── customer_id
└── valor
    ↓
  vendas

Mas ainda precisa verificar:

Existe customer_id?

Existe valor?

De qual relação elas vêm?

Quais são os tipos?

Esse plano ainda precisa ser resolvido.

12. Analysis

Durante a análise, Spark utiliza informações do catálogo e schemas para resolver referências.

Podemos pensar:

"customer_id"
      ↓
Analyzer
      ↓
vendas.customer_id
      ↓
LongType

Se escrevermos:

vendas.select(
    "coluna_que_nao_existe"
)

o Spark não consegue resolver corretamente essa referência.

A análise transforma uma descrição ainda incompleta em um plano semanticamente compreendido.

13. Logical Plan

Depois de resolver as referências, temos uma representação lógica da consulta.

Por exemplo:

Project
    estado
    valor
       │
       ▼
Filter
    valor > 1000
       │
       ▼
Relation
    vendas

Observe que ainda estamos falando sobre:

o que precisa acontecer

e não exatamente:

qual algoritmo físico executará cada etapa

Essa separação é fundamental.

14. Catalyst Optimizer

Agora chegamos a uma das partes mais interessantes:

Catalyst

Catalyst é o framework de otimização de consultas utilizado pelo Spark SQL.

Ele pode aplicar regras ao plano lógico.

Conceitualmente:

Logical Plan
     ↓
Optimization Rules
     ↓
Optimized Logical Plan

Essas regras podem transformar a consulta em outra forma equivalente, mas potencialmente mais eficiente.

Duas otimizações são especialmente intuitivas:

Predicate Pushdown
Projection Pruning

Vamos entendê-las.

15. Predicate Pushdown

Imagine um dataset de:

5 TB

mas queremos apenas:

df.filter(
    "ano = 2026"
)

Uma execução ruim seria:

Read 5 TB
    ↓
Load everything
    ↓
Filter ano = 2026

Se a fonte e o formato suportarem o comportamento adequado, o filtro pode ser levado para mais perto da leitura.

Conceitualmente:

Filter
   ↓
Data Source
   ↓
Read only relevant data

Em vez de:

DATA
 ↓
READ EVERYTHING
 ↓
FILTER

podemos ter algo mais próximo de:

FILTER INFORMATION
        ↓
DATA SOURCE
        ↓
REDUCED READ

Isso pode diminuir drasticamente o volume de dados processado.

16. Partition Pruning

Imagine um Data Lake organizado:

/data/vendas/

ano=2024/
ano=2025/
ano=2026/

Executamos:

df.filter(
    "ano = 2026"
)

Se Spark consegue utilizar a informação de particionamento, não precisa necessariamente ler:

ano=2024
ano=2025

Podemos ter:

/data/vendas/
│
├── ano=2024  ✕
├── ano=2025  ✕
└── ano=2026  ✓

Isso é partition pruning.

Observe como isso se conecta à Parte 3.

Naquele artigo distinguimos:

partições de execução

de:

particionamento de armazenamento

Agora vemos por que essa distinção importa.

17. Projection Pruning

Imagine uma tabela com:

200 colunas

Mas nossa consulta precisa apenas:

df.select(
    "customer_id",
    "valor"
)

Não faz sentido carregar e processar 200 colunas se apenas duas serão necessárias, quando a fonte e o plano permitem evitar isso.

Conceitualmente:

200 columns
     ↓
Projection Pruning
     ↓
2 required columns

Novamente:

Não leia nem mova dados que você não precisa.

Essa ideia apareceu em shuffle, joins e agora aparece no próprio planejamento da consulta.

18. Columnar formats tornam isso ainda mais poderoso

Agora pense em formatos colunares como:

Parquet

Imagine:

customer_id
order_id
product_id
description
category
state
country
value
tax
discount
...

Se precisamos apenas:

customer_id
value

um formato colunar pode permitir que a engine leia apenas as colunas relevantes.

Conceitualmente:

ROW FORMAT

[row completa]
[row completa]
[row completa]

versus:

COLUMNAR

customer_id → [ ... ]
value       → [ ... ]
state       → [ ... ]
product     → [ ... ]

Agora podemos selecionar fisicamente apenas partes necessárias do dataset em determinados cenários.

Isso conecta:

Spark SQL
+
Parquet
+
Projection Pruning
+
Predicate Pushdown

em uma arquitetura muito eficiente.

19. Optimized Logical Plan

Depois das regras de otimização, podemos sair de algo conceitualmente parecido com:

Read everything
      ↓
Filter
      ↓
Select

para algo mais eficiente:

Determine required columns
      ↓
Push applicable filters
      ↓
Read less data
      ↓
Continue execution

O resultado lógico é o mesmo.

O custo físico pode ser muito diferente.

Esse é um dos grandes benefícios de utilizar uma abstração que o engine consegue compreender.

20. Physical Plan

Até agora falamos:

Filter
Project
Join
Aggregate

Mas ainda precisamos transformar isso em operações executáveis.

É aí que chegamos ao:

Physical Plan

Imagine:

pedidos.join(
    clientes,
    "customer_id"
)

O plano lógico diz:

JOIN
customer_id = customer_id

Mas o plano físico precisa decidir:

BroadcastHashJoin?

SortMergeJoin?

Outra estratégia disponível?

Essa é a ponte:

Logical Plan
      ↓
WHAT
      ↓
Physical Plan
      ↓
HOW

Foi exatamente essa diferença que vimos na Parte 5.

21. `explain()` é nossa janela para o optimizer

Considere:

resultado = (
    vendas
    .filter("ano = 2026")
    .groupBy("estado")
    .sum("valor")
)

Podemos executar:

resultado.explain()

Ou:

resultado.explain("formatted")

Também:

resultado.explain("extended")

Dependendo do modo, podemos observar diferentes níveis de informação sobre os planos.

Nosso objetivo não é decorar cada operador.

É conseguir responder perguntas.

22. O que procuro no `explain()`

Quando olho um plano, quero identificar:

Scan
 ↓
Filter
 ↓
Project
 ↓
Exchange?
 ↓
Aggregate
 ↓
Join?
 ↓
Sort?

Principalmente:

Existe Exchange?

porque isso pode indicar redistribuição de dados.

Também quero saber:

O filtro chegou perto da leitura?

As colunas desnecessárias foram removidas?

Qual estratégia de join foi escolhida?

Existe shuffle?

Existe sort?

Existe broadcast?

O plano começa a contar a história da execução.

23. Um exemplo mental

Temos:

resultado = (
    vendas
    .filter("ano = 2026")
    .select(
        "customer_id",
        "estado",
        "valor"
    )
    .groupBy("estado")
    .sum("valor")
)

Uma visão ingênua seria:

Read
 ↓
Filter
 ↓
Select
 ↓
GroupBy
 ↓
Sum

Mas uma visão mais próxima do engine pode ser:

Parquet Scan
      │
      ├── required columns
      │
      └── pushed filters
      ↓
Partial Aggregation
      ↓
Exchange
      ↓
Final Aggregation
      ↓
Result

Agora enxergamos:

Storage Optimization
        +
Distributed Aggregation
        +
Shuffle

no mesmo pipeline.

24. Agregação em duas etapas

Considere:

df.groupBy(
    "estado"
).sum(
    "valor"
)

Uma implementação distribuída inteligente não precisa necessariamente enviar cada registro bruto pela rede antes de somar.

Podemos ter:

Executor A

SP 10
SP 20
RJ 30

     ↓

SP 30
RJ 30

Executor B

SP 50
RJ 20
MG 40

     ↓

SP 50
RJ 20
MG 40

Depois:

SP 30 ──┐
        ├──► SP 80
SP 50 ──┘

Isso reduz o volume movimentado.

É o mesmo princípio que encontramos anteriormente:

Faça o máximo de redução local possível antes de movimentar dados.

25. DataFrame vs. Python UDF

Agora aparece uma consequência importante.

Imagine:

from pyspark.sql.functions import col

resultado = df.withColumn(
    "valor_com_imposto",
    col("valor") * 1.1
)

Spark entende a expressão:

valor * 1.1

Agora imagine encapsular lógica desnecessariamente em uma função Python arbitrária.

O engine pode ter menos capacidade de compreender e otimizar internamente essa lógica, além de poder existir custo adicional associado à execução entre runtimes dependendo do tipo de UDF utilizado.

Isso leva a uma recomendação importante:

Quando possível, prefira funções nativas e expressões do Spark SQL antes de recorrer a UDFs personalizadas.

Não porque UDFs sejam "proibidas".

Mas porque expressões nativas normalmente dão ao engine muito mais informação e oportunidades de otimização.

26. SQL vs. DataFrame API: qual é mais rápido?

Uma pergunta comum:

SQL é mais rápido que DataFrame?

Ou:

DataFrame é mais rápido que SQL?

A resposta interessante é que ambas podem alimentar o mesmo engine e resultar em planos equivalentes.

Compare:

resultado = (
    vendas
    .filter("valor > 1000")
    .groupBy("estado")
    .sum("valor")
)

com:

SELECT
    estado,
    SUM(valor)
FROM vendas
WHERE valor > 1000
GROUP BY estado

Se o plano resultante for equivalente, não existe uma vantagem automática simplesmente por escolher uma sintaxe ou outra.

Uma pergunta melhor é:

Qual plano o Spark gerou?

Por isso:

explain()

é mais interessante do que discutir sintaxe isoladamente.

27. Tungsten

Catalyst trata fortemente da otimização de planos.

Mas Spark também evoluiu muito na maneira como executa computação.

Um nome importante nessa evolução é:

Project Tungsten

Tungsten introduziu uma série de melhorias focadas em eficiência de execução.

Entre os objetivos estavam:

melhor gerenciamento de memória;
representação mais eficiente dos dados;
redução de overhead de objetos;
melhor uso de CPU;
geração de código.

Podemos pensar em duas camadas:

Catalyst
   ↓
Como otimizar o plano?

Tungsten
   ↓
Como executar esse plano eficientemente?

Essa divisão não precisa ser entendida como uma fronteira rígida da implementação atual, mas é um excelente modelo histórico e conceitual.

28. O custo dos objetos

Em aplicações JVM tradicionais, representar milhões ou bilhões de registros como grandes estruturas de objetos pode gerar overhead significativo.

Imagine conceitualmente:

Object
├── header
├── references
├── fields
└── alignment

Multiplique por:

1.000.000.000

de registros.

O overhead começa a importar.

Uma das grandes evoluções do Spark foi trabalhar com representações internas muito mais eficientes para dados estruturados.

Isso ajuda em:

memória;
cache;
CPU;
garbage collection.

Em Big Data, detalhes de representação se tornam problemas de arquitetura.

29. Whole-Stage Code Generation

Outra otimização importante é conhecida como:

Whole-Stage Code Generation

Imagine uma sequência:

Scan
 ↓
Filter
 ↓
Project
 ↓
Aggregate

Uma execução ingênua poderia tratar cada operador como uma camada separada, passando dados entre várias abstrações.

Com geração de código, Spark pode combinar partes compatíveis do pipeline e gerar código especializado para executar várias operações de maneira eficiente.

Conceitualmente:

Operator
   ↓
Operator
   ↓
Operator
   ↓
Operator

pode se tornar algo mais próximo de:

Generated Execution Pipeline

Isso reduz overhead e melhora utilização da CPU em diversos workloads.

30. Catalyst + Tungsten

Agora podemos montar um modelo mais completo:

            SQL / DataFrame API
                    │
                    ▼
               Logical Plan
                    │
                    ▼
                 Catalyst
                    │
                    ▼
          Optimized Logical Plan
                    │
                    ▼
              Physical Plan
                    │
                    ▼
          Efficient Execution
                    │
             Tungsten concepts
                    │
                    ▼
                  Tasks
                    │
                    ▼
                Executors

Essa arquitetura ajuda a explicar por que DataFrames se tornaram tão importantes.

Não estamos apenas utilizando uma API mais bonita.

Estamos dando ao engine informações suficientes para otimizar a computação.

31. Mas o plano inicial ainda pode estar errado

Até aqui parece que Spark consegue decidir tudo antes da execução.

Mas existe um problema:

Estimativas nem sempre correspondem à realidade.

Imagine que Spark espera:

Partition A = 200 MB
Partition B = 200 MB
Partition C = 200 MB

Mas durante a execução encontra:

Partition A = 180 MB
Partition B = 210 MB
Partition C = 12 GB

Temos skew.

O plano foi criado com determinada expectativa.

A realidade é diferente.

É aqui que entra uma das evoluções mais importantes do Spark moderno.

32. Adaptive Query Execution

AQE

Adaptive Query Execution permite que Spark adapte partes do plano utilizando informações coletadas durante a execução.

Conceitualmente:

Initial Plan
     ↓
Execution starts
     ↓
Runtime Statistics
     ↓
AQE
     ↓
Adapted Plan
     ↓
Continue Execution

Essa é uma mudança importante de filosofia.

Antes:

Plan
 ↓
Execute

Agora podemos ter:

Plan
 ↓
Execute
 ↓
Observe
 ↓
Adapt
 ↓
Continue

O engine passa a responder à realidade encontrada durante o processamento.

33. AQE e Shuffle Partitions

Imagine que planejamos:

200 shuffle partitions

mas depois de filtros o dataset ficou muito menor.

Talvez várias dessas partições sejam pequenas demais.

AQE pode ajudar a combinar partições após observar o tamanho real do shuffle.

Conceitualmente:

200 tiny partitions
        ↓
       AQE
        ↓
fewer, larger partitions

Isso pode reduzir:

número de Tasks;
scheduling overhead;
small partitions.

É exatamente o problema que discutimos na Parte 3, agora tratado de forma adaptativa.

34. AQE e Join Strategy

Imagine que inicialmente Spark não consegue concluir que determinado dataset será pequeno.

Mas depois de um filtro:

Table B

Original: 50 GB
     ↓
Filter
     ↓
Runtime: 5 MB

Agora uma estratégia diferente de join pode fazer sentido.

Em determinados cenários, AQE pode adaptar a estratégia física utilizando informações reais coletadas durante a execução.

Isso conecta diretamente a Parte 5 com este artigo.

35. AQE e Skew

Também vimos:

Partition 1 → 200 MB
Partition 2 → 210 MB
Partition 3 → 15 GB

Essa partição enorme pode gerar uma Task extremamente lenta.

AQE possui mecanismos que podem ajudar em determinados cenários de skew, dividindo trabalho de maneira mais adequada.

Isso não significa que skew deixou de ser um problema.

Significa que o engine moderno possui mais ferramentas para lidar com ele.

36. O Spark moderno é cada vez mais adaptativo

Agora nossa visão evoluiu bastante.

Começamos a série pensando:

RDD
 ↓
Transformation
 ↓
Action

Agora estamos pensando:

DataFrame / SQL
       ↓
Logical Plan
       ↓
Catalyst
       ↓
Optimized Plan
       ↓
Physical Plan
       ↓
Execution
       ↓
Runtime Statistics
       ↓
AQE
       ↓
Adapted Execution

Isso mostra o quanto Spark evoluiu.

37. Um pipeline completo

Vamos juntar vários conceitos.

from pyspark.sql import SparkSession
from pyspark.sql.functions import broadcast, col, sum as spark_sum

spark = (
    SparkSession.builder
    .appName("spark-sql-example")
    .getOrCreate()
)

vendas = spark.read.parquet(
    "/data/vendas"
)

estados = spark.read.parquet(
    "/data/estados"
)

vendas_2026 = (
    vendas
    .filter(col("ano") == 2026)
    .select(
        "estado_id",
        "valor"
    )
)

resultado = (
    vendas_2026
    .join(
        broadcast(estados),
        "estado_id"
    )
    .groupBy(
        "estado"
    )
    .agg(
        spark_sum("valor").alias("receita")
    )
)

resultado.explain("formatted")

Agora não quero enxergar apenas o código.

Quero enxergar:

PARQUET SCAN
      ↓
PARTITION / FILTER PRUNING
      ↓
COLUMN PRUNING
      ↓
REDUCED DATASET
      │
      │
      ├──────────────┐
      │              │
      ▼              ▼
LARGE SIDE      SMALL DIMENSION
      │              │
      │          BROADCAST
      │              │
      └───────┬──────┘
              ▼
      BROADCAST HASH JOIN
              ↓
      PARTIAL AGGREGATION
              ↓
           SHUFFLE
              ↓
       FINAL AGGREGATION
              ↓
            RESULT

Isso é uma leitura muito mais próxima de Engenharia de Dados distribuída.

38. E o Spark UI?

explain() mostra o plano.

Spark UI mostra o que aconteceu durante a execução.

Essa distinção é muito útil.

explain()
    ↓
O que Spark planejou?

Spark UI
    ↓
O que aconteceu?

Combinando os dois:

PLAN
  +
RUNTIME
  =
TROUBLESHOOTING

No plano podemos identificar:

joins;
exchanges;
scans;
filters;
aggregates.

No Spark UI podemos investigar:

duração;
Tasks;
shuffle;
spill;
skew;
memória;
executors.

Essa combinação é extremamente poderosa.

39. Um método de troubleshooting

Quando um pipeline DataFrame/SQL está lento, gosto deste raciocínio:

PIPELINE LENTO
      ↓
explain()
      ↓
Qual plano físico?
      ↓
Existe Exchange?
      ↓
Qual Join?
      ↓
Qual Scan?
      ↓
Filtros foram aplicados?
      ↓
Colunas desnecessárias?

Depois vou para runtime:

Spark UI
    ↓
Qual Stage está lento?
    ↓
Quanto Shuffle?
    ↓
Existe Spill?
    ↓
Existe Skew?
    ↓
Como estão as Tasks?
    ↓
Como estão os Executors?

E só depois:

Alterar configuração

Isso evita o anti-pattern:

Job lento
   ↓
Google
   ↓
copiar spark.conf aleatória
   ↓
reiniciar
   ↓
torcer

40. Configuração não substitui arquitetura

Podemos encontrar centenas de parâmetros Spark.

Mas muitas vezes o maior ganho vem de algo muito mais fundamental.

Por exemplo:

ANTES

Read 5 TB
   ↓
Join 5 TB
   ↓
Filter

versus:

DEPOIS

Read
 ↓
Prune
 ↓
Filter
 ↓
500 GB
 ↓
Join

Ou:

ANTES

Large
  ↓
Shuffle
  ↓
Large

versus:

DEPOIS

Large
  +
Small Broadcast

Nenhuma configuração de memória substitui completamente uma estratégia melhor.

41. Por que DataFrames mudaram tanto o Spark

Agora podemos responder.

Com código arbitrário sobre RDDs, Spark possui menos conhecimento sobre a semântica de determinadas operações.

Com DataFrames e SQL, ele consegue trabalhar com:

Schema
Expressions
Columns
Predicates
Joins
Aggregations
Statistics
Logical Plans
Physical Plans

Isso permite:

Optimization

E quanto mais informação o engine possui, maior sua capacidade de escolher uma execução eficiente.

Essa é uma lição que vai além do Spark:

Abstrações de alto nível podem gerar execução de baixo nível mais eficiente quando fornecem informação suficiente ao otimizador.

42. RDDs ainda importam?

Sim.

Mesmo que grande parte do desenvolvimento moderno utilize:

DataFrames
Spark SQL

entender RDDs continua sendo extremamente útil.

Porque eles ensinam os fundamentos:

Partitions
Transformations
Actions
Lazy Evaluation
Lineage
Shuffle
Distributed Execution

Sem esses conceitos, podemos utilizar DataFrames sem entender por que determinada operação ficou lenta.

A evolução não foi:

RDD errado
   ↓
DataFrame certo

Foi:

Fundamentos distribuídos
        ↓
Abstrações estruturadas
        ↓
Mais oportunidades de otimização

43. O que mudou na minha forma de enxergar Spark

No começo, eu via:

df.groupBy("estado").sum("valor")

Hoje eu tento enxergar:

DataFrame Expression
       ↓
Logical Plan
       ↓
Catalyst
       ↓
Optimized Plan
       ↓
Physical Plan
       ↓
Partial Aggregation
       ↓
Exchange
       ↓
Shuffle
       ↓
Final Aggregation
       ↓
Tasks
       ↓
Executors

Essa mudança de perspectiva é uma das coisas mais importantes que aprender Spark me trouxe.

44. A principal lição deste capítulo

Se eu tivesse que resumir Spark SQL em uma única ideia, seria:

Quanto mais o Spark entende sobre a estrutura e a intenção da computação, mais oportunidades ele possui para otimizar sua execução.

É por isso que:

Schema

importa.

É por isso que:

DataFrames

importam.

É por isso que:

Catalyst

importa.

E é por isso que:

explain()

deveria fazer parte da caixa de ferramentas de quem trabalha seriamente com Spark.

Nosso modelo mental agora é:

                SQL / DataFrame
                       │
                       ▼
                 Logical Plan
                       │
                       ▼
                    Catalyst
                       │
                       ▼
               Optimized Plan
                       │
                       ▼
                Physical Plan
                       │
                       ▼
              Efficient Execution
                       │
                       ▼
               Distributed Tasks
                       │
                       ▼
              Runtime Statistics
                       │
                       ▼
                      AQE

Agora já não estamos apenas executando transformações.

Estamos trabalhando com um query engine distribuído e adaptativo.

Próximo artigo

Existe um assunto que apareceu várias vezes nesta série:

Memory

Falamos sobre memória em:

Executors;
Driver;
cache;
broadcast;
shuffle;
spill;
partições;
joins.

Mas ainda não abrimos essa caixa.

No próximo artigo vamos entender:

Memória no Apache Spark

Vamos explorar:

memória do Driver;
memória dos Executors;
execution memory;
storage memory;
cache e persist;
garbage collection;
spill para disco;
spark.executor.memory;
memory overhead;
por que simplesmente aumentar RAM nem sempre resolve;
OutOfMemoryError;
diferença entre problemas de Driver e Executor;
como partições influenciam memória;
e como diagnosticar problemas usando Spark UI e métricas.

Vamos conectar:

Partitions
     ↓
Tasks
     ↓
Memory
     ↓
Shuffle
     ↓
Spill
     ↓
Disk
     ↓
Performance

É um tema que merece um capítulo próprio porque muitos problemas atribuídos ao "Spark lento" são, na verdade, problemas de distribuição, memória ou pressão causada pelo plano de execução.

← Parte 5: Joins — Shuffle, Broadcast e estratégias de execução

Próximo: Parte 7 — Memória no Apache Spark: Executors, Driver, Cache, Spill e OutOfMemory

Como aprendi Apache Spark — Parte 5: Joins — Shuffle, Broadcast e estratégias de execução

renanpyd — Sat, 24 Oct 2026 16:00:38 +0000

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

```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

```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?

```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

Como aprendi Apache Spark — Parte 4: Shuffle — onde a performance começa a ficar avançada

renanpyd — Fri, 23 Oct 2026 14:41:03 +0000

No artigo anterior, vimos que existe uma relação fundamental no Apache Spark:

Dados
  ↓
Partitions
  ↓
Tasks
  ↓
Parallelism
  ↓
Performance

Também vimos que nem sempre os dados permanecem nas mesmas partições durante todo o processamento.

Operações como:

df.repartition(...)

df.groupBy(...)

df.join(...)

podem exigir uma redistribuição dos dados.

E é nesse momento que aparece um dos conceitos mais importantes para entender performance no Spark:

Shuffle

Quando comecei a entender o que realmente acontecia durante um shuffle, vários problemas de performance que antes pareciam aleatórios começaram a fazer sentido.

Porque shuffle não significa apenas:

"Spark reorganizou algumas linhas."

Em um ambiente distribuído, reorganizar dados pode significar:

serializar
    ↓
escrever
    ↓
transferir pela rede
    ↓
ler
    ↓
desserializar
    ↓
agrupar
    ↓
continuar processando

E tudo isso tem custo.

1. O que é Shuffle?

Vamos começar com um exemplo simples.

Imagine que temos dados distribuídos entre três partições:

Partition 1

customer=A
customer=B
customer=C


Partition 2

customer=A
customer=D
customer=B


Partition 3

customer=C
customer=A
customer=D

Agora queremos contar quantos registros existem para cada cliente:

resultado = (
    df
    .groupBy("customer")
    .count()
)

Para calcular:

A → 3
B → 2
C → 2
D → 2

Spark precisa reunir registros relacionados à mesma chave.

Mas os registros de customer=A, por exemplo, estão espalhados entre diferentes partições.

Temos:

P1 → A
P2 → A
P3 → A

Em algum momento esses dados precisam ser redistribuídos.

Conceitualmente:

ANTES

Partition 1     Partition 2     Partition 3

A               A               C
B               D               A
C               B               D


                 ↓
              SHUFFLE
                 ↓


DEPOIS

Partition A     Partition B     Partition C

A               B               C
A               B               C
A               D               D

A distribuição exata depende da estratégia e do particionamento utilizados, mas a ideia central é esta:

Shuffle redistribui dados entre partições para permitir que uma etapa posterior da computação seja executada.

2. Por que Shuffle é caro?

Se todos os dados estivessem dentro de um único processo, reorganizar registros seria relativamente simples.

Mas Spark é distribuído.

Imagine:

Machine A
Executor 1

Machine B
Executor 2

Machine C
Executor 3

Agora precisamos mover dados de uma máquina para outra:

Executor 1 ──────────► Executor 2
     │                      ▲
     │                      │
     ▼                      │
Executor 3 ─────────────────┘

Isso pode envolver:

CPU;
serialização;
memória;
disco;
rede;
desserialização;
gerenciamento de buffers;
sincronização entre etapas.

Portanto, shuffle pode transformar uma operação aparentemente simples em uma operação distribuída relativamente cara.

3. Narrow vs. Wide Dependencies

Na Parte 3 vimos rapidamente a diferença entre operações narrow e wide.

Agora ela se torna essencial.

Narrow dependency

Imagine:

df.filter("valor > 100")

Cada partição pode executar o filtro sobre seus próprios registros.

Partition 1 ───► Filter ───► Partition 1'

Partition 2 ───► Filter ───► Partition 2'

Partition 3 ───► Filter ───► Partition 3'

Não precisamos necessariamente enviar registros de uma partição para outra.

Outro exemplo:

rdd.map(lambda x: x * 2)

Conceitualmente:

P1 ───► P1'
P2 ───► P2'
P3 ───► P3'

Isso é muito diferente de uma operação que precisa reorganizar os dados globalmente.

4. Wide dependency

Agora considere:

df.groupBy("customer_id").count()

Os registros do mesmo cliente podem estar em qualquer partição.

Temos algo como:

P1 ──┐       ┌──► P1'
     ├───────┤
P2 ──┼───────┼──► P2'
     ├───────┤
P3 ──┘       └──► P3'

Uma partição de saída pode depender de dados provenientes de várias partições anteriores.

Isso caracteriza uma dependência ampla — wide dependency.

E frequentemente significa:

Wide Dependency
       ↓
    Shuffle
       ↓
Stage Boundary

Esse relacionamento é uma das chaves para entender o DAG de uma aplicação Spark.

5. Shuffle cria fronteiras entre Stages

Na Parte 2 vimos:

Action
   ↓
 Job
   ↓
Stages
   ↓
Tasks

Agora conseguimos entender melhor como Spark pode separar determinados Stages.

Considere:

resultado = (
    df
    .filter("status = 'ACTIVE'")
    .groupBy("country")
    .count()
)

resultado.show()

Podemos imaginar:

STAGE 1

Read
 ↓
Filter
 ↓
Partial Processing
 ↓

========= SHUFFLE =========

 ↓
STAGE 2

Aggregation
 ↓
Result

O shuffle cria uma dependência entre conjuntos de Tasks.

As Tasks da etapa posterior precisam dos dados produzidos pela etapa anterior.

Isso forma uma fronteira natural na execução.

6. Shuffle Write e Shuffle Read

Esse processo pode ser dividido conceitualmente em duas partes:

Stage anterior
      │
      ▼
SHUFFLE WRITE
      │
      ▼
   Network
      │
      ▼
SHUFFLE READ
      │
      ▼
Próximo Stage

Shuffle Write

As Tasks do Stage anterior produzem dados que serão consumidos posteriormente.

Esses dados precisam ser organizados para as partições de destino.

Podemos imaginar:

Task 1
  │
  ├── dados para Partition A
  ├── dados para Partition B
  └── dados para Partition C

Task 2
  │
  ├── dados para Partition A
  ├── dados para Partition B
  └── dados para Partition C

Esses resultados formam a saída do shuffle.

Shuffle Read

No Stage seguinte, as Tasks precisam buscar os blocos correspondentes à sua partição.

Por exemplo:

                  Task A

Task 1 output ──────┐
Task 2 output ──────┼──► Partition A
Task 3 output ──────┘

Agora começamos a entender métricas encontradas no Spark UI:

Shuffle Read
Shuffle Write

Elas não são apenas números abstratos.

Representam movimentação real de dados durante a computação distribuída.

7. Um exemplo com `groupBy`

Considere:

resultado = (
    df
    .groupBy("estado")
    .count()
)

Antes da agregação:

Executor 1

SP
RJ
SP
MG


Executor 2

MG
SP
BA
RJ


Executor 3

SP
BA
SP
MG

Para calcular os totais por estado, os registros precisam ser reorganizados de acordo com a chave.

Algo conceitualmente semelhante a:

                  SHUFFLE

Executor 1 ─────────┐
Executor 2 ─────────┼──► Partition SP
Executor 3 ─────────┘


Executor 1 ─────────┐
Executor 2 ─────────┼──► Partition RJ
Executor 3 ─────────┘


Executor 1 ─────────┐
Executor 2 ─────────┼──► Partition MG
Executor 3 ─────────┘

Depois disso, cada partição pode executar sua parte da agregação.

8. `groupByKey()` vs. `reduceByKey()`

No mundo dos RDDs, um exemplo clássico para entender shuffle é comparar:

rdd.groupByKey()

com:

rdd.reduceByKey(...)

Imagine:

("spark", 1)
("spark", 1)
("spark", 1)
("python", 1)

Com groupByKey(), os valores associados às chaves precisam ser agrupados para então serem processados.

Conceitualmente:

spark → [1, 1, 1]
python → [1]

Agora considere:

rdd.reduceByKey(lambda a, b: a + b)

Spark pode realizar agregações locais antes de transferir os dados entre partições.

Imagine:

PARTITION 1

spark → 1
spark → 1
python → 1

Antes de enviar:

spark → 2
python → 1

Outra partição:

PARTITION 2

spark → 1
spark → 1

pode produzir:

spark → 2

Então, em vez de mover todos os registros individualmente, podemos movimentar resultados parcialmente agregados.

Partition 1 → spark:2
Partition 2 → spark:2

             ↓

          spark:4

Essa é uma otimização extremamente importante.

A lição é:

Quanto menos dados precisarmos movimentar, melhor.

9. Map-side aggregation

O conceito anterior é frequentemente associado à ideia de realizar parte da agregação antes do shuffle.

Visualmente:

SEM AGREGAÇÃO LOCAL

1
1
1
1
1
1
1
1
      ↓
   NETWORK
      ↓
sum()

versus:

COM AGREGAÇÃO LOCAL

Partition A
1 + 1 + 1 + 1
      ↓
      4


Partition B
1 + 1 + 1 + 1
      ↓
      4


4 ───┐
     ├──► NETWORK ───► 8
4 ───┘

Mover:

4
4

é muito mais eficiente do que mover todos os registros originais.

Essa ideia aparece de diferentes formas nas otimizações do Spark.

10. Joins também podem gerar Shuffle

Agora chegamos a um caso extremamente comum em Engenharia de Dados.

Imagine:

pedidos.join(
    clientes,
    "customer_id"
)

Temos:

PEDIDOS

customer_id=1
customer_id=5
customer_id=3
customer_id=1

CLIENTES

customer_id=1
customer_id=2
customer_id=3
customer_id=5

Para fazer o join, registros com a mesma chave precisam se encontrar.

Conceitualmente:

Pedidos ──────┐
              │
              ▼
           SHUFFLE
              │
              ▼
Clientes ─────┘

customer_id=1 ──► mesma partition
customer_id=2 ──► mesma partition
customer_id=3 ──► mesma partition
...

Dependendo da estratégia de join, isso pode significar mover grandes volumes de dados pela rede.

E aqui surge uma pergunta importante:

E se uma das tabelas for muito pequena?

Talvez não faça sentido redistribuir os dois lados.

É aí que entra uma técnica que veremos em detalhes mais adiante:

Broadcast Join

Em vez de movimentar duas tabelas grandes:

Large Table ──► Shuffle
Large Table ──► Shuffle

podemos, em determinados cenários, distribuir uma tabela pequena para os Executors:

                 Small Table
                /     |     \
               ▼      ▼      ▼
          Executor Executor Executor

             Large Dataset

Assim podemos evitar um shuffle grande de um dos lados.

Essa é uma das otimizações mais importantes em joins Spark.

11. Shuffle não usa apenas rede

É comum resumir:

"Shuffle é caro porque usa rede."

Isso é verdade, mas incompleto.

Shuffle pode envolver:

                    SHUFFLE
                       │
        ┌──────────────┼──────────────┐
        ▼              ▼              ▼
       CPU           Memory          Disk
        │              │              │
 serialization       buffers        spill
        │              │              │
        └──────────────┼──────────────┘
                       ▼
                    Network

Ou seja, um shuffle pesado pode pressionar vários recursos simultaneamente.

12. Quando memória não é suficiente: Spill

Imagine que uma Task precisa organizar uma quantidade significativa de dados durante uma agregação ou ordenação.

Idealmente, parte desse trabalho acontece em memória.

Mas a memória é finita.

Quando Spark não consegue manter todo o estado necessário em memória, pode precisar utilizar disco.

Esse comportamento é conhecido como:

Spill

Conceitualmente:

Dados
  ↓
Memory
  ↓
Memory insuficiente
  ↓
Disk Spill

Depois, esses dados podem precisar ser lidos novamente.

Temos:

Memory
  ↓
Disk Write
  ↓
Disk Read
  ↓
Continue Processing

Agora o custo aumentou.

Além de CPU e rede, temos I/O em disco.

13. Memory Spill vs. Disk Spill

Ao analisar métricas de execução, podemos encontrar indicadores relacionados a spill.

A ideia principal é perceber que:

Uma Task precisou usar disco porque o processamento intermediário ultrapassou o que podia ser tratado eficientemente em memória.

Isso pode acontecer em operações como:

agregações;
joins;
sorts;
shuffles;
processamento de partições muito grandes.

Se encontramos grandes volumes de spill, precisamos investigar.

Não significa automaticamente:

"adicione mais memória."

Pode significar:

partições grandes demais;
skew;
estratégia de join inadequada;
volume excessivo de shuffle;
configuração inadequada;
algoritmo inadequado.

Performance tuning exige descobrir a causa.

14. Shuffle + Data Skew = combinação perigosa

Na Parte 3 vimos Data Skew.

Agora podemos entender por que ele é especialmente problemático durante shuffle.

Imagine:

customer_id

A → 100.000.000 registros
B → 500.000
C → 400.000
D → 300.000

Depois do shuffle por customer_id:

Partition A █████████████████████████████████████████

Partition B █

Partition C █

Partition D █

Agora temos Tasks como:

Task A → 25 minutos

Task B → 20 segundos

Task C → 18 segundos

Task D → 15 segundos

Depois de alguns segundos:

Executor 1 → █████████████████ trabalhando

Executor 2 → idle

Executor 3 → idle

Executor 4 → idle

O cluster possui recursos.

Mas uma única partição concentra grande parte do trabalho.

Esse é um dos padrões mais importantes para reconhecer no Spark UI.

15. Stragglers

Tasks que demoram muito mais que as demais são frequentemente chamadas de stragglers.

Imagine:

Stage 17

Task 001 → 21 s
Task 002 → 19 s
Task 003 → 22 s
Task 004 → 20 s
Task 005 → 18 s
...
Task 199 → 20 s

Task 200 → 14 min

O Stage precisa esperar:

Task 200

Então:

199 Tasks concluídas
         +
1 Task extremamente lenta
         =
Stage ainda executando

O problema pode ser skew.

Mas também pode ser:

hardware;
I/O;
garbage collection;
spill;
rede;
tamanho da partição;
executor problemático.

Por isso precisamos analisar métricas antes de concluir.

16. `repartition()` é um Shuffle explícito

Agora podemos revisitar a Parte 3.

Quando executamos:

df.repartition(200)

Spark precisa redistribuir os registros entre novas partições.

Conceitualmente:

ANTES

P1 ─────┐
P2 ─────┤
P3 ─────┼────► SHUFFLE
P4 ─────┤
P5 ─────┘

                 ↓

DEPOIS

P1'
P2'
P3'
...
P200'

Isso explica por que adicionar repartition() arbitrariamente pode prejudicar performance.

Talvez tenhamos:

Read
 ↓
repartition()
 ↓
SHUFFLE
 ↓
groupBy()
 ↓
SHUFFLE
 ↓
join()
 ↓
SHUFFLE

Agora imagine isso sobre vários terabytes.

Estamos movimentando uma quantidade enorme de dados entre máquinas.

17. Shuffle e serialização

Dados não simplesmente "viajam" pela rede como objetos Python ou JVM abstratos.

Eles precisam ser representados de uma forma adequada para transferência e armazenamento.

Simplificando:

Object
  ↓
Serialization
  ↓
Bytes
  ↓
Network / Disk
  ↓
Deserialization
  ↓
Object

Isso também custa CPU.

Em workloads grandes, detalhes aparentemente pequenos começam a se multiplicar por:

milhões
bilhões
trilhões

de registros.

É por isso que otimizações distribuídas frequentemente tratam não apenas de algoritmos, mas também de:

representação;
serialização;
compressão;
layout;
movimentação de dados.

18. Shuffle e compressão

Existe um trade-off interessante.

Podemos comprimir dados antes de movimentá-los.

Conceitualmente:

100 GB
  ↓
Compression
  ↓
30 GB
  ↓
Network

Reduzimos o volume transferido.

Mas agora precisamos gastar CPU:

CPU → compress
Network → transfer
CPU → decompress

Temos um trade-off:

CPU
 ↕
Network

Em sistemas distribuídos, otimização frequentemente significa escolher onde queremos pagar o custo.

19. Como identificar Shuffle no Spark UI

Agora o Spark UI começa a ficar ainda mais interessante.

Podemos observar métricas como:

Shuffle Read
Shuffle Write
Input
Output
Task Duration
GC Time
Spill

Imagine um Stage:

Input:          200 GB
Shuffle Write:  180 GB

E o próximo:

Shuffle Read:   180 GB
Output:          20 GB

Isso nos conta uma história.

Talvez tenhamos movimentado:

180 GB

para produzir:

20 GB

Então surge uma pergunta:

Poderíamos reduzir os dados antes do shuffle?

Talvez um filtro pudesse acontecer antes.

Talvez uma projeção pudesse eliminar colunas desnecessárias.

Talvez a estratégia de join pudesse mudar.

Esse é o tipo de pergunta que transforma métricas em otimização.

20. Filter Early

Considere:

resultado = (
    df
    .groupBy("customer_id")
    .count()
)

sobre:

1 TB

Agora imagine que só precisamos de registros:

status = 'ACTIVE'

Se filtrarmos antes:

resultado = (
    df
    .filter("status = 'ACTIVE'")
    .groupBy("customer_id")
    .count()
)

e apenas 20% dos registros forem ativos, conceitualmente podemos passar de:

1 TB
 ↓
SHUFFLE

para algo mais próximo de:

1 TB
 ↓
FILTER
 ↓
200 GB
 ↓
SHUFFLE

Reduzimos drasticamente o volume que precisa atravessar a operação cara.

Essa é uma ideia simples e poderosa:

Reduza os dados antes de movimentá-los, quando semanticamente possível.

21. Project Early

O mesmo raciocínio vale para colunas.

Imagine um dataset com:

100 colunas

mas nosso processamento precisa apenas de:

customer_id
valor
estado

Podemos selecionar cedo:

df = df.select(
    "customer_id",
    "valor",
    "estado"
)

Conceitualmente:

100 columns
     ↓
SELECT
     ↓
3 columns
     ↓
SHUFFLE

Dependendo do plano de execução e do formato da fonte, isso também pode reduzir bastante a quantidade de dados manipulada.

A ideia continua:

Não mova dados que você não precisa.

22. Aggregate Early

Quando possível, também podemos reduzir cardinalidade antes de determinadas movimentações.

Imagine:

1 bilhão de registros

que podem ser parcialmente agregados localmente antes de uma redistribuição.

Se conseguirmos transformar:

1.000.000.000 rows

em:

10.000.000 partial results

antes da transferência, temos uma diferença enorme.

Essa é a lógica por trás de várias estratégias de agregação distribuída.

23. O melhor Shuffle é o Shuffle que não precisamos fazer

Essa frase resume boa parte do tuning de sistemas distribuídos:

Movimentação de dados é cara.

Isso não significa que devemos eliminar todos os shuffles.

Muitos são necessários.

Um:

groupBy()

pode exigir redistribuição.

Um:

join()

pode exigir redistribuição.

Um:

repartition()

existe justamente para redistribuir.

A pergunta correta não é:

"Como elimino todos os shuffles?"

É:

"Esse shuffle é necessário e, se for, estou movimentando apenas os dados necessários?"

24. Um exemplo de pipeline ruim

Imagine:

df = spark.read.parquet("/data/events")

df = df.repartition(5000)

df = df.select("*")

resultado = (
    df
    .groupBy("customer_id")
    .count()
)

Podemos ter:

READ
 ↓
REPARTITION
 ↓
SHUFFLE
 ↓
GROUP BY
 ↓
SHUFFLE
 ↓
RESULT

Agora suponha que só precisamos de eventos de compra.

Podemos reconsiderar:

df = (
    spark.read.parquet("/data/events")
    .filter("event_type = 'purchase'")
    .select("customer_id")
)

resultado = (
    df
    .groupBy("customer_id")
    .count()
)

Conceitualmente:

READ
 ↓
FILTER
 ↓
PROJECT
 ↓
menos dados
 ↓
SHUFFLE necessário
 ↓
AGGREGATION

Não estamos "fazendo Spark ficar rápido".

Estamos simplesmente reduzindo trabalho desnecessário.

Essa distinção é importante.

25. Mais memória nem sempre resolve Shuffle

Um padrão comum de troubleshooting é:

Job lento
   ↓
aumentar memória

Depois:

continua lento
   ↓
aumentar memória novamente

Mas talvez o problema real seja:

2 TB de Shuffle

Adicionar memória pode ajudar em alguns aspectos.

Mas não elimina necessariamente:

movimentação de rede;
serialização;
skew;
quantidade de dados;
algoritmo inadequado.

Talvez o problema precise ser atacado no plano da computação.

26. Mais Executors também não resolvem tudo

Outro pensamento comum:

"Vou dobrar o cluster."

Podemos passar de:

10 Executors

para:

20 Executors

Mas se o problema for uma chave extremamente skewed:

customer_id = 123

80% dos registros

podemos continuar tendo:

1 Task gigantesca

e vários Executors ociosos.

Escalar infraestrutura sem entender o gargalo pode simplesmente aumentar custo.

27. Performance e custo caminham juntos

Em ambientes cloud, essa relação é ainda mais importante.

Imagine:

Job A

Cluster: 20 workers
Tempo: 2 horas

Depois de reduzir shuffles desnecessários:

Job A

Cluster: 20 workers
Tempo: 35 minutos

Não ganhamos apenas velocidade.

Reduzimos consumo de:

CPU;
memória;
disco;
rede;
tempo de cluster.

Portanto:

Performance
     ↓
Efficiency
     ↓
Cost

Performance tuning também é FinOps.

28. O Spark moderno consegue ajudar

Nas versões modernas do Spark, mecanismos de otimização conseguem melhorar várias decisões automaticamente.

Um dos mais importantes é:

Adaptive Query Execution — AQE

Durante a execução, Spark pode utilizar estatísticas reais para adaptar partes do plano.

Isso pode ajudar em cenários relacionados a:

shuffle partitions;
estratégias de join;
skew;
coalescing de partições.

Conceitualmente:

Plano inicial
     ↓
Execução começa
     ↓
Estatísticas reais
     ↓
AQE
     ↓
Plano adaptado

Isso é uma evolução enorme em relação aos primeiros anos do Spark.

Mas existe uma lição importante:

Automação não elimina a necessidade de entender o sistema.

Quando sabemos como shuffle funciona, conseguimos interpretar melhor o que o Spark está tentando otimizar.

29. Como eu investigaria um Shuffle caro

Imagine um Job demorando muito.

Meu raciocínio seria algo parecido com:

JOB LENTO
   ↓
Qual Stage?
   ↓
Existe Shuffle?
   ↓
Quanto Shuffle Read/Write?
   ↓
As Tasks têm duração semelhante?
   ↓
Existe Skew?
   ↓
Existe Spill?
   ↓
Qual operação criou o Shuffle?

Depois:

groupBy?
join?
repartition?
distinct?
orderBy?
aggregation?

Então perguntaria:

Posso filtrar antes?

Posso selecionar menos colunas?

Posso agregar antes?

O repartition é necessário?

A chave está skewed?

Existe uma estratégia melhor de join?

Uma tabela poderia ser broadcast?

Isso é muito mais eficiente do que ajustar parâmetros aleatoriamente.

30. Checklist de Shuffle

Quando analiso um pipeline Spark, algumas perguntas são especialmente úteis:

Quanto de Shuffle Write existe?
Quanto de Shuffle Read existe?
Qual operação gerou o shuffle?
O shuffle é realmente necessário?
Estamos movimentando colunas desnecessárias?
Podemos filtrar dados antes?
Existe agregação local possível?
Existe data skew?
Existem Tasks muito maiores que as outras?
Há spill para disco?
repartition() está sendo usado sem necessidade?
O número de shuffle partitions faz sentido?
O join está usando uma estratégia adequada?
Uma das tabelas é pequena o suficiente para outra estratégia?
O AQE está ajudando?
O problema é computação ou movimentação de dados?

Essas perguntas normalmente são mais valiosas do que procurar:

"Qual configuração deixa Spark mais rápido?"

31. A mudança de mentalidade

Quando comecei a aprender Spark, era fácil olhar para:

df.groupBy("customer_id").count()

e enxergar apenas:

agrupar
+
contar

Depois de entender arquitetura distribuída, começo a enxergar:

Dataset
   ↓
Partitions
   ↓
Local Processing
   ↓
Shuffle Write
   ↓
Network
   ↓
Shuffle Read
   ↓
New Partitions
   ↓
Aggregation
   ↓
Result

Essa segunda visão é muito mais próxima do que realmente importa para Engenharia de Dados em escala.

32. O princípio que levo para outros sistemas

Existe uma ideia aqui que vai muito além do Spark:

Em sistemas distribuídos, mover dados costuma ser caro.

Esse princípio aparece em:

bancos distribuídos;
Data Warehouses;
Lakehouses;
sistemas de streaming;
microserviços;
Machine Learning distribuído;
HPC;
arquiteturas cloud.

Quanto maior o volume, mais importante fica pensar:

Onde os dados estão?

Onde a computação está?

Quanto precisa atravessar a rede?

Muitas decisões de arquitetura nascem dessas três perguntas.

33. A principal lição deste capítulo

Se eu pudesse resumir shuffle em uma única sequência, seria:

Dados distribuídos
       ↓
Operação precisa reorganizá-los
       ↓
Shuffle Write
       ↓
Disk / Memory
       ↓
Network
       ↓
Shuffle Read
       ↓
Novas Partitions
       ↓
Próximo Stage

E resumiria a otimização em uma pergunta:

Quanto dado realmente precisa atravessar esse caminho?

Quanto menos dados desnecessários movimentarmos, melhor.

Próximo artigo

Agora chegamos naturalmente a um dos assuntos que mais causam diferença entre um pipeline eficiente e um pipeline caro:

Joins

No próximo artigo vamos entender por que isto:

pedidos.join(
    clientes,
    "customer_id"
)

pode resultar em planos de execução completamente diferentes dependendo dos dados.

Vamos explorar:

joins distribuídos;
shuffle joins;
Broadcast Hash Join;
Sort Merge Join;
estratégias de join;
tabelas grandes vs. pequenas;
estatísticas;
broadcast;
data skew;
joins com chaves problemáticas;
leitura do plano com explain();
Catalyst Optimizer;
AQE;
e como escolher uma estratégia baseada nos dados, não em receitas prontas.

Vamos evoluir deste modelo:

Table A ───┐
           ├──► SHUFFLE ───► JOIN
Table B ───┘

para entender quando podemos fazer algo próximo de:

               Small Table
              /     |     \
             ▼      ▼      ▼

Large ───► Executor Executor Executor
             │        │        │
             ▼        ▼        ▼
            JOIN     JOIN     JOIN

É aí que Spark começa a mostrar de verdade o poder do seu otimizador.

← Parte 3: Partições, paralelismo e performance

Próximo: Parte 5 — Joins no Apache Spark: Shuffle, Broadcast e estratégias de execução

Como aprendi Apache Spark — Parte 3: Partições, paralelismo e performance

renanpyd — Fri, 23 Oct 2026 13:59:09 +0000

No artigo anterior, desmontamos a arquitetura de uma aplicação Apache Spark e entendemos a relação entre:

Application
    ↓
Driver
    ↓
Cluster Manager
    ↓
Executors
    ↓
Tasks

Mas ficou uma pergunta importante.

Se eu tenho vários Executors disponíveis, como o Spark consegue dividir os dados para processá-los em paralelo?

A resposta começa em um dos conceitos mais importantes — e muitas vezes subestimados — do Spark:

Partições

Entender partições mudou bastante a maneira como passei a enxergar processamento distribuído.

Porque um cluster pode ter dezenas de máquinas, centenas de cores e muita memória.

Nada disso garante que esses recursos serão utilizados eficientemente.

No fim das contas, existe uma relação fundamental:

Dados
  ↓
Partitions
  ↓
Tasks
  ↓
Parallelism
  ↓
Performance

E é isso que vamos explorar.

1. O que é uma partição?

Um dataset distribuído não é tratado necessariamente como um único bloco gigantesco.

Spark divide logicamente os dados em partições.

Imagine um dataset:

Dataset
│
├── Partition 0
├── Partition 1
├── Partition 2
└── Partition 3

Essas partições podem ser processadas paralelamente pelos Executors.

Por exemplo:

                    DATASET

        ┌──────────────┼──────────────┐
        │              │              │
        ▼              ▼              ▼

   Partition 0    Partition 1    Partition 2
        │              │              │
        ▼              ▼              ▼

   Executor A     Executor B     Executor C

Isso nos leva a uma relação extremamente importante:

Em uma determinada Stage, cada partição normalmente será processada por uma Task.

Podemos pensar:

Partition 0 ───► Task 0
Partition 1 ───► Task 1
Partition 2 ───► Task 2
Partition 3 ───► Task 3

Portanto:

Partitions
     ↓
   Tasks
     ↓
Parallelism

Essa relação ajuda a explicar muita coisa sobre performance no Spark.

2. Partição não é a mesma coisa que arquivo

Esse é um ponto importante.

Imagine que temos:

/data/
├── vendas_01.parquet
├── vendas_02.parquet
├── vendas_03.parquet
└── vendas_04.parquet

É tentador pensar:

1 arquivo = 1 partição Spark

Mas essa relação não é necessariamente verdadeira.

A forma como Spark cria partições durante a leitura depende de fatores como:

formato dos dados;
tamanho dos arquivos;
fonte de dados;
configuração da aplicação;
capacidade de divisão da fonte;
operações realizadas posteriormente.

Além disso, novas operações podem alterar completamente o particionamento.

Por isso precisamos separar mentalmente:

ARMAZENAMENTO

Arquivos
   ↓
Blocos / objetos / arquivos físicos

de:

EXECUÇÃO SPARK

Dataset
   ↓
Partitions
   ↓
Tasks

Partições pertencem ao modelo de execução distribuída do Spark.

3. Descobrindo quantas partições existem

Com RDDs podemos verificar diretamente:

rdd = sc.parallelize(range(100))

print(rdd.getNumPartitions())

Também podemos escolher uma quantidade inicial:

rdd = sc.parallelize(
    range(100),
    4
)

Agora temos, conceitualmente:

100 elementos
      │
      ▼
4 partitions
      │
      ├── P0
      ├── P1
      ├── P2
      └── P3

Com DataFrames também podemos acessar o RDD subjacente para inspeção:

df.rdd.getNumPartitions()

Isso é útil durante análise e troubleshooting.

Mas saber quantas partições existem é apenas o começo.

A pergunta realmente importante é:

Essas partições estão adequadas para o workload?

4. Partições determinam oportunidades de paralelismo

Imagine um cluster com:

4 Executors
4 cores por Executor

Temos potencialmente:

4 × 4 = 16 cores

Simplificando, podemos pensar em até 16 Tasks executando simultaneamente, dependendo das configurações e dos recursos disponíveis.

Agora imagine que nosso Stage possui apenas:

4 partitions

Teremos aproximadamente:

Task 1
Task 2
Task 3
Task 4

Mesmo tendo capacidade para mais execução concorrente.

Visualmente:

Capacidade disponível

Core 01 → Task
Core 02 → Task
Core 03 → Task
Core 04 → Task

Core 05 → ocioso
Core 06 → ocioso
Core 07 → ocioso
Core 08 → ocioso
...
Core 16 → ocioso

Temos um cluster relativamente grande, mas pouco paralelismo.

Agora considere:

64 partitions

Podemos ter ondas sucessivas de Tasks:

16 Tasks
   ↓
16 Tasks
   ↓
16 Tasks
   ↓
16 Tasks

Os recursos podem ser utilizados de maneira muito mais eficiente.

Mas isso não significa que:

Quanto mais partições, melhor.

Porque existe o outro extremo.

5. Partições demais também são um problema

Imagine um dataset relativamente pequeno dividido em:

100.000 partitions

Agora Spark precisa administrar:

100.000 Tasks

Mesmo que cada Task processe pouquíssimos dados.

Cada Task possui algum custo operacional.

Existe overhead relacionado a:

scheduling;
serialização;
comunicação;
inicialização;
acompanhamento;
coordenação;
metadados.

Podemos chegar a uma situação como:

Task
├── 20 ms processando dados
└── overhead operacional significativo

Multiplique isso por dezenas ou centenas de milhares de Tasks.

O problema deixa de ser processamento.

Passa a ser coordenação.

Esse é um padrão comum em sistemas distribuídos:

Paralelismo tem custo.

6. Partições grandes demais

Agora vamos para o outro extremo.

Imagine:

Dataset: 1 TB
Partitions: 4

Simplificando:

Partition 1 ≈ 250 GB
Partition 2 ≈ 250 GB
Partition 3 ≈ 250 GB
Partition 4 ≈ 250 GB

Isso pode gerar vários problemas.

Cada Task terá uma quantidade enorme de dados para processar.

Podemos ter:

baixa utilização do cluster;
pressão de memória;
spills para disco;
garbage collection;
Tasks muito longas;
maior custo de recuperação em caso de falha.

Visualmente:

Cluster

Executor 1 → ████████████████████████
Executor 2 → ████████████████████████
Executor 3 → ████████████████████████
Executor 4 → ████████████████████████

Executor 5 → ocioso
Executor 6 → ocioso
Executor 7 → ocioso
Executor 8 → ocioso

O cluster possui recursos.

Mas não existem Tasks suficientes para utilizá-los.

7. Partições pequenas demais

Agora imagine o inverso:

Dataset: 1 TB
Partitions: 1.000.000

Teríamos uma quantidade gigantesca de Tasks.

Conceitualmente:

1.000.000 partitions
          ↓
1.000.000 tasks
          ↓
scheduling
          ↓
coordenação
          ↓
overhead

Mesmo que o cluster consiga executar muitas Tasks paralelamente, gerenciar essa quantidade pode ser extremamente caro.

Portanto, procuramos equilíbrio.

Poucas partitions
       ↓
baixo paralelismo
Tasks muito grandes


Partitions demais
       ↓
overhead excessivo
Tasks muito pequenas


Quantidade adequada
       ↓
bom paralelismo
Tasks administráveis

Não existe um número universal perfeito.

O tamanho adequado depende do workload.

8. O problema mais perigoso: partições desbalanceadas

Até agora assumimos algo parecido com:

P1 = 1 GB
P2 = 1 GB
P3 = 1 GB
P4 = 1 GB

Mas dados reais raramente são tão bonitos.

Podemos encontrar:

P1 = 500 MB
P2 = 600 MB
P3 = 550 MB
P4 = 18 GB

Agora imagine quatro Tasks:

Task 1 → 20 segundos
Task 2 → 23 segundos
Task 3 → 21 segundos
Task 4 → 8 minutos

O Stage não termina quando as três primeiras Tasks acabam.

Ele precisa esperar a última.

Visualmente:

Task 1 ████ ✓
Task 2 █████ ✓
Task 3 ████ ✓
Task 4 ███████████████████████████████████████...

Isso é um sinal clássico de desequilíbrio.

E uma das causas pode ser:

Data Skew

9. O que é Data Skew?

Imagine dados de vendas agrupados por estado.

Suponha:

SP → 60 milhões
RJ → 15 milhões
MG → 10 milhões
BA → 5 milhões
AC → 200 mil
RR → 100 mil

Se uma operação distribuir dados utilizando estado como chave, algumas partições podem ficar muito maiores que outras.

Podemos terminar com:

Partition SP █████████████████████████████████
Partition RJ █████████
Partition MG ██████
Partition BA ███
Partition AC ▏
Partition RR ▏

Os Executors responsáveis pelas partições pequenas terminam rapidamente.

Enquanto isso, uma Task continua trabalhando.

O cluster pode ficar assim:

Executor 1 → idle
Executor 2 → idle
Executor 3 → idle
Executor 4 → idle
Executor 5 → █████████████ trabalhando

Esse fenômeno é importante porque mostra que:

Ter muitas partições não garante paralelismo eficiente.

Precisamos também de boa distribuição dos dados entre elas.

10. `repartition()`

Spark oferece mecanismos para alterar o particionamento.

Um deles é:

df = df.repartition(100)

Conceitualmente:

ANTES

P1
P2
P3
P4

       repartition(100)

              ↓

DEPOIS

P1
P2
P3
...
P100

Mas existe um detalhe fundamental.

Redistribuir dados entre partições pode exigir movimentação pela rede.

Ou seja:

repartition()
      ↓
   Shuffle

Imagine:

Executor A                     Executor B

[A, A, B, C]                  [A, B, C, C]

             repartition

                   ↓

Executor A                     Executor B

[A, A, A]                     [B, B]
                               [C, C, C]

Dados precisaram viajar entre executors.

E rede não é gratuita.

Por isso:

repartition() não deve ser usado como um botão mágico de performance.

Ele resolve alguns problemas criando outro custo: shuffle.

11. Repartition por coluna

Também podemos particionar logicamente usando uma ou mais colunas:

df = df.repartition(
    "customer_id"
)

Ou especificar quantidade:

df = df.repartition(
    200,
    "customer_id"
)

Isso pode ser útil quando operações posteriores dependem dessa chave.

Mas novamente precisamos pensar na distribuição.

Se customer_id estiver bem distribuído:

customer_id
     ↓
hash
     ↓
P1 P2 P3 P4 P5 ...

podemos conseguir uma distribuição razoável.

Se a chave tiver forte skew, o resultado pode continuar ruim.

Particionar não corrige automaticamente uma distribuição ruim.

12. `coalesce()`

Outro método importante é:

df = df.coalesce(10)

Normalmente utilizamos coalesce() quando queremos reduzir a quantidade de partições.

Por exemplo:

ANTES

P1 P2 P3 P4 P5 P6 P7 P8

        coalesce(4)

             ↓

DEPOIS

P1 P2 P3 P4

A grande diferença conceitual é que coalesce() pode reduzir partições evitando uma redistribuição completa dos dados.

Compare:

repartition()
     ↓
redistribuição ampla
     ↓
shuffle

com:

coalesce()
     ↓
combinação de partitions
     ↓
pode evitar full shuffle

Isso pode tornar coalesce() mais barato quando o objetivo é simplesmente reduzir paralelismo.

Mas também existe uma consequência.

13. O perigo de usar `coalesce(1)`

Esse comando aparece bastante:

df.coalesce(1)

Por quê?

Normalmente porque alguém quer:

"Salvar tudo em um único arquivo."

Funciona em determinados cenários.

Mas imagine:

100 partitions
       ↓
coalesce(1)
       ↓
1 partition
       ↓
1 Task

Acabamos de destruir praticamente todo o paralelismo daquela etapa.

Imagine processar:

200 GB

em uma única partição.

Podemos transformar um processamento distribuído em um enorme gargalo.

Por isso, quando encontro:

df.coalesce(1)

em um pipeline de produção, minha primeira pergunta é:

Por que precisamos de exatamente um arquivo?

Às vezes existe uma justificativa.

Muitas vezes existe apenas uma dependência legada ou uma decisão que nunca foi questionada.

14. `repartition()` vs. `coalesce()`

Uma forma simples de lembrar:

                repartition()        coalesce()

Aumentar               ✓                 não é o uso típico

Reduzir                ✓                 ✓

Shuffle          normalmente sim     pode evitar full shuffle

Distribuição        melhor             pode ficar desigual

Custo              maior              geralmente menor

Isso não significa:

coalesce = bom
repartition = ruim

Significa:

Cada operação resolve um problema diferente.

Se preciso redistribuir os dados de forma mais uniforme, repartition() pode ser exatamente o que quero.

Se tenho muitas partições pequenas e apenas quero consolidá-las, coalesce() pode ser mais apropriado.

15. Partições e Shuffle

Agora começamos a conectar os capítulos.

Considere:

resultado = (
    df
    .groupBy("customer_id")
    .count()
)

Para contar todos os registros do mesmo cliente, Spark precisa garantir que dados da mesma chave possam ser processados conjuntamente.

Imagine:

Executor A

customer 1
customer 2
customer 3


Executor B

customer 1
customer 4
customer 2

Precisamos reorganizar:

ANTES

Executor A       Executor B

1                1
2                4
3                2


        ↓ SHUFFLE ↓


DEPOIS

Partition A
1
1

Partition B
2
2

Partition C
3

Partition D
4

É nesse momento que rede, disco, memória e particionamento começam a interagir.

Por isso shuffle é uma das operações mais caras e importantes de entender no Spark.

16. Partições e arquivos pequenos

Particionamento também aparece em outro problema clássico de Data Engineering:

Small Files Problem

Imagine um pipeline que grava:

/year=2026/month=10/day=23/

part-000001.parquet
part-000002.parquet
part-000003.parquet
...
part-200000.parquet

Temos centenas de milhares de arquivos minúsculos.

Isso pode prejudicar:

leitura;
listagem de arquivos;
planejamento;
metadados;
armazenamento;
engines de consulta;
Data Lakes.

Muitas vezes a quantidade de arquivos gerados está relacionada à quantidade de partições no momento da escrita.

Por exemplo:

df.write.parquet(
    "/data/output"
)

As partições do DataFrame influenciam a quantidade de tarefas de escrita e, consequentemente, podem influenciar a quantidade de arquivos produzidos.

Isso conecta performance computacional com arquitetura de armazenamento.

17. Particionamento de execução ≠ particionamento de armazenamento

Aqui existe uma distinção importante.

Podemos ter:

df.repartition(100)

Isso está relacionado ao particionamento do DataFrame durante a execução.

Mas também podemos escrever:

df.write \
  .partitionBy("ano", "mes") \
  .parquet("/data/vendas")

Agora estamos falando de organização física/lógica dos dados no armazenamento:

/data/vendas/

ano=2025/
    mes=12/

ano=2026/
    mes=01/

ano=2026/
    mes=02/

São conceitos relacionados, mas diferentes.

Uma forma útil de separar:

repartition()
     ↓
EXECUÇÃO


partitionBy()
     ↓
ARMAZENAMENTO

Confundir os dois pode levar a decisões ruins de arquitetura.

18. E quanto ao tamanho ideal de uma partição?

Essa é uma pergunta clássica:

Quantos MB uma partição Spark deveria ter?

A resposta correta é:

depende.

Você encontrará recomendações práticas em diferentes contextos, frequentemente trabalhando com partições na ordem de dezenas ou centenas de megabytes.

Mas transformar um número em regra universal é perigoso.

Precisamos considerar:

tipo de operação;
volume de dados;
formato;
compressão;
memória dos Executors;
quantidade de cores;
número de Executors;
custo de serialização;
skew;
shuffle;
origem dos dados;
destino dos dados.

Em vez de decorar:

Partition = X MB

prefiro pensar:

A Task termina rápido demais?
        ↓
Talvez partitions pequenas demais


A Task demora excessivamente?
        ↓
Talvez partition grande


Algumas Tasks demoram muito mais?
        ↓
Talvez skew


Muitos cores ociosos?
        ↓
Talvez poucas partitions


Scheduler sobrecarregado?
        ↓
Talvez partitions demais

O Spark UI ajuda muito nessa análise.

19. Um exemplo prático

Imagine:

Dataset = 800 GB

Cluster:
10 Executors
4 cores por Executor

Temos aproximadamente:

40 cores

Agora suponha:

8 partitions

Temos no máximo oito Tasks para aquela etapa.

Então:

40 cores disponíveis
 8 Tasks

Grande parte do cluster ficará ociosa.

Agora imagine:

800 partitions

Teríamos aproximadamente:

1 GB por partition

Ainda pode ser grande dependendo da operação.

Se utilizarmos:

3200 partitions

teríamos aproximadamente:

250 MB por partition

Isso poderia ser um ponto inicial interessante para investigação.

Mas atenção:

não estou dizendo que 3200 é a configuração correta.

Estou mostrando como raciocinar.

A validação real vem das métricas:

Spark UI
   │
   ├── Task Duration
   ├── Input Size
   ├── Shuffle Read
   ├── Shuffle Write
   ├── Spill
   ├── GC Time
   └── Executor Utilization

Engenharia de performance não deveria ser:

"Vi alguém usando 3200 partitions."

Deveria ser:

"Escolhi um ponto inicial, medi o comportamento e ajustei baseado em evidências."

20. O número de cores não conta toda a história

Suponha:

Executors: 10
Cores por Executor: 4

Total: 40 cores

Podemos pensar inicialmente:

40 cores
   ↓
até aproximadamente 40 Tasks simultâneas

Mas isso não significa que:

40 partitions

é necessariamente ideal.

Por quê?

Porque se temos exatamente 40 Tasks e uma delas é muito lenta:

39 Tasks → 30 segundos
 1 Task  → 8 minutos

depois dos primeiros 30 segundos grande parte do cluster pode ficar ociosa esperando aquela última Task.

Ter múltiplas ondas de Tasks pode ajudar na utilização dos recursos.

Novamente:

não existe uma fórmula universal.

Existe análise de workload.

21. Quando particionamento vira arquitetura

Esse assunto parece inicialmente uma otimização de baixo nível.

Mas não é.

Particionamento influencia:

                    PARTITIONING
                         │
          ┌──────────────┼──────────────┐
          │              │              │
          ▼              ▼              ▼
      Parallelism      Memory         Network
          │              │              │
          ▼              ▼              ▼
        Tasks          Spill          Shuffle
          │              │              │
          └──────────────┼──────────────┘
                         ▼
                    PERFORMANCE
                         │
                         ▼
                       COST

Em cloud isso fica ainda mais importante.

Um pipeline ineficiente não significa apenas:

"demorou mais."

Pode significar:

"custou mais."

Se um cluster fica ativo durante duas horas em vez de vinte minutos, existe impacto financeiro.

Por isso performance também é uma preocupação de arquitetura e FinOps.

22. Um exemplo de raciocínio de troubleshooting

Imagine que um Job demora 45 minutos.

No Spark UI encontramos:

Stage 8

Tasks: 500
Median Duration: 20 s
Max Duration: 31 min

Isso é um enorme sinal.

Não começaria aumentando memória aleatoriamente.

Perguntaria:

Por que uma Task levou 31 minutos
quando a mediana foi 20 segundos?

Possíveis hipóteses:

Task lenta
   │
   ├── partition muito grande?
   ├── data skew?
   ├── shuffle excessivo?
   ├── spill para disco?
   ├── garbage collection?
   ├── problema no executor?
   └── problema de I/O?

Esse é o tipo de raciocínio que quero desenvolver ao longo desta série.

Não decorar configurações.

Mas saber onde procurar.

23. `repartition()` também pode piorar tudo

Imagine um pipeline:

df = spark.read.parquet("/data/events")

df = df.repartition(5000)

resultado = (
    df
    .filter("event_type = 'purchase'")
    .groupBy("customer_id")
    .count()
)

Talvez alguém tenha adicionado:

repartition(5000)

porque:

"Mais partições deixam Spark mais rápido."

Mas acabamos de introduzir potencialmente uma redistribuição cara antes mesmo da operação principal.

Podemos ter:

Read
 ↓
Repartition
 ↓
SHUFFLE
 ↓
Filter
 ↓
GroupBy
 ↓
SHUFFLE
 ↓
Aggregation

Talvez tenhamos criado dois movimentos grandes de dados onde não precisávamos.

Esse é um exemplo perfeito de uma regra importante:

Otimização sem compreender o plano de execução pode virar desotimização.

24. O melhor `repartition()` às vezes é nenhum

Antes de escrever:

df.repartition(...)

eu perguntaria:

Qual problema estou tentando resolver?

Se a resposta for:

"Não sei, mas acho que melhora performance."

Não é uma boa justificativa.

Perguntas melhores:

Tenho poucas partitions?

Tenho muitas?

Existe skew?

Preciso distribuir por determinada chave?

Estou preparando os dados para um join?

Estou gerando small files?

Existe um shuffle inevitável logo depois?

Qual é o plano físico?

Só depois escolheria a estratégia.

25. O que mudou no Spark moderno

Existe uma diferença importante entre estudar Spark historicamente e trabalhar com versões modernas.

Hoje o Spark possui mecanismos muito mais sofisticados de otimização.

Um deles é o:

Adaptive Query Execution — AQE

Com AQE, Spark pode utilizar informações coletadas durante a própria execução para ajustar partes do plano.

Entre outras capacidades, isso pode ajudar com:

coalescing de shuffle partitions;
otimização de joins;
tratamento de determinados cenários de skew.

Isso representa uma evolução enorme.

Historicamente, muitas decisões precisavam ser tomadas antecipadamente com informações limitadas.

Hoje Spark consegue adaptar determinadas decisões durante a execução.

Mas isso não elimina a necessidade de entender partições.

Na verdade, torna o conhecimento ainda mais útil.

Porque agora podemos entender o que o engine está tentando otimizar por nós.

26. Como penso em partições hoje

Meu modelo mental é aproximadamente este:

                  DATASET
                     │
                     ▼
                PARTITIONS
                     │
          ┌──────────┼──────────┐
          │          │          │
          ▼          ▼          ▼
       Size       Balance      Count
          │          │          │
          └──────────┼──────────┘
                     ▼
                   TASKS
                     │
                     ▼
                PARALLELISM
                     │
          ┌──────────┼──────────┐
          ▼          ▼          ▼
         CPU       Memory     Network
          │          │          │
          └──────────┼──────────┘
                     ▼
                 PERFORMANCE
                     │
                     ▼
                    COST

Não penso:

"Qual número de partitions devo usar?"

Penso:

"Como os dados estão sendo distribuídos e como isso afeta a execução?"

Essa é uma pergunta muito melhor.

27. Checklist rápido de particionamento

Quando encontro problemas de performance, algumas perguntas ajudam bastante:

Quantas partições existem?
Qual o tamanho médio?
Qual o tamanho máximo?
As Tasks possuem duração semelhante?
Existe uma ou poucas Tasks muito mais lentas?
Todos os cores estão sendo utilizados?
Existe shuffle?
Quanto está sendo lido e escrito no shuffle?
Existe spill para disco?
Há indícios de data skew?
Algum repartition() foi adicionado sem necessidade?
Existe coalesce(1)?
Estamos gerando arquivos pequenos demais?
O problema está na execução ou no layout do armazenamento?

Essas perguntas transformam:

"Meu Spark está lento."

em uma investigação muito mais objetiva.

28. A principal lição deste capítulo

Se eu pudesse resumir tudo em uma frase:

O cluster fornece capacidade. As partições determinam quanto dessa capacidade conseguimos explorar.

Um cluster enorme com particionamento ruim continua sendo um sistema mal utilizado.

Podemos ter:

100 CPUs disponíveis

e apenas:

4 Tasks

Ou:

100 CPUs disponíveis

1.000.000 Tasks minúsculas

Nos dois casos existe desperdício.

O objetivo é encontrar uma distribuição que permita ao Spark executar trabalho suficiente em paralelo sem transformar coordenação, memória ou rede no novo gargalo.

Próximo artigo

Neste capítulo apareceu repetidamente uma palavra:

Shuffle

Ela surgiu em:

repartition()

em:

groupBy()

em:

reduceByKey()

e também aparecerá quando falarmos sobre:

join()

Shuffle é uma das operações mais importantes para entender performance em Apache Spark.

No próximo artigo vamos descobrir:

o que realmente acontece durante um shuffle;
por que dados precisam atravessar a rede;
narrow vs. wide dependencies;
shuffle read;
shuffle write;
agregações;
joins;
impacto sobre memória e disco;
spill;
e por que reduzir movimentação de dados costuma ser uma das melhores otimizações possíveis.

Vamos seguir o fluxo:

Partitions
     ↓
Wide Transformation
     ↓
Shuffle Write
     ↓
Network
     ↓
Shuffle Read
     ↓
New Partitions
     ↓
Next Stage

É aqui que começamos a entrar de verdade em performance tuning de Spark.

[← Parte 2: Por dentro da arquitetura — Driver, Executors, Jobs, Stages e Tasks]

Próximo: Parte 4 — Shuffle no Apache Spark: onde a performance começa a ficar cara

Próximo: Parte 2 — Por dentro da arquitetura do Apache Spark: Driver, Executors, Jobs, Stages e Tasks

renanpyd — Fri, 23 Oct 2026 13:54:36 +0000

No artigo anterior, vimos por que o Apache Spark chamou tanta atenção no cenário de processamento distribuído e passamos pelos conceitos de RDDs, lineage, transformations, actions e lazy evaluation.

Mas uma pergunta ficou aberta:

Quando executamos uma aplicação Spark, quem realmente faz o trabalho?

Quando comecei a estudar Spark, entender essa arquitetura foi um divisor de águas.

Até então, código como este parecia simples:

resultado = (
    sc.textFile("dados.txt")
      .flatMap(lambda linha: linha.split())
      .map(lambda palavra: (palavra, 1))
      .reduceByKey(lambda a, b: a + b)
      .collect()
)

Mas existe uma diferença enorme entre entender o que esse código faz e entender como ele é executado em um cluster.

Por trás dessas poucas linhas existe uma arquitetura envolvendo:

Application
    │
    ▼
Driver
    │
    ▼
Cluster Manager
    │
    ├─────────────┬─────────────┐
    ▼             ▼             ▼
Executor 1    Executor 2    Executor 3
    │             │             │
   Tasks         Tasks         Tasks

E é essa arquitetura que vamos desmontar neste artigo.

1. Começando pelo todo: o que é uma aplicação Spark?

Uma aplicação Spark é composta, de maneira simplificada, por dois grandes elementos:

Driver
Executors

Entre eles, normalmente temos um Cluster Manager, responsável pela alocação dos recursos do cluster.

Podemos pensar inicialmente assim:

                 SPARK APPLICATION

                      Driver
                        │
                        ▼
                 Cluster Manager
                        │
             ┌──────────┼──────────┐
             │          │          │
             ▼          ▼          ▼
         Executor    Executor    Executor
             │          │          │
          Tasks       Tasks       Tasks

Essa imagem mental é importante.

O Driver coordena.

Os Executors executam.

Mas vamos detalhar cada parte.

2. Driver: o cérebro da aplicação

O Driver é o processo responsável por coordenar a aplicação Spark.

É nele que o código principal da aplicação é executado.

Considere:

from pyspark.sql import SparkSession

spark = (
    SparkSession.builder
    .appName("MinhaAplicacao")
    .getOrCreate()
)

df = spark.read.parquet("/dados/vendas")

resultado = (
    df
    .filter(df.valor > 1000)
    .groupBy("estado")
    .count()
)

resultado.show()

Esse programa não simplesmente "vai para o cluster".

Existe um processo central responsável por interpretar a aplicação, construir planos de execução e coordenar o trabalho distribuído.

Esse processo é o Driver.

De maneira simplificada:

                    DRIVER

              Código da aplicação
                      │
                      ▼
             Planejamento do trabalho
                      │
                      ▼
             Divisão da execução
                      │
                      ▼
              Envio de Tasks
                      │
                      ▼
            Acompanhamento do Job

O Driver precisa saber:

quais transformações foram definidas;
quais ações precisam ser executadas;
quais recursos estão disponíveis;
quais executors estão ativos;
quais tasks precisam ser executadas;
quais tasks terminaram;
quais falharam.

Por isso, uma regra importante começa a aparecer:

O Driver não é apenas "mais uma máquina" do cluster. Ele é o coordenador da aplicação.

3. SparkContext: a porta de entrada histórica

Nos primeiros anos do Spark, um dos objetos mais importantes era o:

SparkContext

Em PySpark, era comum encontrarmos:

from pyspark import SparkContext

sc = SparkContext(
    "local[*]",
    "MinhaAplicacao"
)

E então:

rdd = sc.parallelize([1, 2, 3, 4, 5])

ou:

rdd = sc.textFile("dados.txt")

O SparkContext representava a conexão da aplicação com o ambiente de execução Spark.

Podemos visualizar:

Aplicação
    │
    ▼
SparkContext
    │
    ▼
Cluster

Hoje, normalmente começamos por uma abstração de nível mais alto:

from pyspark.sql import SparkSession

spark = (
    SparkSession.builder
    .appName("MinhaAplicacao")
    .getOrCreate()
)

E podemos acessar o contexto subjacente:

sc = spark.sparkContext

Essa mudança é interessante porque mostra uma evolução importante da própria plataforma.

Historicamente, aprender Spark significava começar por:

SparkContext
      ↓
     RDD

Hoje é muito mais comum começar por:

SparkSession
      ↓
 DataFrame / SQL

Os fundamentos distribuídos, entretanto, continuam existindo.

4. Cluster Manager: quem distribui os recursos?

O Driver precisa de máquinas para executar o trabalho.

É aí que entra o Cluster Manager.

Sua função é gerenciar recursos disponíveis no cluster.

Historicamente, Spark podia trabalhar com diferentes gerenciadores, como:

Spark Standalone;
Apache Mesos;
Hadoop YARN.

Hoje também encontramos Spark executado em diferentes ambientes e plataformas gerenciadas.

Conceitualmente, porém, podemos pensar:

Driver
   │
   │ "Preciso de recursos"
   ▼
Cluster Manager
   │
   ├─────────────┬─────────────┐
   ▼             ▼             ▼
Máquina A     Máquina B     Máquina C

O Cluster Manager disponibiliza recursos para que os Executors possam ser iniciados.

Uma distinção importante:

Cluster Manager e Executor não são a mesma coisa.

O Cluster Manager administra recursos.

Os Executors realizam o processamento da aplicação.

5. Executors: onde o trabalho acontece

Os Executors são processos responsáveis por executar as tarefas da aplicação Spark.

Imagine um cluster com três executors:

                      DRIVER
                         │
          ┌──────────────┼──────────────┐
          ▼              ▼              ▼
      EXECUTOR 1      EXECUTOR 2      EXECUTOR 3
          │              │              │
      Task 1          Task 2          Task 3
      Task 4          Task 5          Task 6

Cada Executor possui recursos como:

Executor
├── CPU / cores
├── memória
├── tasks
└── dados armazenados/cacheados

Quando configuramos recursos para uma aplicação Spark, começamos a encontrar parâmetros relacionados justamente a essa arquitetura.

Por exemplo:

spark.executor.memory
spark.executor.cores

Isso deixa muito mais claro o que significa dizer:

"Meu executor está ficando sem memória."

Estamos falando de um processo específico responsável por executar partes da computação distribuída.

6. Driver vs. Executor

Essa diferença é fundamental.

Podemos resumir assim:

DRIVER
│
├── executa o programa principal
├── coordena a aplicação
├── planeja o trabalho
├── agenda tarefas
└── acompanha a execução


EXECUTOR
│
├── executa tasks
├── processa partições
├── mantém dados quando necessário
└── devolve resultados/status ao Driver

Essa separação ajuda a entender vários erros.

Por exemplo:

df.collect()

O que collect() faz?

Ele coleta os resultados distribuídos e os envia para o Driver.

Imagine um DataFrame com:

500 GB

distribuídos entre vários executors.

Se tentarmos:

dados = df.collect()

conceitualmente estamos pedindo:

Executor 1 ───┐
Executor 2 ───┤
Executor 3 ───┼───► DRIVER
Executor 4 ───┤
Executor 5 ───┘

Agora imagine tentar colocar centenas de gigabytes no processo do Driver.

Temos um problema.

É por isso que um comando aparentemente inocente como:

collect()

pode ser extremamente perigoso em produção.

7. Partições: a unidade que conecta dados e paralelismo

No artigo anterior vimos que um RDD pode ser dividido em partições.

Agora isso começa a fazer mais sentido.

Considere:

Dataset
   │
   ├── Partition 1
   ├── Partition 2
   ├── Partition 3
   └── Partition 4

Essas partições permitem distribuir o processamento.

Por exemplo:

Partition 1 ───► Executor A
Partition 2 ───► Executor A
Partition 3 ───► Executor B
Partition 4 ───► Executor B

Em uma determinada etapa da execução, uma task trabalha tipicamente sobre uma partição.

Essa relação é extremamente importante:

Stage
  │
  ├── Partition 1 ──► Task 1
  ├── Partition 2 ──► Task 2
  ├── Partition 3 ──► Task 3
  └── Partition 4 ──► Task 4

E daqui surge uma das ideias mais importantes para performance:

O número e o tamanho das partições influenciam diretamente o paralelismo da aplicação.

Vamos aprofundar isso em um artigo específico.

8. Action → Job

No artigo anterior vimos a diferença entre Transformations e Actions.

Considere:

rdd = sc.textFile("dados.txt")

filtrado = rdd.filter(
    lambda linha: "ERROR" in linha
)

resultado = filtrado.count()

Temos:

textFile()
    │
    ▼
filter()
    │
    ▼
count()

filter() é uma transformation.

count() é uma action.

Quando uma Action precisa ser executada, Spark dispara computação para produzir aquele resultado.

Isso normalmente resulta na criação de um:

Job

Uma forma útil de construir o modelo mental é:

ACTION
   │
   ▼
 JOB

Mas o Job ainda pode ser grande demais para ser executado como uma única unidade.

Então Spark o divide.

9. Job → Stages

Um Job é dividido em Stages.

Mas Spark não divide os stages aleatoriamente.

As fronteiras aparecem principalmente quando determinadas operações exigem redistribuição de dados entre partições.

Ou seja:

shuffle.

Considere:

rdd = sc.textFile("dados.txt")

resultado = (
    rdd
    .flatMap(lambda linha: linha.split())
    .map(lambda palavra: (palavra, 1))
    .reduceByKey(lambda a, b: a + b)
)

resultado.collect()

Uma representação simplificada seria:

STAGE 1

textFile
   │
flatMap
   │
 map
   │
   ▼

======= SHUFFLE =======

   │
   ▼

STAGE 2

reduceByKey
   │
collect

Por quê?

Porque reduceByKey() precisa reunir valores que possuem a mesma chave.

Imagine:

Executor A:
spark -> 1
data  -> 1

Executor B:
spark -> 1
python -> 1

Executor C:
data  -> 1
spark -> 1

Para somar todos os valores de spark, os registros precisam ser reorganizados.

Algo como:

spark -> 1 ──┐
spark -> 1 ──┼──► spark -> 3
spark -> 1 ──┘

Essa movimentação distribuída é um shuffle.

E shuffles são tão importantes que terão um artigo próprio nesta série.

10. Stage → Tasks

Agora chegamos à menor unidade prática de execução que precisamos entender neste momento:

Task.

Um Stage é composto por várias Tasks.

Exemplo:

Stage 1
│
├── Task 1 → Partition 1
├── Task 2 → Partition 2
├── Task 3 → Partition 3
└── Task 4 → Partition 4

Essas tasks podem ser distribuídas entre os Executors:

                    STAGE

        ┌─────────────┼─────────────┐
        │             │             │
        ▼             ▼             ▼

    Executor A    Executor B    Executor C

      Task 1        Task 2        Task 3
      Task 4        Task 5        Task 6

Agora começamos a enxergar o caminho completo.

11. O caminho completo de uma aplicação Spark

Vamos juntar tudo.

Temos nosso programa:

resultado = (
    sc.textFile("dados.txt")
      .flatMap(lambda linha: linha.split())
      .map(lambda palavra: (palavra, 1))
      .reduceByKey(lambda a, b: a + b)
      .collect()
)

Conceitualmente:

                    APPLICATION
                         │
                         ▼
                       DRIVER
                         │
                         ▼
                       ACTION
                         │
                         ▼
                        JOB
                         │
                 ┌───────┴───────┐
                 ▼               ▼
              STAGE 1          STAGE 2
                 │               │
            ┌────┼────┐      ┌───┼───┐
            ▼    ▼    ▼      ▼   ▼   ▼
           Task Task Task    Task Task Task
            │    │    │      │   │   │
            └────┼────┘      └───┼───┘
                 │               │
                 ▼               ▼
              EXECUTORS       EXECUTORS

E agora temos uma hierarquia muito útil:

Application
    ↓
Action
    ↓
Job
    ↓
Stages
    ↓
Tasks
    ↓
Partitions
    ↓
Executors

Não trate esse desenho como uma equivalência matemática perfeita entre todos os elementos.

Ele é um modelo mental para entender a execução.

12. Um exemplo mais próximo do Spark moderno

Vamos considerar:

df = spark.read.parquet("/data/vendas")

resultado = (
    df
    .filter("valor > 100")
    .groupBy("estado")
    .sum("valor")
)

resultado.show()

O código parece simples.

Mas internamente precisamos pensar em algo próximo de:

Leitura
  │
  ▼
Filter
  │
  ▼
GroupBy
  │
  ▼
Shuffle
  │
  ▼
Aggregation
  │
  ▼
Result

Distribuído entre:

Driver
  │
  ├──────────┬──────────┬──────────┐
  ▼          ▼          ▼          ▼
Executor   Executor   Executor   Executor

É por isso que duas linhas aparentemente semelhantes podem ter comportamentos de performance completamente diferentes.

Compare:

df.filter("valor > 100")

com:

df.groupBy("cliente_id").count()

A primeira pode ser executada localmente sobre cada partição.

A segunda pode exigir redistribuição dos registros pela chave cliente_id.

Em outras palavras:

o código parece simples, mas a topologia de execução pode ser completamente diferente.

13. Narrow vs. Wide Transformations

Aqui aparece outra distinção extremamente importante.

Narrow Transformation

Uma transformação é considerada narrow quando cada partição de saída depende de um conjunto limitado de partições anteriores, frequentemente uma única partição.

Por exemplo:

rdd.map(...)

ou:

rdd.filter(...)

Podemos imaginar:

P1 ───► P1'
P2 ───► P2'
P3 ───► P3'

Não precisamos necessariamente redistribuir todos os dados pela rede.

Wide Transformation

Uma wide transformation pode exigir dados provenientes de várias partições anteriores.

Exemplo clássico:

rdd.reduceByKey(...)

Podemos imaginar:

P1 ──┐     ┌──► P1'
     ├─────┤
P2 ──┤     ├──► P2'
     ├─────┤
P3 ──┘     └──► P3'

Agora existe movimentação entre partições.

Isso geralmente significa:

Wide Transformation
        ↓
     Shuffle
        ↓
Nova fronteira de Stage

Essa relação é extremamente importante para otimização.

14. Agora o Spark UI começa a fazer sentido

Quando abrimos o Spark UI, encontramos termos como:

Jobs
Stages
Tasks
Executors
Storage
SQL

Sem entender a arquitetura, essas telas parecem apenas um painel cheio de métricas.

Depois de entender o modelo de execução, elas começam a contar uma história.

Imagine encontrar:

Stage 12

Tasks: 200
Completed: 199
Running: 1

Uma pergunta interessante aparece:

Por que uma única task está demorando muito mais do que as outras 199?

Talvez exista:

uma partição muito maior;
data skew;
garbage collection;
spill para disco;
gargalo de I/O;
problema de rede.

Agora imagine:

Executor 3
Memory: 95%
GC Time: alto

Outra hipótese aparece:

Talvez esse executor esteja sofrendo pressão de memória.

É aqui que Spark deixa de ser apenas uma API.

Começamos a pensar como engenheiros de sistemas distribuídos.

15. Driver também pode virar gargalo

Muita gente olha apenas para os Executors.

Mas o Driver também possui recursos limitados.

Um exemplo clássico:

dados = df.collect()

Outro:

dados = df.toPandas()

Dependendo do tamanho do dataset, estamos tentando trazer uma quantidade enorme de dados distribuídos para um único processo.

Visualmente:

              100 GB distribuídos

Executor 1 ─────┐
Executor 2 ─────┤
Executor 3 ─────┼────► DRIVER
Executor 4 ─────┤
Executor 5 ─────┘

O cluster inteiro pode ter muita memória.

Isso não significa que o Driver tenha memória suficiente para receber tudo.

Essa distinção parece simples depois que aprendemos.

Antes disso, é uma fonte comum de problemas.

16. Mais executors não significa automaticamente mais performance

Outro aprendizado importante:

Distribuição não elimina custo.

Adicionar máquinas pode aumentar paralelismo.

Mas também podemos aumentar:

comunicação;
serialização;
coordenação;
shuffle;
overhead de scheduling.

Imagine um dataset pequeno distribuído entre dezenas de executors.

Talvez o custo de coordenar todo esse ambiente seja maior do que o benefício.

Por outro lado, poucas partições em um cluster enorme também desperdiçam recursos.

Por exemplo:

Cluster:

Executor 1  ████████
Executor 2  ████████
Executor 3  ████████
Executor 4  ████████
Executor 5  ████████
Executor 6  ████████
Executor 7  ████████
Executor 8  ████████

Dataset:

Partition 1
Partition 2

Não importa termos oito executors.

Se determinada etapa possui apenas duas tasks disponíveis, boa parte dos recursos ficará ociosa.

Daí surge uma relação fundamental:

PARTITIONS
    ↓
 TASKS
    ↓
PARALLELISM

E é exatamente por isso que particionamento merece tanta atenção.

17. O modelo mental que eu gostaria de ter aprendido primeiro

Se eu pudesse resumir toda essa arquitetura em um único desenho, seria este:

                    ┌───────────────────────┐
                    │      APPLICATION      │
                    └───────────┬───────────┘
                                │
                                ▼
                    ┌───────────────────────┐
                    │        DRIVER         │
                    │ coordena a aplicação │
                    └───────────┬───────────┘
                                │
                                ▼
                    ┌───────────────────────┐
                    │    CLUSTER MANAGER    │
                    │  gerencia recursos   │
                    └───────────┬───────────┘
                                │
                ┌───────────────┼───────────────┐
                │               │               │
                ▼               ▼               ▼
          ┌──────────┐    ┌──────────┐    ┌──────────┐
          │Executor 1│    │Executor 2│    │Executor 3│
          └────┬─────┘    └────┬─────┘    └────┬─────┘
               │               │               │
          ┌────┴────┐     ┌────┴────┐     ┌────┴────┐
          │  Tasks  │     │  Tasks  │     │  Tasks  │
          └────┬────┘     └────┬────┘     └────┬────┘
               │               │               │
          Partitions      Partitions      Partitions

Enquanto isso, do ponto de vista lógico:

ACTION
   │
   ▼
 JOB
   │
   ├───────────────┐
   ▼               ▼
STAGE 1         STAGE 2
   │               │
 Tasks           Tasks

Esses dois desenhos juntos explicam uma grande parte do comportamento de uma aplicação Spark.

18. Da teoria para troubleshooting

Agora podemos transformar sintomas em perguntas técnicas.

Aplicação lenta

Em vez de:

"Spark está lento."

Podemos perguntar:

Qual Job está lento?
        ↓
Qual Stage?
        ↓
Quais Tasks?
        ↓
Quais Partitions?
        ↓
Em quais Executors?

OutOfMemory

Em vez de apenas aumentar memória:

OutOfMemory
    │
    ├── Driver?
    │
    └── Executor?

Se for Executor:

Executor OOM
    │
    ├── partição muito grande?
    ├── data skew?
    ├── cache excessivo?
    ├── operação cara?
    └── overhead?

A arquitetura transforma tentativa e erro em investigação.

19. O que mudou e o que continua igual

As APIs do Spark evoluíram enormemente.

Saímos de um mundo muito centrado em:

SparkContext
     +
    RDD

para um ecossistema no qual trabalhamos frequentemente com:

SparkSession
     +
 DataFrames
     +
 Spark SQL

Mas alguns conceitos continuam fundamentais:

Driver;
Executors;
Jobs;
Stages;
Tasks;
Partitions;
Shuffle;
paralelismo;
memória distribuída;
movimentação de dados.

É por isso que aprender arquitetura continua valendo a pena mesmo quando escrevemos praticamente todo o código utilizando DataFrames e SQL.

A principal lição deste capítulo

Se existe uma ideia que quero deixar deste artigo é esta:

Código Spark é apenas a descrição de uma computação distribuída.

Para entender performance, precisamos enxergar a execução por trás do código.

Quando vemos:

df.groupBy("customer_id").count()

devemos começar a imaginar:

Data
 ↓
Partitions
 ↓
Tasks
 ↓
Executors
 ↓
Shuffle
 ↓
New Stage
 ↓
Tasks
 ↓
Result

É essa mudança de perspectiva que começa a separar simplesmente usar Spark de realmente entender Spark.

Próximo artigo

Agora sabemos quem executa o trabalho.

Mas deixamos uma questão enorme em aberto:

Como Spark decide quantas partes do trabalho podem ser executadas em paralelo?

A resposta passa diretamente por:

Partições

No próximo artigo vamos entender:

o que realmente é uma partition;
relação entre partitions e tasks;
paralelismo;
repartition();
coalesce();
partições pequenas demais;
partições grandes demais;
como particionamento afeta CPU, memória e rede;
e por que um cluster enorme pode ficar quase parado mesmo tendo recursos disponíveis.

Vamos chegar a uma das relações mais importantes de toda a série:

Dados
  ↓
Partitions
  ↓
Tasks
  ↓
Parallelism
  ↓
Performance

← Parte 1: Por que o Spark mudou o processamento de Big Data

Próximo: Parte 3 — Partições no Apache Spark: a base do paralelismo e da performance

Como aprendi Apache Spark — Parte 1: Por que o Spark mudou o processamento de Big Data

renanpyd — Tue, 06 Oct 2026 06:26:49 +0000

Hoje falamos naturalmente sobre Data Lakes, Lakehouses, processamento distribuído em cloud, Delta Lake, Apache Iceberg, pipelines em tempo real e plataformas como Databricks.

Naquele período, porém, uma palavra dominava grande parte das discussões sobre processamento distribuído:

Hadoop.

Mais especificamente, Hadoop MapReduce.

Entender esse contexto é importante porque Spark não surgiu simplesmente como "mais um framework de Big Data".

Ele nasceu tentando resolver limitações reais do processamento distribuído da época.

E foi justamente isso que chamou minha atenção.

Antes do Spark: o mundo era muito mais MapReduce

Imagine que temos alguns terabytes de logs e precisamos descobrir quantas vezes cada palavra aparece nesses dados.

Conceitualmente, podemos dividir o problema em duas etapas:

MAP
documento -> palavras -> (palavra, 1)

REDUCE
(palavra, 1), (palavra, 1), (palavra, 1)
                     ↓
              (palavra, 3)

Essa ideia é extremamente poderosa.

Podemos dividir um dataset enorme entre diversas máquinas, processar partes dele paralelamente e depois combinar os resultados.

O Hadoop transformou esse modelo em uma plataforma capaz de processar volumes enormes de dados utilizando clusters de máquinas.

Mas existia um problema.

Para determinados workloads, principalmente aqueles compostos por múltiplas etapas, havia muita leitura e escrita de dados entre operações.

Imagine um algoritmo com várias etapas:

Dados
  ↓
Job 1
  ↓
Disco
  ↓
Job 2
  ↓
Disco
  ↓
Job 3
  ↓
Resultado

Quando comecei a estudar processamento distribuído, uma das primeiras coisas que percebi foi:

Em Big Data, mover dados frequentemente custa mais do que executar a própria transformação.

Rede custa.

Disco custa.

Serialização custa.

Shuffle custa.

E essa observação continua extremamente relevante na Engenharia de Dados atual.

Então apareceu o Spark

Apache Spark nasceu como um projeto de pesquisa no AMPLab da University of California, Berkeley.

Uma das ideias centrais era permitir que determinados conjuntos de dados fossem reutilizados eficientemente durante diferentes etapas de uma computação distribuída.

Isso era especialmente interessante para workloads iterativos.

Pense, por exemplo, em algoritmos que repetem operações sobre os mesmos dados:

Dataset
   ↓
Iteração 1
   ↓
Iteração 2
   ↓
Iteração 3
   ↓
Iteração 4
   ↓
Resultado

Machine Learning é um ótimo exemplo.

Muitos algoritmos precisam executar várias iterações sobre um mesmo conjunto de dados.

Se cada etapa precisar reconstruir todo o estado intermediário utilizando armazenamento persistente, o custo pode crescer rapidamente.

Spark propôs uma abstração muito interessante para esse problema.

Os RDDs.

RDD: Resilient Distributed Dataset

RDD significa:

Resilient Distributed Dataset

Em português, podemos pensar em algo como:

conjunto de dados distribuído e resiliente.

Mas essa tradução não explica por que a ideia foi tão importante.

Um RDD representa uma coleção de elementos distribuída pelas máquinas de um cluster e capaz de ser processada paralelamente.

Visualmente:

                    RDD
                     │
          ┌──────────┼──────────┐
          │          │          │
          ▼          ▼          ▼
      Partição 1  Partição 2  Partição 3
          │          │          │
          ▼          ▼          ▼
      Executor A  Executor B  Executor C

Isso permite que diferentes partes do conjunto de dados sejam processadas simultaneamente.

Mas a parte mais interessante está no Resilient.

Por que "Resilient"?

Clusters falham.

Máquinas podem parar.

Executors podem morrer.

Processos podem ser encerrados.

Nós podem ficar indisponíveis.

Uma plataforma de processamento distribuído precisa assumir que falhas vão acontecer.

Spark consegue reconstruir partições perdidas utilizando informações sobre como aquele dataset foi produzido.

Essa cadeia de dependências é chamada de:

lineage.

Imagine:

Arquivo
   │
   ▼
textFile()
   │
   ▼
  RDD A
   │
   │ filter()
   ▼
  RDD B
   │
   │ map()
   ▼
  RDD C

Spark conhece essa sequência de transformações.

Se uma partição de RDD C for perdida, ele não necessariamente precisa manter uma cópia completa de tudo.

Em muitos casos, pode reconstruir a partição a partir do lineage.

Essa ideia é fundamental para entender a arquitetura do Spark.

Um primeiro RDD em PySpark

Vamos olhar um exemplo extremamente simples.

numeros = sc.parallelize([1, 2, 3, 4, 5])

Aqui criamos um RDD contendo cinco elementos.

Agora podemos aplicar uma transformação:

dobrados = numeros.map(lambda x: x * 2)

Intuitivamente, poderíamos imaginar:

[1, 2, 3, 4, 5]

       map(x * 2)

[2, 4, 6, 8, 10]

Mas existe um detalhe fundamental.

Quando executamos:

dobrados = numeros.map(lambda x: x * 2)

Spark não precisa calcular imediatamente o resultado.

Ele registra a transformação.

Esse comportamento é chamado de:

Lazy Evaluation

E esse é um dos conceitos mais importantes de toda a arquitetura Spark.

Spark não executa tudo imediatamente

Considere:

numeros = sc.parallelize([1, 2, 3, 4, 5])

pares = numeros.filter(lambda x: x % 2 == 0)

dobrados = pares.map(lambda x: x * 2)

Podemos imaginar o plano:

parallelize
     │
     ▼
   filter
     │
     ▼
    map

Até aqui estamos descrevendo o que queremos fazer.

Agora executamos:

resultado = dobrados.collect()

print(resultado)

Resultado:

[4, 8]

collect() é uma ação.

É nesse momento que Spark precisa efetivamente produzir o resultado.

Essa separação nos leva a dois conceitos fundamentais.

Transformations vs Actions

Grande parte da programação com Spark pode ser compreendida através dessa divisão.

Transformations

Transformations criam um novo dataset a partir de outro.

Exemplos:

rdd.map(...)
rdd.filter(...)
rdd.flatMap(...)
rdd.distinct()

Normalmente elas são avaliadas de forma lazy.

Podemos imaginar:

RDD
 │
 ├── filter()
 │
 ▼
RDD
 │
 ├── map()
 │
 ▼
RDD

Estamos construindo um plano de transformação.

Actions

Actions solicitam algum resultado da computação.

Exemplos clássicos:

rdd.count()
rdd.collect()
rdd.first()
rdd.take(10)

Quando uma Action é executada, Spark precisa avaliar as transformações necessárias para produzir o resultado.

Por exemplo:

resultado = (
    sc.parallelize(range(100))
      .filter(lambda x: x % 2 == 0)
      .map(lambda x: x * 10)
      .take(5)
)

Temos:

Dados
  │
  ▼
filter()
  │
  ▼
map()
  │
  ▼
take(5)
  │
  ▼
Resultado

Esse modelo permite ao Spark entender uma sequência de operações antes da execução.

Mais tarde veremos como isso se relaciona com jobs, stages, tasks e DAGs.

O clássico WordCount

Nenhuma viagem pela história do processamento distribuído estaria completa sem ele.

O famoso:

WordCount.

linhas = sc.textFile("dados.txt")

palavras = linhas.flatMap(
    lambda linha: linha.split()
)

pares = palavras.map(
    lambda palavra: (palavra, 1)
)

contagem = pares.reduceByKey(
    lambda a, b: a + b
)

resultado = contagem.collect()

Conceitualmente:

Arquivo
   │
   ▼
Linhas
   │
   │ flatMap()
   ▼
Palavras
   │
   │ map()
   ▼
(palavra, 1)
   │
   │ reduceByKey()
   ▼
(palavra, quantidade)

Esse pequeno exemplo contém vários conceitos importantes:

leitura distribuída;
transformação dos dados;
criação de pares chave/valor;
agregação;
execução distribuída;
movimentação de dados entre partições.

E existe uma palavra especialmente importante nessa lista:

movimentação.

Porque reduceByKey() pode exigir que registros com a mesma chave sejam reunidos.

Isso nos leva a um dos assuntos mais importantes de performance em Spark:

shuffle.

Vamos dedicar um artigo específico a isso mais adiante.

Spark não é simplesmente "Hadoop mais rápido"

Durante muito tempo tornou-se comum encontrar explicações como:

"Spark é Hadoop em memória."

Essa descrição é conveniente.

Mas é incompleta.

Spark não é apenas uma implementação mais rápida de MapReduce.

Ele oferece um modelo de execução diferente e uma API capaz de representar pipelines de transformação muito mais naturalmente.

Compare conceitualmente:

MapReduce

Job
 ↓
Disco
 ↓
Job
 ↓
Disco
 ↓
Job

com:

Spark

Transformação
     ↓
Transformação
     ↓
Transformação
     ↓
Action

Essa mudança no modelo de programação é tão importante quanto a questão de performance.

Spark também não mantém "tudo em memória"

Outro erro que encontrei muitas vezes ao aprender Spark foi:

"Spark coloca todos os dados na RAM."

Não.

Spark pode manter dados em memória quando isso é útil e possível, mas isso não significa que todo processamento acontece exclusivamente na memória.

Dependendo da operação, configuração e volume de dados, Spark pode utilizar:

memória;
disco;
rede;
armazenamento externo;
recomputação.

O modelo real é muito mais sofisticado.

E entender isso evita várias decisões ruins de arquitetura.

O que continua relevante em 2026?

Hoje, na maior parte dos pipelines modernos, provavelmente não começaremos escrevendo diretamente RDDs.

É muito mais comum trabalharmos com:

df = spark.read.parquet("/dados/clientes")

e depois:

resultado = (
    df
    .filter(df.ativo == True)
    .groupBy("estado")
    .count()
)

Ou utilizando SQL:

SELECT
    estado,
    COUNT(*) AS quantidade
FROM clientes
WHERE ativo = true
GROUP BY estado;

DataFrames e Spark SQL permitem que o Spark realize otimizações muito mais sofisticadas sobre as consultas.

Então por que estudar RDD?

Porque vários conceitos fundamentais continuam aparecendo por baixo das abstrações modernas:

DataFrame / SQL
       │
       ▼
Plano lógico
       │
       ▼
Otimização
       │
       ▼
Plano físico
       │
       ▼
Stages
       │
       ▼
Tasks
       │
       ▼
Executors

Quando alguma coisa fica lenta, cara ou instável, conhecer apenas:

df.groupBy(...)

geralmente não é suficiente.

Precisamos entender o que está acontecendo por baixo da API.

Uma lição que continua válida

Uma das coisas mais importantes que aprendi estudando Spark foi que Engenharia de Dados distribuída não consiste apenas em escrever transformações.

Precisamos pensar em:

Dados
  ↓
Particionamento
  ↓
Transformações
  ↓
Dependências
  ↓
Shuffle
  ↓
Stages
  ↓
Tasks
  ↓
Executors
  ↓
CPU / Memória / Disco / Rede

Quando começamos a enxergar um pipeline dessa forma, vários problemas deixam de parecer misteriosos.

Um job lento deixa de ser simplesmente:

"Spark está lento."

E passa a gerar perguntas melhores:

Houve shuffle?

Existe data skew?

As partições estão bem dimensionadas?

Estamos movimentando dados desnecessariamente?

Algum executor está sofrendo pressão de memória?

O driver está recebendo dados demais?

Estamos usando a abstração correta?

Essas perguntas continuam extremamente atuais.

Daqui para frente

Neste artigo vimos os fundamentos que me fizeram começar a olhar Spark de forma diferente:

o contexto do Hadoop MapReduce;
processamento distribuído;
RDDs;
resiliência;
lineage;
transformations;
actions;
lazy evaluation;
e a ideia de distribuir uma computação entre várias máquinas.

Agora podemos entrar de verdade na arquitetura.

No próximo artigo vamos responder uma pergunta fundamental:

Quando executo um programa Spark, quem realmente faz o trabalho?

Vamos conhecer:

Driver, SparkContext, Cluster Manager, Executors, Jobs, Stages e Tasks.

E vamos montar o caminho completo:

Aplicação
    ↓
Driver
    ↓
Cluster Manager
    ↓
Executors
    ↓
Stages
    ↓
Tasks
    ↓
Dados

Entender esse fluxo muda completamente a maneira como diagnosticamos performance, memória e falhas em aplicações Spark.

Série: Como aprendi Apache Spark

← Parte 0: Como aprendi Apache Spark: revisitando uma jornada pela Engenharia de Dados

Próximo: Parte 2 — Por dentro da arquitetura do Apache Spark: Driver, Executors, Jobs, Stages e Tasks

Como aprendi Apache Spark: revisitando uma jornada pela Engenharia de Dados

renanpyd — Tue, 06 Oct 2026 04:05:21 +0000

Quando comecei a estudar Apache Spark, o ecossistema de Big Data era muito diferente do que conhecemos hoje.

Hadoop e MapReduce dominavam boa parte das conversas sobre processamento distribuído. Trabalhar com grandes volumes de dados normalmente significava pensar em clusters, HDFS, processamento em lote e muitas operações de leitura e escrita em disco.

> Foi nesse cenário que o Spark começou a chamar minha atenção.

**E esta série nasce justamente de uma ideia: revisitar meus antigos materiais de estudo sobre Apache Spark e reconstruí-los com o olhar que tenho hoje como profissional de Engenharia de Dados e IA.

Não quero simplesmente republicar anotações antigas.

Quero entender o que envelheceu, o que continua extremamente relevante e, principalmente, quais conceitos atravessaram diferentes gerações do Spark e continuam presentes na engenharia de dados moderna.**

## Por que revisitar Apache Spark?

Algumas tecnologias mudam tanto que estudar suas versões antigas parece arqueologia.

Spark é diferente.

As APIs evoluíram. O ecossistema mudou. RDDs deixaram de ser a principal abstração utilizada em muitas aplicações. DataFrames, Spark SQL, Catalyst, Structured Streaming e diversas otimizações passaram a ocupar um papel central.

Mas vários fundamentos continuam essenciais.

Conceitos como:

processamento distribuído;
particionamento;
transformações e ações;
lazy evaluation;
lineage;
DAGs;
shuffle;
execução distribuída;
tolerância a falhas;
serialização;
gerenciamento de memória;

continuam ajudando a explicar por que um job Spark funciona — ou por que ele fica terrivelmente lento.

E é justamente aí que estudar os fundamentos continua tendo valor.

Um aviso importante sobre esta série:
Esta não será simplesmente uma reprodução dos meus materiais antigos.

Alguns deles utilizam versões antigas do Spark e APIs que foram substituídas ou evoluíram significativamente.

Sempre que isso acontecer, vou separar três coisas:

Como funcionava
O comportamento ou API utilizada naquele momento da evolução do Spark.
O conceito por trás

A ideia de Engenharia de Dados que continua importante independentemente da versão.

Como pensamos nisso hoje: A maneira moderna de resolver ou compreender o mesmo problema. Isso significa que eventualmente veremos código antigo. E isso é proposital.

> Código legado também conta a história de uma tecnologia.

De Hadoop MapReduce ao Spark

Uma das ideias que tornou Spark especialmente interessante foi reduzir a dependência de sucessivas operações de leitura e escrita em disco durante determinados tipos de processamento.

Em workloads iterativos e análises interativas, isso fazia uma diferença enorme.

Mas seria simplista resumir Spark a:

"Spark é rápido porque processa tudo em memória."

A arquitetura é muito mais interessante.

Spark introduziu um modelo de computação distribuída baseado em abstrações capazes de manter informações sobre como os dados foram produzidos.

Uma dessas abstrações tornou-se fundamental para entender os primeiros anos do projeto:

RDD — Resilient Distributed Dataset

Um RDD representa uma coleção distribuída de dados que pode ser processada paralelamente.

Mas a parte realmente interessante não é apenas o fato de os dados estarem distribuídos.

É a combinação de:

particionamento;
imutabilidade;
operações funcionais;
execução lazy;
lineage;
recomputação;
tolerância a falhas.

Essas ideias ajudaram a criar um modelo poderoso para computação distribuída.

Em 2012, o trabalho sobre Resilient Distributed Datasets recebeu o prêmio de melhor artigo no NSDI.

Anos depois, muitas das abstrações de mais alto nível utilizadas no Spark seriam construídas sobre fundamentos estabelecidos nesse período.

E então vieram DataFrames, Spark SQL e muito mais

A história não parou nos RDDs.

O ecossistema evoluiu.

Spark SQL tornou o processamento de dados estruturados muito mais acessível.

DataFrames elevaram o nível de abstração.

O Catalyst Optimizer passou a desempenhar um papel fundamental na otimização de consultas.

O projeto Tungsten trouxe importantes avanços na execução e no gerenciamento de memória.

Structured Streaming aproximou processamento batch e streaming sob um modelo mais uniforme.

E Spark tornou-se uma das principais tecnologias da Engenharia de Dados moderna.

É essa evolução que quero explorar ao longo desta série.

O que vamos estudar

Nos próximos artigos vamos passar por assuntos como:

arquitetura do Apache Spark;
RDDs;
transformations e actions;
lazy evaluation;
SparkContext;
execução distribuída;
Client Mode e Cluster Mode;
particionamento;
joins e shuffles;
accumulators;
broadcast variables;
JSON;
DataFrames;
Spark SQL;
schemas;
fontes de dados;
window functions;
Spark Streaming;
testes;
gerenciamento de memória;
memoryOverhead;
otimização e troubleshooting.

Também vamos comparar algumas APIs históricas com a forma como resolveríamos os mesmos problemas atualmente.

O objetivo não é decorar Spark

Depois de trabalhar com Engenharia de Dados por algum tempo, uma coisa fica cada vez mais clara:

saber escrever código Spark não significa necessariamente saber Spark.

É relativamente fácil escrever:

df.groupBy("customer_id").count()

O mais importante é conseguir perguntar:

Quantas partições existem?
Haverá shuffle?
Qual será o volume movimentado pela rede?
Existe data skew?
Como o plano será executado?
O dataset cabe adequadamente nas partições?

O problema está no driver ou nos executors?

O gargalo é CPU, memória, disco ou rede?

Essa transformação realmente precisa existir?

É esse tipo de raciocínio que transforma conhecimento de uma API em conhecimento de Engenharia de Dados distribuída.

Próximo artigo

Na Parte 1, vamos começar pelos fundamentos:

O que é Apache Spark e por que sua arquitetura foi tão importante?

Vamos entender:

o problema que Spark buscava resolver;

a relação histórica com Hadoop;

o papel dos RDDs;

processamento em memória;

tolerância a falhas;

e como essas decisões influenciaram o Spark que usamos atualmente.

Esta série será uma viagem pelas versões do Spark, mas principalmente pelos conceitos de Engenharia de Dados que sobreviveram a elas.

Nos vemos na Parte 1.

DEV Community: renanpyd

Como aprendi Apache Spark — Parte 7: Memória — Executors, Driver, Cache, Spill e OutOfMemory

1. Memória em uma aplicação Spark

2. Driver Memory

3. O perigo do collect()

4. collect() não é ruim — o contexto é que importa

5. Outros caminhos para sobrecarregar o Driver

6. Executor Memory

7. Uma partição vira trabalho para uma Task

8. Particionamento e memória são inseparáveis

9. Execution Memory

10. Storage Memory

11. cache() não significa "deixe rápido"

12. Cache também é Lazy

13. persist()

14. E quando a memória não é suficiente?

Spill

15. Spill troca memória por I/O

16. Shuffle + Memory + Spill

17. Data Skew volta novamente

18. OutOfMemory não é um único problema

19. Um Driver OOM pode indicar problema de arquitetura

20. Executor OOM também pode esconder outro problema

21. spark.executor.memory

Memory Overhead

22. O que é Memory Overhead?

23. Um exemplo conceitual

24. Por que isso aparece muito com PySpark?

25. Memory Overhead no material histórico

26. Heap vs. Off-Heap vs. Overhead

27. Garbage Collection

GC

28. GC alto é sintoma

29. Mais cores por Executor também afetam memória

30. Executors gigantes também não são automaticamente melhores

31. Broadcast também consome memória

32. Cache também compete por recursos

33. unpersist()

34. O Spark UI volta a ser essencial

35. Um padrão de investigação

36. Caso 1 — Driver morrendo

37. Caso 2 — Uma Task morre repetidamente

38. Caso 3 — Todas as Tasks sofrem Spill

39. Caso 4 — OOM depois de Broadcast

40. Caso 5 — PySpark e memória externa ao heap

41. Não ajuste memoryOverhead às cegas

42. Memória em Kubernetes

43. Memória em YARN

44. A configuração depende do ambiente

45. Um modelo mental melhor para OOM

46. Performance tuning começa com observabilidade

47. Um checklist para problemas de memória

48. O que eu evitaria fazer automaticamente

49. Memória, performance e custo

50. A conexão entre tudo que vimos até aqui

51. A principal lição deste capítulo

52. Como penso hoje

Próximo artigo

Como aprendi Apache Spark — Parte 6: Spark SQL e DataFrames — do código ao plano de execução

Spark SQL e DataFrames

1. Antes dos DataFrames: o mundo dos RDDs

2. Estrutura muda tudo

3. De RDD para DataFrame

4. Spark SQL

5. SparkSession: a entrada moderna

6. Schema: Spark precisa entender os dados

7. Schema inference ou schema explícito?

8. DataFrame é declarativo

9. Lazy Evaluation continua existindo

10. Do código ao plano

11. Unresolved Logical Plan

12. Analysis

13. Logical Plan

14. Catalyst Optimizer

Catalyst

15. Predicate Pushdown

16. Partition Pruning

17. Projection Pruning

18. Columnar formats tornam isso ainda mais poderoso

19. Optimized Logical Plan

3. O perigo do `collect()`

4. `collect()` não é ruim — o contexto é que importa

11. `cache()` não significa "deixe rápido"

13. `persist()`

21. `spark.executor.memory`

33. `unpersist()`

41. Não ajuste `memoryOverhead` às cegas

21. `explain()` é nossa janela para o optimizer

22. O que procuro no `explain()`

22. Um caso clássico: valores `NULL`

34. O que eu procuro no `explain()`