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
e:
Executor Memory
Essa distinção é fundamental.
Porque:
Driver OOM
e:
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
e:
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
e:
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#cluster #spark #python #bigdata #software #coding #development #engineering #inclusive #community
Como aprendi Apache Spark — Parte 7: Memória — Executors, Driver, Cache, Spill e OutOfMemory
