Conceitualmente Machine Learning em larga escala é quando processamos mais de 100 bilhões de registros. Para esse processamento ocorrer em tempo satisfatório se faz necessário o uso de cluster de computadores, onde cada máquina vai ficar a cargo do processamento de parte desses registros. Para operacionalizar esse processamento usamos algum software de processamento de dados distribuído, como Hadoop MapReduce, Spark ou Storm. Todos podem trabalhar em cluster e cada um tem seu caso de uso mais indicado. Em nosso exemplo, vamos fazer o uso do Spark em um cluster de três máquinas para processamento do famoso big data.
Escalabilidade é a capacidade de um sistema crescer facilmente para atender uma demanda maior de processamento. Devemos implementar um sistema para execução ao longo do tempo, onde as cargas de trabalho provavelmente vão aumentar. Esse sistema precisa ter a capacidade de facilmente aceitar mais ou menos servidores na sua infraestrutura e utilizá-los para suprir a necessidade de processamento adicional.
A escalabilidade pode ser horizontal ou vertical. Na horizontal fazemos a adição de máquinas/nós na infraestrutura da aplicação. Requer que a aplicação esteja desenhada para uso de diversas máquinas. Já na vertical adicionamos mais recursos (memória, CPU, espaço em disco, etc) na máquina atual da aplicação.
Em nosso exemplo vamos trabalhar com escalabilidade horizontal. O Spark consegue lidar muito bem com a adição ou remoção de nós do cluster onde ele está em execução.
Uma aplicação em larga escala tem as seguintes características:
- Distribuída: uma aplicação que é executada em mais de uma máquina (aplicação distribuída) pode ser considerada de larga escala;
- Escalável: o ambiente da aplicação consegue se adequar a necessidade de processamento requerida no momento;
- Alta performance: é esperado uma boa performance de aplicações de larga escala. Isso é garantido pela execução distribuída e escalável que vimos a pouco;
- Alta disponibilidade: é esperado que a aplicação esteja a maior parte do tempo online, principalmente nos momentos onde a aplicação é vital para a execução de algum processo de negócio;
- Arquitetura moderna: arquitetura que facilite tudo que foi citado, de preferencia microsserviços.
Apache Spark é uma plataforma de computação em cluster (conjunto de computadores), criado para ser veloz e de uso geral, sendo ideal para processamento iterativo (trabalho que precisa ser repetido diversas vezes), interativo (podemos executar comandos e obter feedbacks em tempo real) e processamento de streaming de dados (dados gerados de forma continua). Foi construído com a intenção de ser veloz, tanto no processamento de queries, quanto de algoritmos. Além de processamento em memória e eficiente recuperação de falhas.
Spark é um framework para processamento de dados de forma distribuída, em larga escala. Ele não é responsável por armazenamento de dados, isso é feito pelo Hadoop HDFS, por exemplo.
O Spark faz o processamento dos dados em memória, ao contrário do Hadoop MapReduce que faz o processamento em disco. Assim o processamento do Spark é bem mais performático.
Atentar que se houverem mais dados que memória disponível no cluster de computadores, usar o Hadoop MapReduce pode ser mais interessante.
Vamos montar um multinode cluster Spark com três máquinas, onde teremos um nó master e dois nós slaves. Vamos utilizar como sistema operacional dessas máquinas o CentOS, que é uma distribuição Linux de classe corporativa derivada de códigos fonte gratuitamente distribuídos pela Red Hat Enterprise Linux (RHEL).
Vamos utilizar virtualização local dessas máquinas, com uso do VirtualBox como software de virtualização. Poderíamos criá-las também utilizando algum serviço de nuvem, como AWS ou GCP.
Vamos montar nossas máquinas virtuais (VM) utilizando o VirtualBox. Elas serão os nós do nosso cluster Spark. Nesse exemplo estou usando a importação de um appliance para ganhar tempo na montagem das VMs. Abaixo os detalhes. Nossas máquinas vão se chamar “spark_cluster_server1”, “spark_cluster_server2” e “spark_cluster_server3”.
Importante que as VMs tenham comunicação entre si, que estejam em rede. Como configuramos no VirtualBox para as máquinas usarem placa de rede em modo “Bridge”, isso já fica resolvido.
Por padrão o usuário que criamos na instalação, nesse caso “rodrigo”, não faz parte do grupo “sudoers” e assim não pode usar o comando “sudo” do CentOS. Resolvemos isso editando o arquivo “/etc/sudoers” inserindo esse nosso usuário. Para isso, entramos com o usuário “su”, que é o root, inserimos a senha dele e nesse momento nosso usuário tem permissão de “super user”.
Para não precisar ficar entrando com os dados do “su”, vamos editar o arquivo citado e inserir nosso usuário. Com o comando abaixo abrimos o arquivo “sudoers”.
gedit /etc/sudoers
Inserimos nosso usuário abaixo do “root” e salvamos o arquivo. Pronto. Com isso conseguimos executar o comando “sudo” direto do nosso usuário, sem necessidade de acessar pelo usuário “su”.
Para facilitar a vida, vamos configurar o arquivo “hosts” (/etc/hosts) de cada VM para conter o IP e o nome host das outras VMs. Assim podemos interagir entre as máquinas usando o nome host em vez do IP.
sudo gedit /etc/hosts
Arquivo hosts da VM “server1”.
Arquivo hosts da VM “server2”.
Arquivo hosts da VM “server3”.
Também precisamos do Java JDK instalado e configurado em nossas VMs para execução do Spark. Vamos baixar o JDK “jdk-8u181-linux-x64.tar.gz” disponível nesse link (necessário autenticar no site da Oracle). Precisamos do JDK porque o Spark foi construído com a linguagem Scala e essa linguagem executa em uma JVM, e para termos uma JVM, precisamos do JDK do Java.
Após realizar o download, descompactamos com o comando abaixo no terminal.
tar -xvf jdk-8u181-linux-x64.tar.gz
Agora movemos o conteúdo que acabamos de descompactar para o diretório “/opt/jdk”.
sudo mv jdk1.8.0_181/ /opt/jdk
Agora configuramos as variáveis de ambiente. Para isso editamos o arquivo “.bash_profile” da home do nosso usuário e colocamos no final do arquivo o diretório do JDK e JRE e o endereço do executável de ambos.
gedit .bash_profile
Recarregamos as variáveis de ambiente com o comando “source .bash_profile” e testamos se o Java está funcionando verificando a versão do mesmo.
Caso apareça o “Java Open JDK”, remova o mesmo com os comandos abaixo, Após isso o Java que instalamos será o exibido ao executar o comando de versão do Java.
yum list java*
sudo yum -y remove java*
java -version
A comunicação entre as máquinas de um cluster Spark é via protocolo SSH e não faz sentido usar SSH com senha nessa comunicação. Vamos configurar nosso ambiente para que as máquinas possam conversar entre si por SSH sem a necessidade de informar usuário e senha do SSH.
Por padrão o SSH já vem instalado no CentOS. Basta configura-lo para não pedir a senha nas comunicações entre nós. Para isso vamos criar as chaves públicas e privadas, que são utilizadas na comunicação por SSH. Em nosso caso a chave privada vai ficar no nó master e a publica nos nós slaves.
Para criar as chaves usamos o comando abaixo no terminal no nosso nó master.
ssh-keygen -t rsa -P ""
Chave gerada.
Conferindo os arquivos com as chaves. O primeiro arquivo é o com a chave privada. Já o segundo é o com a chave pública.
Vamos configurar a a chave pública no master primeiro. Com o comando abaixo executado no nó master, autorizamos o próprio nó se conectar nele sem uso de senha via SSH. O “authorized_keys” é um arquivo com a relação de chaves autorizadas a conectar nessa máquina.
cat id_rsa.pub > authorized_keys
Note que estamos dentro de “/home/rodrigo/.ssh” em nosso nó master.
Agora vamos levar a chave pública do master para os nós slaves. Vamos fazer essa cópia via SSH mesmo, nesse caso passando a senha de acesso aos nós de destino. Abaixo exemplo da cópia a partir do server1 para o server2 (fazer o mesmo para o server3).
ssh-copy-id -i id_rsa.pub rodrigo@server2
Chave já no destino com o diretório “.ssh” criado automaticamente no server2.
Pronto. Veja que estamos no server1, mas acessando o terminal do server2 e não pediu senha para o acesso ao server2.
Não é necessário configurar o SSH sem senha para os nós slaves, pois eles não precisam de tal característica em um cluster com Spark.
Vamos fazer o download dos arquivos de instalação do Spark. Usaremos essa versão do Spark. Link para download.
wget https://downloads.apache.org/spark/spark-2.4.7/spark-2.4.7-bin-hadoop2.7.tgz
Após realizar o download, descompactamos com o comando abaixo no terminal.
tar -xvf spark-2.4.7-bin-hadoop2.7.tgz
Agora movemos o conteúdo que acabamos de descompactar para o diretório “/opt/spark”.
sudo mv spark-2.4.7-bin-hadoop2.7/ /opt/spark
Agora configuramos as variáveis de ambiente. Para isso editamos o arquivo “.bash_profile” da home do nosso usuário e colocamos no final do arquivo o diretório do JDK e JRE e o endereço do executável de ambos.
gedit .bash_profile
Recarregamos as variáveis de ambiente com o comando “source .bash_profile” e testamos se o Spark está funcionando chamando o shell dele com o comando “spark-shell”. Agora fazemos o mesmo para as demais máquinas do cluster.
Na última etapa que fizemos o Spark está funcionando e executando em uma máquina apenas (standalone). Ele executa com sucesso nas três máquinas, mas elas ainda não estão interligadas, é como se cada máquina fosse um cluster de uma máquina só.
Agora vamos configurar o Spark para executar em modo cluster real, com as três máquinas trabalhando em conjunto (cluster gerenciado).
Vamos para o diretório de configuração do Spark “/opt/spark/conf” em nosso nó master, o “server1”.
Nele temos diversos arquivos de template das configurações do Spark. Vamos fazer uma cópia de dois desses arquivos de template para editarmos na sequência.
cp spark-env.sh.template spark-env.sh
cp slaves.template slaves
Agora editamos o arquivo “slaves” informando quem são os nós slaves, que são os nós workes do nosso cluster Spark.
Agora editamos o arquivo “spark-env.sh” informando a máquina que é o nó master e o diretório do Java JDK.
Fazemos o mesmo processo dos arquivos de configuração do Spark nos demais nós do cluster.
Agora no nó master acessamos o diretório “/opt/spark/sbin” e executamos o comando “./start-all.sh”. Isso inicializa o cluster Spark. Esse comando só deve ser executado no nó que é o master do cluster.
cd /opt/spark/sbin
./start-all.sh
No nó master devemos ter os seguintes serviços executando.
Já nos nós slaves devemos ter os seguintes serviços executando:
Pronto. Cluster Spark em execução e pronto para uso.
O “spark-shell” é uma espécie de cliente para o Spark, mas só aceita comandos/programas em linguagem Scala. O cliente para aceitar comandos/programas em Python é o “pyspark”, que é uma interface de programação Python para o Spark. O Python que acompanha o Spark é ainda a 2.7 e vai ser descontinuada em breve.
Vamos atualizar para o Python 3, utilizando a distribuição Python “Anaconda Python”, que já traz pacotes adicionais que vão nos ajudar mais a frente. Para baixar o instalador, usamos o comando abaixo.
wget https://repo.anaconda.com/archive/Anaconda3-5.3.0-Linux-x86_64.sh
Para realizar a instalação, usamos o comando abaixo.
sudo yum install bzip2
bash Anaconda3-5.3.0-Linux-x86_64.sh
Aceitamos os termos de uso, instalamos no diretório sugerido pelo próprio instalador e aceitamos a atualização do arquivo “.bashrc”.
Recarregamos as variáveis de ambiente.
source .bash_profile
Pronto. PySpark operacional com Python 3 em nosso master node.
Agora temos um terminal com tudo que o Python e o Spark podem oferecer. Toda vez que abrimos o PySpark, é criado um “Spark Context”, apelidado de “sc”, que permite a conexão do nosso terminal (cliente) com o cluster Spark.
Em nosso exemplo, “numeros” é uma lista normal Python. Já “numeros_distribuidos” é um RDD Spark, ou seja, uma lista/coleção paralelizada. Essa RDD pode ser processada de forma distribuída no cluster, que é o objetivo do uso do cluster Spark.
Agora criamos uma função que retorna os números menores que 5 e na sequência executamos ela, fazendo uma soma dos valores retornados. Note que usamos o RDD criado a pouco para chamar as funcionalidades do Spark.
Resumindo: Usamos uma linguagem da nossa escolha (Python, Java, Scala ou R) para prepararmos os dados, criando algum objeto que represente esses dados, ainda nessa linguagem. Após isso, paralelizamos esse objeto no Spark e a partir daí podemos executar as operações, fornecidas pelo Spark, em cima desses dados de forma distribuída.
Concluímos assim essa primeira etapa onde vimos a montagem passo a passo de um cluster Spark multinode utilizando máquinas virtuais com sistema operacional CentOS. Também exploramos um pouco sobre o que é esse framework e alguns detalhes sobre o que é escalabilidade e larga escala para o universo de aplicações. No próximo artigo vamos explorar um pouco mais o Spark, conhecendo como é seu funcionamento.