DEV Community

renanpyd
renanpyd

Posted on

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

#dataengineering #bigdata #pyspark #spark

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()
)
Enter fullscreen mode Exit fullscreen mode

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
Enter fullscreen mode Exit fullscreen mode

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
Enter fullscreen mode Exit fullscreen mode

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()
Enter fullscreen mode Exit fullscreen mode

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
Enter fullscreen mode Exit fullscreen mode

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
Enter fullscreen mode Exit fullscreen mode

Em PySpark, era comum encontrarmos:

from pyspark import SparkContext

sc = SparkContext(
    "local[*]",
    "MinhaAplicacao"
)
Enter fullscreen mode Exit fullscreen mode

E então:

rdd = sc.parallelize([1, 2, 3, 4, 5])
Enter fullscreen mode Exit fullscreen mode

ou:

rdd = sc.textFile("dados.txt")
Enter fullscreen mode Exit fullscreen mode

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

Podemos visualizar:

Aplicação
    │
    ▼
SparkContext
    │
    ▼
Cluster
Enter fullscreen mode Exit fullscreen mode

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

from pyspark.sql import SparkSession

spark = (
    SparkSession.builder
    .appName("MinhaAplicacao")
    .getOrCreate()
)
Enter fullscreen mode Exit fullscreen mode

E podemos acessar o contexto subjacente:

sc = spark.sparkContext
Enter fullscreen mode Exit fullscreen mode

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

Historicamente, aprender Spark significava começar por:

SparkContext
      ↓
     RDD
Enter fullscreen mode Exit fullscreen mode

Hoje é muito mais comum começar por:

SparkSession
      ↓
 DataFrame / SQL
Enter fullscreen mode Exit fullscreen mode

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
Enter fullscreen mode Exit fullscreen mode

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
Enter fullscreen mode Exit fullscreen mode

Cada Executor possui recursos como:

Executor
├── CPU / cores
├── memória
├── tasks
└── dados armazenados/cacheados
Enter fullscreen mode Exit fullscreen mode

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
Enter fullscreen mode Exit fullscreen mode

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
Enter fullscreen mode Exit fullscreen mode

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

Por exemplo:

df.collect()
Enter fullscreen mode Exit fullscreen mode

O que collect() faz?

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

Imagine um DataFrame com:

500 GB
Enter fullscreen mode Exit fullscreen mode

distribuídos entre vários executors.

Se tentarmos:

dados = df.collect()
Enter fullscreen mode Exit fullscreen mode

conceitualmente estamos pedindo:

Executor 1 ───┐
Executor 2 ───┤
Executor 3 ───┼───► DRIVER
Executor 4 ───┤
Executor 5 ───┘
Enter fullscreen mode Exit fullscreen mode

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

Temos um problema.

É por isso que um comando aparentemente inocente como:

collect()
Enter fullscreen mode Exit fullscreen mode

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
Enter fullscreen mode Exit fullscreen mode

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
Enter fullscreen mode Exit fullscreen mode

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
Enter fullscreen mode Exit fullscreen mode

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()
Enter fullscreen mode Exit fullscreen mode

Temos:

textFile()
    │
    ▼
filter()
    │
    ▼
count()
Enter fullscreen mode Exit fullscreen mode

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
Enter fullscreen mode Exit fullscreen mode

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()
Enter fullscreen mode Exit fullscreen mode

Uma representação simplificada seria:

STAGE 1

textFile
   │
flatMap
   │
 map
   │
   ▼

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

   │
   ▼

STAGE 2

reduceByKey
   │
collect
Enter fullscreen mode Exit fullscreen mode

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
Enter fullscreen mode Exit fullscreen mode

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

Algo como:

spark -> 1 ──┐
spark -> 1 ──┼──► spark -> 3
spark -> 1 ──┘
Enter fullscreen mode Exit fullscreen mode

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
Enter fullscreen mode Exit fullscreen mode

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
Enter fullscreen mode Exit fullscreen mode

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()
)
Enter fullscreen mode Exit fullscreen mode

