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
estadoevalorserã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
e:
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
e:
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
e:
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?
e:
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
Top comments (0)