Nos artigos anteriores, construímos uma visão progressiva da arquitetura do Apache Spark.
Começamos com RDDs, passamos por partições, Tasks, Executors, Shuffle, Spark SQL, memória e chegamos à tolerância a falhas.
Na Parte 8, vimos que o Spark consegue reconstruir dados perdidos utilizando lineage, dependências e recomputação.
Agora surge uma pergunta igualmente importante:
O que acontece com as variáveis do nosso programa quando o processamento sai do Driver e passa a acontecer nos Executors?
Em uma aplicação Python tradicional, podemos escrever:
contador = 0
for numero in range(10):
contador += 1
print(contador)
Resultado:
10
Tudo acontece dentro de um único processo.
Mas o Spark trabalha com computação distribuída.
Temos:
DRIVER
|
Distribuição de Tasks
|
+---------+---------+
| | |
v v v
Executor 1 Executor 2 Executor 3
| | |
Task A Task B Task C
E isso muda profundamente o significado de uma variável.
Neste capítulo, vamos entender:
- Closures e serialização.
- Variáveis locais do Driver.
- Broadcast Variables.
- Accumulators.
- Diferenças entre variáveis comuns e mecanismos distribuídos.
- Side effects e retries.
- Problemas de consistência.
- Boas práticas em PySpark.
- Quando usar cada mecanismo.
1. O problema das variáveis em sistemas distribuídos
Considere este exemplo:
from pyspark.sql import SparkSession
spark = (
SparkSession.builder
.appName("shared-variables")
.getOrCreate()
)
sc = spark.sparkContext
numeros = sc.parallelize(
range(100),
4
)
contador = 0
def incrementar(numero):
global contador
contador += 1
return numero
resultado = numeros.map(incrementar)
resultado.count()
print(contador)
Alguém acostumado com Python local poderia esperar:
100
Mas em uma execução distribuída típica, o valor observado no Driver permanece:
0
Por quê?
Porque as Tasks não executam diretamente dentro do mesmo espaço de memória do Driver.
O Spark serializa funções e os dados necessários para executá-las em outros processos.
O contador utilizado pelos workers não é um contador global compartilhado com o Driver.
Essa é nossa primeira lição:
Uma variável Python comum não se transforma automaticamente em estado compartilhado apenas porque está sendo utilizada dentro de uma transformação Spark.
2. Driver e Executors possuem espaços de memória diferentes
Vamos visualizar:
DRIVER
|
| contador = 0
|
+-------------------------------+
|
Serialização
|
+-----------------+-----------------+
| | |
v v v
EXECUTOR 1 EXECUTOR 2 EXECUTOR 3
| | |
Python Worker Python Worker Python Worker
| | |
Estado local Estado local Estado local
O código enviado para execução remota não passa a compartilhar livremente a memória do processo original.
Mesmo quando diferentes Tasks são executadas no mesmo Executor, não devemos assumir que variáveis Python comuns constituem um estado global consistente entre elas.
Isso também significa que:
minha_lista.append(valor)
dentro de uma Task não é uma maneira confiável de preencher uma lista existente no Driver.
Para produzir resultados distribuídos, normalmente devemos utilizar as próprias transformações e Actions do Spark.
3. O que são Closures?
Uma closure é uma função que utiliza valores definidos em seu contexto externo.
Por exemplo:
multiplicador = 10
def multiplicar(numero):
return numero * multiplicador
A função utiliza:
multiplicador
mesmo que essa variável não tenha sido definida dentro dela.
Agora utilizamos a função no Spark:
multiplicador = 10
numeros = sc.parallelize([1, 2, 3, 4])
resultado = numeros.map(
lambda numero: numero * multiplicador
)
print(resultado.collect())
Resultado esperado:
[10, 20, 30, 40]
Nesse caso, a função utiliza o valor necessário durante sua execução distribuída.
O Spark precisa preparar e serializar a computação que será enviada aos workers.
Isso funciona bem para valores pequenos e serializáveis.
Mas existem consequências importantes.
4. Serialização de Closures
Imagine:
configuracao = {
"taxa": 1.15,
"moeda": "BRL"
}
Depois:
resultado = numeros.map(
lambda numero: numero * configuracao["taxa"]
)
O worker precisa ter acesso aos valores necessários para executar a função.
Em PySpark, a serialização permite transportar funções e determinados objetos Python entre processos.
Mas nem todo objeto pode ser serializado corretamente.
Exemplos problemáticos incluem objetos que dependem de recursos locais, como determinadas conexões abertas, locks ou handles de arquivos.
Outro problema aparece quando uma closure captura acidentalmente uma estrutura muito grande.
5. O problema de capturar objetos gigantes
Imagine uma estrutura de referência:
referencias = {
1: "Categoria A",
2: "Categoria B",
3: "Categoria C"
}
Com poucos registros, não há grande preocupação.
Agora imagine que referencias contenha milhões de entradas.
Se essa estrutura for capturada por uma função enviada para várias Tasks, poderemos gerar custos relevantes de serialização e distribuição.
Conceitualmente:
DRIVER
|
Large Dictionary
|
Closure
|
Task Serialization
|
+----------+----------+
| | |
v v v
Task 1 Task 2 Task 3
Precisamos de um mecanismo mais apropriado quando muitos workers devem consultar os mesmos dados de referência.
É aqui que entram as Broadcast Variables.
6. O que é uma Broadcast Variable?
Uma Broadcast Variable permite disponibilizar um valor somente para leitura aos Executors de maneira eficiente.
Em vez de incluir repetidamente uma estrutura grande nas closures, podemos criar uma variável broadcast.
Exemplo:
categorias = {
1: "Eletrônicos",
2: "Livros",
3: "Alimentos"
}
categorias_broadcast = sc.broadcast(
categorias
)
Agora temos um objeto broadcast.
Para acessar seu conteúdo dentro de uma Task:
categorias_broadcast.value
Exemplo:
produtos = sc.parallelize([
(101, 1),
(102, 2),
(103, 3)
])
resultado = produtos.map(
lambda produto: (
produto[0],
categorias_broadcast.value.get(
produto[1],
"Desconhecida"
)
)
)
print(resultado.collect())
Resultado esperado:
[
(101, 'Eletrônicos'),
(102, 'Livros'),
(103, 'Alimentos')
]
Agora as Tasks consultam uma referência distribuída pelo mecanismo de broadcast.
7. Como Broadcast funciona conceitualmente?
Imagine uma tabela de referência que várias Tasks precisam utilizar.
Sem broadcast explícito:
DRIVER
|
Reference Data
|
Task Serialization
|
+-----------+-----------+
| | |
v v v
Task A Task B Task C
Com Broadcast:
DRIVER
|
Broadcast Data
|
+-----------+-----------+
| | |
v v v
Executor 1 Executor 2 Executor 3
| | |
Shared local Shared local Shared local
broadcast broadcast broadcast
| | |
Tasks Tasks Tasks
O Spark utiliza mecanismos próprios de distribuição e armazenamento para disponibilizar esses dados aos workers.
O objetivo é reduzir o custo de transportar repetidamente a mesma referência.
Entretanto, broadcast não significa memória gratuita.
Os valores precisam caber nos recursos disponíveis.
8. Broadcast Variables são somente para leitura
Uma regra importante:
Broadcast Variables devem ser tratadas como valores imutáveis distribuídos.
Não devemos utilizar:
categorias_broadcast.value[1] = "Nova categoria"
como mecanismo para atualizar uma referência global.
Modificar uma cópia local não propaga uma atualização consistente para os demais Executors.
Se a referência mudar, a aplicação deve distribuir uma nova versão conforme a necessidade.
Podemos pensar:
Broadcast Version 1
|
v
Distributed Read-Only Reference
Depois:
Updated Data
|
v
New Broadcast Version
Esse modelo evita expectativas incorretas sobre sincronização.
9. Quando usar Broadcast Variables?
Broadcast Variables são úteis quando:
- Existe uma referência relativamente pequena.
- Muitas Tasks precisam consultar os mesmos dados.
- O conteúdo não precisa ser modificado durante a execução.
- O custo de distribuir repetidamente a referência seria relevante.
Exemplos:
- Dicionários de classificação.
- Parâmetros de algoritmos.
- Regras estáticas de processamento.
- Tabelas pequenas de correspondência.
- Configurações utilizadas em transformações distribuídas.
Mas existe uma distinção importante.
10. Broadcast Variable não é exatamente Broadcast Join
Na Parte 5, utilizamos:
from pyspark.sql.functions import broadcast
resultado = pedidos.join(
broadcast(clientes),
"customer_id"
)
Agora utilizamos:
referencia = sc.broadcast(
dicionario
)
Essas operações compartilham o conceito de distribuir dados para evitar movimentação desnecessária.
Mas pertencem a mecanismos e abstrações diferentes.
Broadcast Variable
sc.broadcast(valor)
É um mecanismo para disponibilizar valores somente para leitura ao código executado nos workers.
Broadcast Join
broadcast(dataframe)
É uma indicação ao planejador do Spark SQL para considerar uma estratégia de join baseada em broadcast.
O Spark SQL pode executar um Broadcast Hash Join quando as condições e a semântica permitirem.
Não devemos tratar os dois recursos como APIs intercambiáveis.
11. Quando preferir um Join em vez de um dicionário Broadcast?
Imagine que temos:
PEDIDOS
e:
CLIENTES
Ambos como DataFrames.
Poderíamos transformar CLIENTES em um dicionário Python e utilizar uma Broadcast Variable.
Mas isso nem sempre é uma boa escolha.
Se o problema é um join relacional, geralmente é melhor permitir que o Spark SQL trabalhe com seus mecanismos nativos:
resultado = pedidos.join(
clientes,
"customer_id"
)
Ou, quando apropriado:
resultado = pedidos.join(
broadcast(clientes),
"customer_id"
)
Assim preservamos oportunidades de otimização pelo Catalyst e pelo planejador físico.
Uma boa regra é:
Para relacionamentos entre DataFrames, comece pela API nativa de join. Para referências Python utilizadas em funções distribuídas, avalie Broadcast Variables.
12. E quando precisamos acumular resultados?
Agora vamos ao problema oposto.
Broadcast Variables distribuem valores para leitura.
Mas imagine que queremos contar eventos observados pelas Tasks.
Por exemplo:
Quantos registros inválidos foram encontrados?
Talvez desejemos produzir:
Executor 1 → 12 inválidos
Executor 2 → 8 inválidos
Executor 3 → 15 inválidos
E obter:
Total = 35
Para determinados casos, Spark disponibiliza outro mecanismo:
Accumulators.
13. O que é um Accumulator?
Um Accumulator permite que Tasks contribuam para um valor agregado que pode ser consultado pelo Driver.
Em PySpark:
invalidos = sc.accumulator(0)
Agora podemos incrementá-lo durante uma computação distribuída.
Exemplo:
dados = sc.parallelize([
10,
-5,
20,
-1,
30
])
invalidos = sc.accumulator(0)
def validar(numero):
if numero < 0:
invalidos.add(1)
return numero
dados.map(validar).count()
print(invalidos.value)
Em uma execução simples, o resultado esperado é:
2
Mas precisamos entender o comportamento antes de utilizar Accumulators em sistemas reais.
14. Como Accumulators funcionam conceitualmente?
DRIVER
|
Accumulator
|
+----------+----------+
| | |
v v v
Executor 1 Executor 2 Executor 3
| | |
add(10) add(5) add(8)
| | |
+----------+----------+
|
v
DRIVER
|
v
23
As Tasks contribuem com atualizações.
O Driver consulta o valor agregado.
Diferentemente de uma variável Python comum, o Spark possui um mecanismo próprio para essas contribuições.
Mas isso não transforma o Accumulator em uma variável global de leitura e escrita livre.
15. Broadcast vs. Accumulator
| Característica | Broadcast Variable | Accumulator |
|---|---|---|
| Objetivo | Distribuir dados de referência | Agregar contribuições |
| Fluxo conceitual | Driver → Executors | Tasks → Driver |
| Leitura pelas Tasks | Sim | Não como estado global |
| Atualização pelas Tasks | Não deve ocorrer | Sim, via add()
|
| Uso típico | Lookup e parâmetros | Contadores e métricas |
| Substitui DataFrame? | Não | Não |
Uma forma simples de memorizar:
BROADCAST
Driver
|
v
Executors
ACCUMULATOR
Tasks
|
v
Driver
Essa distinção é o coração do capítulo.
16. Accumulators não devem controlar a lógica do pipeline
Imagine uma regra:
Se contador global ultrapassar 1000,
interrompa o processamento.
Parece simples.
Mas um Accumulator não fornece às Tasks uma visão global sincronizada e atualizada em tempo real.
Portanto, não devemos construir lógica distribuída dependente da leitura de um Accumulator como se fosse uma variável compartilhada convencional.
Seu uso é mais apropriado para observabilidade e agregações auxiliares.
Exemplos:
Registros inválidos
Eventos descartados
Erros de parsing
Ocorrências de determinado padrão
Mesmo nesses casos, precisamos compreender retries e recomputações.
17. O problema dos Side Effects
Agora entramos em um assunto extremamente importante.
Considere:
def processar(registro):
enviar_para_api(registro)
return registro
Depois:
df.rdd.map(processar).count()
Parece razoável.
Mas existe um problema:
Tasks podem ser executadas novamente.
Lembre-se da Parte 8:
Task
|
v
Failure
|
v
Retry
Se a função executa uma operação externa com efeito permanente, uma nova tentativa pode repetir esse efeito.
Por exemplo:
Task Attempt 1
|
v
POST /pagamento
|
v
Sucesso externo
|
v
Falha antes da conclusão da Task
|
v
Retry
|
v
POST /pagamento novamente
Agora temos uma duplicação potencialmente grave.
A tolerância a falhas do Spark não garante automaticamente que efeitos externos aconteçam exatamente uma vez.
18. Retries e Accumulators
Accumulators também precisam ser interpretados com cuidado.
A semântica das atualizações depende do contexto de execução.
Em particular, atualizações realizadas dentro de transformações podem produzir resultados surpreendentes quando existem novas Actions, recomputações ou tentativas de execução.
Por exemplo:
contador = sc.accumulator(0)
rdd = sc.parallelize(
[1, 2, 3, 4]
)
def contar(numero):
contador.add(1)
return numero
transformado = rdd.map(contar)
Agora:
transformado.count()
e depois:
transformado.collect()
podem executar a transformação novamente.
Portanto, o contador pode acumular contribuições de mais de uma Action.
Isso não significa que existam mais registros únicos na fonte.
Significa que estamos contando execuções de código.
19. Contar registros é diferente de contar execuções
Essa diferença merece destaque.
Imagine:
Dataset:
100 registros
Executamos:
Action 1
Depois:
Action 2
Se uma transformação que atualiza o Accumulator for executada nas duas Actions, as contribuições podem ocorrer novamente.
Portanto:
Número de registros de negócio
não é necessariamente igual a:
Número de vezes que uma função foi executada
Essa distinção é essencial em sistemas com lazy evaluation e tolerância a falhas.
20. Para métricas exatas, prefira operações determinísticas
Se precisamos contar registros inválidos com precisão como resultado de negócio, podemos usar as APIs nativas do Spark.
Por exemplo:
from pyspark.sql.functions import col
invalidos = (
df
.filter(col("valor") < 0)
.count()
)
Agora o resultado vem de uma operação distribuída que calcula explicitamente a quantidade de linhas que satisfazem a condição.
Para múltiplas métricas:
from pyspark.sql.functions import (
col,
count,
sum as spark_sum,
when
)
metricas = df.agg(
count("*").alias("total_registros"),
spark_sum(
when(col("valor") < 0, 1).otherwise(0)
).alias("registros_invalidos")
)
metricas.show()
Essa abordagem é mais adequada quando o número calculado faz parte da lógica de negócio ou de uma validação de qualidade.
Accumulators continuam úteis, mas não deveriam substituir automaticamente agregações distribuídas.
21. Closures podem carregar estado inesperado
Considere uma classe:
class Processador:
def __init__(self):
self.configuracao = {
"taxa": 1.1
}
def transformar(self, valor):
return valor * self.configuracao["taxa"]
Agora:
processador = Processador()
resultado = rdd.map(
processador.transformar
)
Dependendo do objeto e de suas referências, a serialização pode incluir mais estado do que imaginamos.
Em aplicações maiores, isso pode gerar:
- Serialização custosa.
- Objetos não serializáveis.
- Closures grandes.
- Dependências desnecessárias.
- Problemas difíceis de diagnosticar.
Uma alternativa é capturar apenas o valor necessário:
taxa = processador.configuracao["taxa"]
resultado = rdd.map(
lambda valor: valor * taxa
)
Ou, se a referência for grande e reutilizada:
config_broadcast = sc.broadcast(
processador.configuracao
)
A escolha depende do volume e do padrão de utilização.
22. Evite criar conexões externas no Driver para enviá-las às Tasks
Imagine:
conexao = criar_conexao_banco()
Depois:
rdd.map(
lambda registro: conexao.executar(registro)
)
Isso é problemático.
Conexões abertas geralmente estão associadas ao processo, à rede e ao runtime onde foram criadas.
Elas não deveriam ser tratadas como objetos Python comuns que podemos transportar livremente para outros processos.
Uma estratégia mais apropriada, quando realmente precisamos de integração externa por Tasks, pode envolver inicialização controlada por partição.
Por exemplo:
def processar_particao(registros):
conexao = criar_conexao()
try:
for registro in registros:
yield processar_com_conexao(
conexao,
registro
)
finally:
conexao.close()
E:
resultado = rdd.mapPartitions(
processar_particao
)
Mas existe uma ressalva fundamental: operações externas precisam considerar idempotência, retries, limites de conexão e tratamento de falhas.
Em pipelines de produção, conectores nativos e mecanismos de escrita transacional geralmente são preferíveis quando disponíveis.
23. Por que mapPartitions() pode ajudar?
Imagine:
Partition 1
|
+-- 100.000 registros
Se criarmos uma conexão por registro:
100.000 conexões
podemos gerar overhead enorme.
Com processamento por partição, podemos reutilizar uma conexão dentro da execução daquela partição.
Conceitualmente:
Task
|
v
Open Connection
|
+-- Record 1
+-- Record 2
+-- Record 3
+-- ...
|
v
Close Connection
Isso pode reduzir overhead.
Mas não resolve automaticamente o problema de duplicação causado por retries.
O sistema externo ainda precisa estar preparado.
24. Idempotência é essencial
Uma operação é idempotente quando repeti-la não altera o resultado final além da primeira aplicação bem-sucedida.
Por exemplo, uma gravação baseada em chave única com semântica apropriada de upsert pode ser projetada para tolerar repetição.
Conceitualmente:
Task Attempt 1
|
v
UPSERT customer_id=123
|
v
Retry
|
v
UPSERT customer_id=123
Se a operação for corretamente idempotente, a repetição não deve gerar um registro duplicado.
Agora compare com:
INSERT sem chave de deduplicação
Repetições podem gerar múltiplos registros.
Essa preocupação não é específica de Spark.
É um princípio geral de sistemas distribuídos.
25. Broadcast não substitui gerenciamento de estado
Imagine uma referência que muda continuamente:
Preço atual
Estoque atual
Status do pedido
Saldo financeiro
Se os dados precisam ser atualizados em tempo real e compartilhados consistentemente entre workers, uma Broadcast Variable não é um banco de dados distribuído.
Ela representa uma referência distribuída somente para leitura.
Portanto:
Broadcast
≠
Distributed Mutable Database
Se precisamos de estado mutável com garantias específicas, devemos escolher uma arquitetura apropriada.
Isso pode envolver:
- Sistemas externos de armazenamento.
- Operações transacionais.
- Processamento estruturado de estado.
- Estratégias de consistência.
- Mecanismos específicos de streaming.
O importante é não atribuir a Broadcast Variables capacidades que elas não possuem.
26. Accumulator não substitui agregação distribuída
Da mesma forma:
Accumulator
≠
Distributed Database
E também:
Accumulator
≠
Reliable Business Counter
em todos os contextos.
Para contar linhas:
df.count()
Para somar valores:
df.agg(
{"valor": "sum"}
)
Para agrupar:
df.groupBy(
"categoria"
).count()
Essas operações representam resultados de dados.
Accumulators representam um mecanismo auxiliar de agregação de contribuições das Tasks.
A diferença é sutil, mas importante.
27. Um exemplo prático: validação de dados
Imagine uma ingestão de eventos.
Cada evento possui:
event_id
customer_id
event_type
value
Queremos identificar registros inválidos.
Podemos utilizar Spark SQL:
from pyspark.sql.functions import (
col,
when,
count,
sum as spark_sum
)
validado = df.withColumn(
"registro_valido",
(
col("customer_id").isNotNull()
& col("event_type").isNotNull()
& (col("value") >= 0)
)
)
metricas = validado.agg(
count("*").alias("total"),
spark_sum(
when(
col("registro_valido"),
1
).otherwise(0)
).alias("validos"),
spark_sum(
when(
~col("registro_valido"),
1
).otherwise(0)
).alias("invalidos")
)
metricas.show()
Essa solução é declarativa.
O Spark entende as expressões.
O resultado pode ser utilizado como métrica de qualidade de dados.
Não precisamos de um Accumulator para essa contagem.
28. Um exemplo prático: classificação por referência
Agora imagine que precisamos aplicar uma pequena referência Python a registros de um RDD.
classificacoes = {
"A": "Alta prioridade",
"B": "Média prioridade",
"C": "Baixa prioridade"
}
Criamos:
classificacoes_bc = sc.broadcast(
classificacoes
)
Depois:
eventos = sc.parallelize([
("EVT001", "A"),
("EVT002", "C"),
("EVT003", "B")
])
resultado = eventos.map(
lambda evento: (
evento[0],
classificacoes_bc.value.get(
evento[1],
"Não classificado"
)
)
)
print(resultado.collect())
Resultado:
[
('EVT001', 'Alta prioridade'),
('EVT002', 'Baixa prioridade'),
('EVT003', 'Média prioridade')
]
Esse é um caso simples em que a Broadcast Variable representa bem a intenção.
29. Um exemplo prático: observabilidade com Accumulator
Imagine que estamos estudando um parser de linhas.
linhas = sc.parallelize([
"10",
"20",
"erro",
"30",
"invalido"
])
Criamos:
falhas_parse = sc.accumulator(0)
Depois:
def converter(linha):
try:
return int(linha)
except ValueError:
falhas_parse.add(1)
return None
Executamos uma Action:
resultado = linhas.map(
converter
).collect()
print(resultado)
print(falhas_parse.value)
Em uma execução simples:
[10, 20, None, 30, None]
2
Isso demonstra o mecanismo.
Mas, em produção, eu não utilizaria automaticamente o valor do Accumulator como uma métrica de negócio com garantia de contagem exata.
Para isso, preferiria estruturar os resultados e agregá-los explicitamente.
30. O impacto da Lazy Evaluation
Vamos conectar este capítulo aos anteriores.
Considere:
contador = sc.accumulator(0)
transformado = rdd.map(
lambda x: registrar_e_transformar(
x,
contador
)
)
Até aqui:
Transformation definida
Mas ainda não necessariamente executada.
Quando fazemos:
transformado.count()
a computação acontece.
Depois:
transformado.collect()
ela pode acontecer novamente.
Isso significa que side effects dentro de transformações precisam ser tratados com muito cuidado.
A combinação:
Lazy Evaluation
+
Retries
+
Recomputation
+
Side Effects
é uma das fontes clássicas de comportamento inesperado em sistemas distribuídos.
31. A relação com Fault Tolerance
Na Parte 8, vimos:
Task Failed
|
v
Retry
|
v
Recomputation
Agora entendemos uma consequência importante.
Se a Task executa apenas transformações determinísticas:
lambda x: x * 2
recomputar é relativamente simples.
Mas se a Task executa:
enviar_email()
ou:
cobrar_pagamento()
ou:
inserir_registro_externo()
recomputar pode repetir efeitos no mundo externo.
Portanto:
A capacidade de recomputar dados não equivale à garantia de executar efeitos externos exatamente uma vez.
Essa distinção é fundamental para projetar pipelines confiáveis.
32. Variáveis compartilhadas e performance
Broadcast Variables também estão relacionadas a performance.
Imagine:
1000 Tasks
e uma referência Python relativamente grande.
Se cada Task precisar receber repetidamente essa estrutura, podemos gerar overhead.
Com Broadcast, Spark pode distribuir e reutilizar a referência de maneira mais eficiente.
Mas isso precisa ser equilibrado com:
Serialization
Distribution
Executor Memory
Object Size
Reuse
Broadcast é uma ferramenta de otimização.
Não uma obrigação para toda variável.
33. O que não devemos fazer
Alguns padrões merecem atenção especial.
Modificar variáveis Python locais esperando que o Driver enxergue as alterações:
contador = 0
def funcao(x):
global contador
contador += 1
return x
Utilizar Broadcast Variable como estado mutável global:
bc.value["contador"] += 1
Utilizar Accumulator para controlar decisões distribuídas em tempo real.
Executar operações externas não idempotentes dentro de Tasks sem estratégia de recuperação.
Capturar objetos enormes ou não serializáveis em closures.
Forçar Broadcast Join sem avaliar o tamanho real do dataset.
Todos esses padrões podem produzir problemas de correção, performance ou estabilidade.
34. Um modelo mental para escolher o mecanismo
Quando preciso compartilhar informações em Spark, começo perguntando:
Qual é o objetivo?
|
+-----------------------------+
| |
v v
Fornecer referência Produzir resultado
aos workers agregado
| |
v v
É somente leitura? É dado de negócio?
| |
v v
Broadcast Variable? DataFrame Aggregation?
|
v
Métrica auxiliar?
|
v
Accumulator?
E, se a necessidade envolver:
Estado mutável consistente
ou:
Efeitos externos
precisamos de uma arquitetura apropriada, não apenas de uma variável compartilhada.
35. O que eu procuro em uma revisão de código
Quando reviso um pipeline PySpark, gosto de investigar:
- A closure captura objetos desnecessários?
- Algum objeto não serializável está sendo enviado aos workers?
- Uma referência grande é distribuída repetidamente?
- Broadcast seria apropriado?
- Existe um join nativo mais adequado?
- Alguma Task modifica estado local esperando atualização no Driver?
- Accumulators estão sendo usados como métricas de negócio?
- Existem side effects?
- A operação externa é idempotente?
- O código suporta retries?
- O comportamento permanece correto após recomputação?
- O resultado é determinístico?
Essas perguntas vão muito além de saber utilizar:
map()
ou:
filter()
Elas fazem parte da engenharia de sistemas distribuídos.
36. A principal lição deste capítulo
Em uma aplicação local, podemos pensar:
Uma variável
|
v
Um processo
|
v
Uma memória
Em Spark:
Driver
|
+--------------------------+
| |
v v
Executor 1 Executor 2
| |
Python Worker Python Worker
| |
Memória local Memória local
Isso exige mecanismos explícitos para determinados padrões de comunicação.
Broadcast Variables ajudam a distribuir referências somente para leitura.
Accumulators permitem contribuições agregadas das Tasks para o Driver.
Closures transportam funções e valores necessários para execução.
E a tolerância a falhas significa que qualquer efeito colateral precisa ser pensado considerando possíveis reexecuções.
37. Como penso hoje
Quando vejo:
resultado = rdd.map(
minha_funcao
)
não penso apenas:
Aplicar função a cada elemento
Também penso:
Onde a função será executada?
|
v
O que será serializado?
|
v
Quais variáveis ela captura?
|
v
Existe estado compartilhado?
|
v
Existe side effect?
|
v
O que acontece em um retry?
|
v
O resultado continua correto?
Essa mudança de perspectiva é uma das partes mais importantes de aprender computação distribuída.
Porque o desafio não é apenas executar código em várias máquinas.
É garantir que o resultado continue correto quando o sistema se comporta como um sistema distribuído de verdade.
Próximo artigo
Agora entendemos como funções, variáveis e estado se comportam quando atravessam processos.
No próximo artigo vamos entrar em outro tema importante:
Modos de execução e implantação do Apache Spark.
Vamos explorar:
- Client Mode.
- Cluster Mode.
- Onde o Driver executa.
- Onde os Executors executam.
- Cluster Managers.
- Spark Standalone.
- YARN.
- Kubernetes.
-
spark-submit. - Diferenças entre desenvolvimento local e produção.
- Logs e observabilidade.
- Falhas de Driver e Executors.
- Boas práticas de implantação.
A pergunta central será:
Quando submetemos uma aplicação Spark, onde cada componente realmente está executando?
Série: Como aprendi Apache Spark
← Parte 8: RDD Lineage e tolerância a falhas — como o Spark se recupera
Próximo: Parte 10 — Client Mode vs. Cluster Mode: onde o Spark realmente executa
Top comments (0)