Nesta postagem, gostaria de apresentar mais uma abordagem para ajudar na compreensão de conceitos relacionados ao Apache Spark. Muito do que é apresentado aqui encontra-se em livros, blogs (como este), cursos pagos e vídeo-aulas no Youtube. Apenas compilei parte deste material e adicionei algumas coisas que julgo importantes com base na minha experiência. Este texto foi dividido em três grandes seções.

Na primeira seção, “Uma visão geral dos componentes do Apache Spark”, abordo de forma ampla os componentes básicos do Apache Spark e suas respectivas responsabilidades, bem como descrevo a anatomia de uma aplicação do Apache Spark. Na segunda seção, trato dos conceitos de “Ações, transformações e avaliação preguiçosa”, que são fundamentais para o entendimento da execução de uma aplicação. Por fim, concluo com um resumo dos pontos apresentados nas sessões anteriores.

Uma visão geral dos componentes básicos do Apache Spark

A execução de uma aplicação do Apache Spark envolve um programa Driver, alguns executores e um gerenciador de recursos. Na documentação oficial do Apache Spark temos a seguinte imagem:

Arquitetura de uma aplicação Spark

A figura acima representa uma aplicação Spark em modo cluster. Neste modo, cada componente é alocado em um servidor (nó ou “node”) separado. O processo Driver é alocado no driver node e os executores são alocados nos worker nodes. Tasks são processadas nos worker nodes. O Gerenciador de Recursos se comunica com o driver para prover recursos para os executores. O gerenciador de recurso, em ambientes de produção, é um cluster por si só. No momento em que este texto foi escrito, clusteres Apache Mesos, Apache Hadoop YARN e Kubernetes podem ser utilizados como Gerenciador de Recursos do Spark.

Também é possível fazer o deploy do processo driver em um nó fora do cluster. Nestes casos, dizemos que o deploy da aplicação foi realizado em modo client.

Por hora, vamos adaptar a imagem acima para uma versão simplificada, omitindo os detalhes dos nós workers e que também pode ser generalizada para outros modos de deploy.

Arquitetura simplificada de uma aplicação Spark

Desta forma, trataremos dos seguintes componentes:

  • Processo Driver.
  • Gerenciador de recursos.
  • Executores.

Driver

O processo driver de uma aplicação Spark tem as seguintes responsabilidades:

  • Requisitar recursos físicos ao gerenciador de recursos, como memória e CPU, para executar a aplicação Spark.
  • Manter as informações sobre o estado da aplicação em execução.
  • Interagir com o código do usuário.
  • Analisar, distribuir e agendar a execução da aplicação através dos executores.

Toda interação entre o código do usuário e os demais recursos é realizada através da SparkSession de uma determinada aplicação. Em casos onde o usuário deseja especificar, por exemplo, a quantidade de executores para uma aplicação, essa especificação se dá através da mudança de parâmetros da SparkSession.

Apesar de ser o componente que gerencia a execução de uma aplicação, o programa Driver geralmente não demanda muitos recursos. De fato, muitas das aplicações são executadas em clusteres com driver nodes menos potentes que os worker nodes.

Em ações no Spark, como por exemplo o .collect()os dados são coletados dos executores pelo driver, envolvendo o mesmo no processamento dos dados. Nesses casos, é possível que o processo driver demande mais recursos que os executores, havendo então a necessidade de disponibilizar uma máquina mais poderosa para alocá-lo.

Gerenciador de Recursos

O Gerenciador de Recursos possui as seguintes responsabilidades:

  • Fornece recursos físicos ou virtuais, como CPU e Memória, conforme requisitado pelo Driver.
  • Reserva os recursos sendo usados por uma aplicação.
  • Libera os recursos de uma aplicação após sua finalização.
  • Substitui recursos, sejam eles nós ou processos falhos, para garantir a continuidade da aplicação.

O gerenciador de recursos pode ser um processo Spark em cenários de teste, quando o framework é executado em modo standalone. Em ambientes produtivos, no entanto, o gerenciador de recursos é um cluster por si só. No momento da escrita deste texto, é possível utilizar clusteres Apache Mesos, Apache Hadoop YARN e Kubernetes como gerenciadores de recursos do Apache Spark.

Na minha experiência, a pessoa que desenvolve a aplicação Apache Spark não tem gerência sobre as configurações do gerenciador de recursos. Por outro lado, é comum que ela monitore algumas métricas, como uso geral de CPU e memória do cluster, para tirar melhor proveito dos recursos disponibilizados. Essas métricas ajudam a entender se o cluster está superdimensionado ou, em alguns casos, se há espaço para uso de recursos como autoscaling no Databricks.