Conceitualmente:

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

E agora temos uma hierarquia muito útil:

Application
    ↓
Action
    ↓
Job
    ↓
Stages
    ↓
Tasks
    ↓
Partitions
    ↓
Executors
Enter fullscreen mode Exit fullscreen mode

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()
Enter fullscreen mode Exit fullscreen mode

O código parece simples.

Mas internamente precisamos pensar em algo próximo de:

Leitura
  │
  ▼
Filter
  │
  ▼
GroupBy
  │
  ▼
Shuffle
  │
  ▼
Aggregation
  │
  ▼
Result
Enter fullscreen mode Exit fullscreen mode

Distribuído entre:

Driver
  │
  ├──────────┬──────────┬──────────┐
  ▼          ▼          ▼          ▼
Executor   Executor   Executor   Executor
Enter fullscreen mode Exit fullscreen mode

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

Compare:

df.filter("valor > 100")
Enter fullscreen mode Exit fullscreen mode

com:

df.groupBy("cliente_id").count()
Enter fullscreen mode Exit fullscreen mode

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(...)
Enter fullscreen mode Exit fullscreen mode

ou:

rdd.filter(...)
Enter fullscreen mode Exit fullscreen mode

Podemos imaginar:

P1 ───► P1'
P2 ───► P2'
P3 ───► P3'
Enter fullscreen mode Exit fullscreen mode

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(...)
Enter fullscreen mode Exit fullscreen mode

Podemos imaginar:

P1 ──┐     ┌──► P1'
     ├─────┤
P2 ──┤     ├──► P2'
     ├─────┤
P3 ──┘     └──► P3'
Enter fullscreen mode Exit fullscreen mode

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

Isso geralmente significa:

Wide Transformation
        ↓
     Shuffle
        ↓
Nova fronteira de Stage
Enter fullscreen mode Exit fullscreen mode

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
Enter fullscreen mode Exit fullscreen mode

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
Enter fullscreen mode Exit fullscreen mode

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
Enter fullscreen mode Exit fullscreen mode

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()
Enter fullscreen mode Exit fullscreen mode

Outro:

dados = df.toPandas()
Enter fullscreen mode Exit fullscreen mode

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 ─────┘
Enter fullscreen mode Exit fullscreen mode

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
Enter fullscreen mode Exit fullscreen mode

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
Enter fullscreen mode Exit fullscreen mode

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
Enter fullscreen mode Exit fullscreen mode

Enquanto isso, do ponto de vista lógico:

ACTION
   │
   ▼
 JOB
   │
   ├───────────────┐
   ▼               ▼
STAGE 1         STAGE 2
   │               │
 Tasks           Tasks
Enter fullscreen mode Exit fullscreen mode

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?
Enter fullscreen mode Exit fullscreen mode

OutOfMemory

Em vez de apenas aumentar memória:

OutOfMemory
    │
    ├── Driver?
    │
    └── Executor?
Enter fullscreen mode Exit fullscreen mode

Se for Executor:

Executor OOM
    │
    ├── partição muito grande?
    ├── data skew?
    ├── cache excessivo?
    ├── operação cara?
    └── overhead?
Enter fullscreen mode Exit fullscreen mode

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
Enter fullscreen mode Exit fullscreen mode

para um ecossistema no qual trabalhamos frequentemente com:

SparkSession
     +
 DataFrames
     +
 Spark SQL
Enter fullscreen mode Exit fullscreen mode

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()
Enter fullscreen mode Exit fullscreen mode

devemos começar a imaginar:

Data
 ↓
Partitions
 ↓
Tasks
 ↓
Executors
 ↓
Shuffle
 ↓
New Stage
 ↓
Tasks
 ↓
Result
Enter fullscreen mode Exit fullscreen mode

É 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
Enter fullscreen mode Exit fullscreen mode

← 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

Top comments (0)