Executores

Os executores são os componentes que de fato processam a workload de uma aplicação conforme requisitado pelo Driver, Estes são, na minha experiência, o componente mais importante para se olhar quando desejamos buscar por oportunidades de otimização.

Quando uma aplicação Spark é iniciada, o processo driver a divide visando otimizar o processamento em paralelo:

  • Trabalho (“job”): uma sequência de etapas criada toda a vez que uma ação é executada.
  • Etapa (“stage”): uma sequência de transformações estreitas (“narrow transformations”) executadas em um job. Uma etapa é delimitada por transformações amplas (“wide transformations”).
  • Tarefa (“task”): é a menor unidade de trabalho que pode ser agendada para realizar o processamento dos dados. Uma tarefa age sobre um subconjunto dos dados que chamamos de partição. Um conjunto de tarefas compõe uma transformação.

Na imagem abaixo, ilustro uma aplicação Spark que possui duas ações e dois jobs:

Composição de uma aplicação Apache Spark.

Na imagem acima, as tarefas 1.1 e 1.2 na Stage 1 do Job 1 podem ser executadas ao mesmo tempo, em paralelo, caso haja recursos disponíveis[. O mesmo pode acontecer para as tarefas 2.1 a 2.4 na Stage 2 do Job 2. O que náo acontece, por exemplo, é de tarefas em Stages diferentes em um mesmo Job serem executadas ao mesmo tempo.

Ter clareza sobre os conceitos até agora apresentados pode ajudar na busca por oportunidades de otimização da aplicação. Por exemplo, o alto número de tarefas pode ser um gargalo na execução da aplicação caso o Spark precise acessar muitas partições espalhadas pelo cluster em ordem aleatória. Por outro lado, uma aplicação com poucas partições grandes, pode sofrer gargado devido a falta de memória no executor. Vamos agora tratar dos conceitos de ações, transformações e avaliação preguiçosa

Ações, transformações e avaliação preguiçosa

Os termos do título desta seção são muito usados em quaisquer textos que tratam sobre Spark. Neste texto, não poderia ser diferente.

Existem apenas duas grandes famílias de operações no Spark: as transformações e as ações. Transformações são operações que agem sobre um conjunto de dados e geram um novo, sem modificar o primeiro conjunto. Este detalhe é muito importante: em Spark, os conjuntos de dados são imutáveis. Transformações não transformam um conjunto de dados, mas os consome para criar um novo. Transformações, por sua vez, só acontecem quando uma ação as solicita.

Ações são operações que disparam a execução de uma série de transformações em vários conjuntos de dados e retornam um dado para o processo driver ou persiste em um sistema de armazenamento. No Spark, as transformações são executadas apenas quando demandado uma por ação. Esse é o conceito de avaliação preguiçosa (“lazy evaluation”).

Para tentar ilustrar melhor a mensagem que quero passar, farei uso de pseudo-códigos e alguns diagramas para descrever uma situação hipotética.

Considere que temos um conjunto de dados 1, onde existem losangos, círculos e quadrados. Ao final do processo, queremos um conjunto de dados 2 que possui apenas losangos e quadrados. Como parâmetro inicial, temos o caminho para os arquivos que compõem o primeiro conjunto de dados.

Alguém sem familiaridade com o conceito de avaliação preguiçosa pode escrever o seguinte:

conjunto_de_dados_1 = ler_arquivos(caminho_para_o_arquivo)
conjunto_de_dados_2 = remover_circulos(conjunto_de_dados_1)
  • Ilustrando o raciocínio quando se desconsidera a avaliaçao preguiçosa

Contudo, o código possui apenas transformações, representados pelas funções ler_arquivos e remover_circulos, não existindo nenhuma ação invocando as transformações. Sendo assim, nenhum processamento aconteceu.

Para gerar o conjunto de dados 2, devemos ter uma ação em algum momento, como no seguinte exemplo:

conjunto_de_dados_1 = ler_arquivos(caminho_para_o_arquivo)
conjunto_de_dados_2 = remover_circulos(conjunto_de_dados_1)
conjunto_de_dados_2.collect()

Neste cenário, ocorre o seguinte:

  • Um parâmetro caminho_para_o_arquivo é passado para a função ler_arquivos. Em algum momento, o Spark gera as instruções de baixo nível para que o executor acesse esses arquivos em disco e crie o objeto conjunto_de_dados_1 em memória.
  • Um parâmetro conjunto_de_dados_1 é passado para a função remover_circulos. Em algum momento, o Spark gera as instruções de baixo nível para que o executor acesse os dados armazenados em memória do objeto conjunto_de_dados_1, remova os círculos, e crie o objeto conjunto_de_dados_2 em memória.
  • O método collect do objeto conjunto_de_dados_2, que é uma ação, é executado. Ele solicita a execução do melhor plano de processamento onde todas as transformações são executadas, gerando o conjunto_de_dados_2 para coleta.

Ilustrando o raciocínio quando se considera a avaliaçao preguiçosa.

Transformações estreitas e transformações amplas

Do conceito geral do que é uma transformação, podemos apresentar os conceitos de transformações estreitas (“narrow transformations”) e transformações amplas (“wide transformations”). A diferença entre ambas se dá na necessidade de recombinar dados existentes em executores diferentes para criar uma nova partição.

Uma transformação estreita consome partições existentes dentro de um executor para criar uma nova dentro do mesmo executor. Neste caso, não há a necessidade de dados serem transferidos pela rede. Transformações estreitas fazem uso de uma operação conhecida como “pipelining”, onde as instruções são escritas em uma fila na memória do nó e executadas uma após a outra neste mesmo nó.

Na imagem abaixo, cada partição é submetida a uma mesma transformação, produzindo uma nova partição. As etapas de transformação sobre elas ocorrem em um nó específico do cluster.

Transformação estreita.

Já em transformações amplas, dados em partições existentes em diferentes executores precisam ser recombinados para criar novas partições. Isso leva a situações em que o número de partições na entrada de uma transformação é menor, maior ou mesmo igual ao número de partições de saída. Em transformações com groupBy, partições podem ser agregadas em uma só. Transformações envolvendo join, por outro lado, podem partir de 2 partições e produzir 3. Ainda podemos ter um orderBy que apenas ordena os dados de três partições, resultando num mesmo volume ao fim do processo.

Na imagem abaixo, ilustramos o resultado de uma possível agregação, onde duas partições são combinadas em uma só nova partição. Isso é o que acontece em operações onde a cláusula groupBy é utilizada:

Transformação ampla: agregação.

Na imagem abaixo, ilustramos uma situação em que dados existentes em 2 partições são recombinados e geram 3 partições. Isso é algo que pode acontecer em alguma operação envolvendo join.

Transformação ampla: join.

Por último, ilustramos o resultado de um possível rearranjo onde a quantidade de partições na entrada e na saída são iguais, porém, os dados dessas partições são recombinados entre si para a criação das novas partições. Esse tipo de situação pode acontecer em uma operação envolvendo orderBy.

Transformação ampla: orderby.

Conclusão

É importante ter clareza a respeito de como o Spark divide uma aplicação e como os componentes desta ferramenta interagem entre si. Tratamos dos componentes driver, gerenciador de recursos e executor. Para tratar de como uma aplicação é dividida, e executada precisamos entender os conceitos de transformações, ações e avaliação preguiçosa.

Vimos que o processo driver é o componente do Spark que realiza a interface entre o código da aplicação, o gerenciador de recursos, e os executores. O driver interage com o código da aplicação e se comunica tanto com o gerenciador de recursos quanto com os executores, mas geralmente não é um componente que interage com os dados da workload da aplicação. Por isso, é comum o processo driver ser alocado em um nó menos potente do que os nós que alocam o processo executor.

O gerenciador de recursos é o componente responsável por comunicar ao driver quais os recursos, como CPU e memória, disponíveis para criar os executores nos worker nodes. O gerenciador de recursos também é responsável por reservar ou liberar os recursos solicitados pelo driver, Além disso, é o gerenciador de recursos que substitui executores que eventualmente falharem durante a execução de uma aplicação, ou até mesmo substituir o worker node caso este apresente um problema. O gerenciador de recurso, em ambientes produtivos, éum cluster inteiro (Mesos, YARN ou Kubernetes). Já em ambientes de teste, quando o Spark é disponibilizado em modo standalone, o gerenciador de recursos é apenas mais um processo criado pelo Spark

Os executores, por sua vez, são responsáveis pelo processamento da workload da aplicação. Em geral, estes processos são alocados em máquinas mais potentes que o processo driver. Os executores recebem jobs do driver que, por sua vez, são divididos em stages, onde as transformações geram um determinado número de tarefas, cada uma agindo em apenas uma partição.

Transformações ocorrem em processamento por avaliação preguiçosa, ou seja, são executadas apenas mediante uma ação. Uma transformação pode ser considerada estreita ou ampla. Já nas transformações amplas, esta necessidade existe. É comum em transformações estreitas que o número de partições de entrada ser igual ao número de partições de saída, já o mesmo não é verdade para o caso de transformações amplas.

Referências