Eai Galera,
Agora que já terminei o livro e está pendente para a revisão técnica, vou iniciar a minha nova empreitada em DataOps, aqui na alemanha eu fui contratado para ser um engenheiro de DevOps em um time de Dados, o que eles chama de DataOps, basicamente a gente faz as mesmas coisas, mas ao invés de fazer o deploy de aplicações são de modelos de Machine Learning, Virtualização de Dados, automatizar governança em um Data Lake e coisas do tipo.
Atualmente eu trabalho com todas as ferramentas da Azure, Azure Data Lake, Azure DevOps, Azure Data Factory, mas como sabemos sempre existem alternativas open source para isso.
Então aqui vou mostrar como podemos subir um Data Lake utilizando como base o Vagrant, que seria o equivalente a nossa plataforma cloud, o HDFS que seria o equivalente ao Azure Data Lake e o Airflow que entraria no lugar do Azure DataFactory.
Vamos lá.
Estou considerando que você já tem o Vagrant e o VirtualBox instalado, caso você não tenha, siga esse post primeiro https://alissonmachado.com.br/vagrant-ambiente-de-desenvolvimento-agil/ .
Crie uma pasta chamada DataOps, onde vamos criar os arquivos:
PS C:\Users\1511 MXTI> mkdir DataOps
Directory: C:\Users\1511 MXTI
Mode LastWriteTime Length Name
---- ------------- ------ ----
d----- 8/3/2020 6:19 PM DataOps
PS C:\Users\1511 MXTI> cd .\DataOps\
PS C:\Users\1511 MXTI\DataOps>
Dentro dela crie o arquivo do Vagrant.
PS C:\Users\1511 MXTI\DataOps> vagrant init
A `Vagrantfile` has been placed in this directory. You are now
ready to `vagrant up` your first virtual environment! Please read
the comments in the Vagrantfile as well as documentation on
`vagrantup.com` for more information on using Vagrant.
PS C:\Users\1511 MXTI\DataOps>
Substitúa o conteudo do arquivo por este abaixo:
# -*- mode: ruby -*-
# vi: set ft=ruby :
Vagrant.configure("2") do |config|
config.vm.box = "ubuntu/bionic64"
config.vm.define "datalake" do |datalake|
datalake.vm.hostname = "datalake"
datalake.vm.network "private_network", ip: "192.168.33.100"
datalake.vm.provider "virtualbox" do |vb|
vb.memory = "4096"
end
datalake.vm.provision "shell", inline: <<-SHELL
echo "export HADOOP_HOME=/opt/hadoop" >> /etc/profile
echo "export JAVA_HOME=/usr/lib/jvm/java-1.8.0-openjdk-amd64/" >> /etc/profile
echo "export HDFS_NAMENODE_USER=root" >> /etc/profile
echo "export HDFS_DATANODE_USER=root" >> /etc/profile
echo "export HDFS_SECONDARYNAMENODE_USER=root" >> /etc/profile
echo "export YARN_RESOURCEMANAGER_USER=root" >> /etc/profile
echo "export YARN_NODEMANAGER_USER=root" >> /etc/profile
apt clean
apt update
apt install python-pip -y
apt install -y openjdk-8-jre
apt install openjdk-8-jdk-headless -y
apt install -y git ant gcc g++ libffi-dev libkrb5-dev libmysqlclient-dev \
libsasl2-dev libsasl2-modules-gssapi-mit libsqlite3-dev libssl-dev \
libxml2-dev libxslt-dev make maven libldap2-dev python-dev python-setuptools libgmp3-dev
cd /opt
wget http://ftp.unicamp.br/pub/apache/hadoop/common/stable/hadoop-3.2.1.tar.gz
tar -xf hadoop-3.2.1.tar.gz
mv -v hadoop-3.2.1 /opt/hadoop
echo "export JAVA_HOME=/usr/lib/jvm/java-1.8.0-openjdk-amd64/" >> /opt/hadoop/etc/hadoop/hadoop-env.sh
cp /vagrant/core-site.xml /opt/hadoop/etc/hadoop/ -v
cp /vagrant/hdfs-site.xml /opt/hadoop/etc/hadoop/ -v
source /etc/profile
yes | /opt/hadoop/bin/hdfs namenode -format
yes | ssh-keygen -q -t rsa -N '' -f ~/.ssh/id_rsa
cat /root/.ssh/id_rsa.pub >> /root/.ssh/authorized_keys
/opt/hadoop/sbin/start-dfs.sh
SHELL
end
config.vm.define "airflow" do |airflow|
airflow.vm.hostname = "airflow"
airflow.vm.network "private_network", ip: "192.168.33.111"
airflow.vm.provider "virtualbox" do |vb|
vb.memory = "2048"
end
airflow.vm.provision "shell", inline: <<-SHELL
apt clean
apt update
apt install python3 python3-pip -y
python3 -m pip install pyspark
python3 -m pip install apache-airflow[postgres,google]==1.10.10 \
--constraint https://raw.githubusercontent.com/apache/airflow/1.10.10/requirements/requirements-python3.7.txt
echo "PATH=$PATH:~/.local/bin" >> /etc/bash.bashrc
apt install openjdk-8-jre-headless -y
python3 -m pip install hdfs
SHELL
end
end
No Vagrantfile que eu criei, vamos subir duas VMs, uma chamada Datalake e a outra chamada Airflow:
Datalake: 192.168.33.100
Ariflow: 192.168.33.111
Agora dentro do mesmo diretório também crie alguns arquivos que serão necessários para fazer o provisionamento do HDFS.
Arquivo: core-site.xml
fs.defaultFS
hdfs://192.168.33.100:9000
dfs.data.dir
/
Arquivo: hdfs-site.xml
dfs.namenode.rpc-bind-host
0.0.0.0
dfs.namenode.servicerpc-bind-host
0.0.0.0
dfs.namenode.http-bind-host
0.0.0.0
dfs.datanode.use.datanode.hostname
false
dfs.namenode.https-bind-host
0.0.0.0
dfs.replication
1
dfs.permissions
false
Agora para o Airflow, vamos precisar criar um arquivo chamado dataops.py
from datetime import timedelta
from airflow import DAG
from airflow.operators.bash_operator import BashOperator
from airflow.utils.dates import days_ago
default_args = {
'owner': 'airflow',
'depends_on_past': False,
'start_date': days_ago(2),
'email': ['alisson.copyleft@gmail.com'],
'email_on_failure': False,
'email_on_retry': False,
'retries': 1,
'retry_delay': timedelta(minutes=5),
}
dag = DAG(
'dataops',
default_args=default_args,
description='Retrieving apache logs',
schedule_interval=timedelta(days=1),
)
landing_folder = BashOperator(
task_id='landing_folder',
bash_command='rm -rf /tmp/$(date +%Y%m%d) ; mkdir /tmp/$( date +%Y%m%d )',
dag=dag,
)
copying_logs = BashOperator(
task_id='copying_logs',
bash_command='scp -i /root/chave.pem alisson@alissonmachado.com.br:/var/log/apache2/* /tmp/$(date +%Y%m%d)/ ',
dag=dag,
)
uncompress = BashOperator(
task_id='uncompress',
depends_on_past=False,
bash_command='cd /tmp/$(date +%Y%m%d) ; ls -1 *.gz | xargs -i gunzip {}',
retries=3,
dag=dag,
)
send_to_datalake = BashOperator(
task_id='send_to_datalake',
depends_on_past=False,
bash_command='/opt/hadoop/bin/hdfs dfs -fs hdfs://192.168.33.100:9000/ -put /tmp/$(date +%Y%m%d)/access* /raw',
retries=3,
dag=dag,
)
landing_folder >> copying_logs >> uncompress >> send_to_datalake
Isso teoricamente é o necessário para começarmos a trabalhar com o DataOps.
Agora que você já tem todos os pré requisitos, vamos subir o ambiente utilizando o comando vagrant up --provision.
PS C:\Users\1511 MXTI\DataOps> vagrant up --provision
Bringing machine 'datalake' up with 'virtualbox' provider...
Bringing machine 'airflow' up with 'virtualbox' provider...
==> datalake: Importing base box 'ubuntu/bionic64'...
Aguarde até que o provisionamento das duas VMs termine, você pode usar o comando vagrant status pra saber se elas estão no ar.
PS C:\Users\1511 MXTI\DataOps> vagrant status
Current machine states:
datalake running (virtualbox)
airflow running (virtualbox)
Se você acessar o endereço pelo navegador: http://192.168.33.100:9870/
Isso mostra que o seu Hadoop já está em execução.
Agora vamos criar um diretório dentro do Data Lake para armazenar os arquivos em sua forma original, normalmente chamados esses diretórios dentro do Data Lake de Layers.
Então eu vou crair uma Layer chamada Raw, o que seria o arquivo cru, sem nenhum tipo de transformaçào.
Acesse a máquina via SSH.
PS C:\Users\1511 MXTI\DataOps> vagrant ssh datalake
Welcome to Ubuntu 18.04.3 LTS (GNU/Linux 4.15.0-55-generic x86_64)
* Documentation: https://help.ubuntu.com
* Management: https://landscape.canonical.com
* Support: https://ubuntu.com/advantage
System information as of Mon Aug 3 19:11:48 UTC 2020
System load: 0.0 Processes: 98
Usage of /: 31.4% of 9.63GB Users logged in: 0
Memory usage: 24% IP address for enp0s3: 10.0.2.15
Swap usage: 0% IP address for enp0s8: 192.168.33.100
139 packages can be updated.
83 updates are security updates.
*** System restart required ***
vagrant@datalake:~$ sudo su -
root@datalake:~#
E crie a Raw layer.
root@datalake:~# /opt/hadoop/bin/hdfs dfs -mkdir /raw
root@datalake:~# /opt/hadoop/bin/hdfs dfs -ls /
Found 1 items
drwxr-xr-x - root supergroup 0 2020-08-03 19:12 /raw
root@datalake:~#
Agora que isso já está criado.
Vamos alterar de máquina e acessar o airflow.
PS C:\Users\1511 MXTI\DataOps> vagrant ssh airflow
Welcome to Ubuntu 18.04.3 LTS (GNU/Linux 4.15.0-55-generic x86_64)
* Documentation: https://help.ubuntu.com
* Management: https://landscape.canonical.com
* Support: https://ubuntu.com/advantage
System information as of Mon Aug 3 19:13:28 UTC 2020
System load: 0.08 Processes: 93
Usage of /: 25.1% of 9.63GB Users logged in: 0
Memory usage: 10% IP address for enp0s3: 10.0.2.15
Swap usage: 0% IP address for enp0s8: 192.168.33.111
149 packages can be updated.
93 updates are security updates.
*** System restart required ***
vagrant@airflow:~$ sudo su -
root@airflow:~#
Dentro dela também vamos precisar do client do HDFS para poder enviar os arquivos para lá, então vamos baixar is binários com os seguintes comandos:
root@airflow:/opt# cd /opt/
root@airflow:/opt# wget http://apache.mirror.iphh.net/hadoop/common/hadoop-3.3.0/hadoop-3.3.0.tar.gz
--2020-08-03 19:15:59-- http://apache.mirror.iphh.net/hadoop/common/hadoop-3.3.0/hadoop-3.3.0.tar.gz
Resolving apache.mirror.iphh.net (apache.mirror.iphh.net)... 62.201.161.83, 2001:868:0:182::3
Connecting to apache.mirror.iphh.net (apache.mirror.iphh.net)|62.201.161.83|:80... connected.
HTTP request sent, awaiting response... 200 OK
Length: 500749234 (478M) [application/x-gzip]
Saving to: ‘hadoop-3.3.0.tar.gz’
Quando você terminar o download, descompacte no diretório /opt/hadoop.
root@airflow:/opt# tar -xf hadoop-3.3.0.tar.gz
root@airflow:/opt# mv hadoop-3.3.0 hadoop
root@airflow:/opt# ls hadoop
LICENSE-binary LICENSE.txt NOTICE-binary NOTICE.txt README.txt bin etc include lib libexec licenses-binary sbin share
root@airflow:/opt#
Agora vamos testar a conexão entre a máquina do airflow com o hdfs na máquina datalake.
root@airflow:/opt# export JAVA_HOME=/usr/lib/jvm/java-1.8.0-openjdk-amd64/
root@airflow:/opt# /opt/hadoop/bin/hdfs dfs -ls hdfs://192.168.33.100:9000/
Found 1 items
drwxr-xr-x - root supergroup 0 2020-08-03 19:12 hdfs://192.168.33.100:9000/raw
root@airflow:/opt#
Vamos tentar fazer um upload para ver se funciona.
root@airflow:/opt# /opt/hadoop/bin/hdfs dfs -put /etc/vim/vimrc hdfs://192.168.33.100:9000/raw/
root@airflow:/opt# /opt/hadoop/bin/hdfs dfs -ls hdfs://192.168.33.100:9000/raw
Found 1 items
-rw-r--r-- 3 root supergroup 2469 2020-08-03 19:32 hdfs://192.168.33.100:9000/raw/vimrc
Perfeito, então temos a comunicação entre as duas máquinas funcionando, tudo instalado, vamos para a parte do airflow.
Dentro do diretório /vagrant na máquina airflow, você encontrará o arquivo dataops.py que foi criado no início desse post.
Esse arquivo nada mais é do que uma Pipeline, como as que temos no Jenkins, porém usando uma ferramenta diferente.
Então o que essa pipeline faz?
Ela pega os arquivos de log do apache do servidor do meu blog, salva na máquina local, descompacta esses arquivos e manda para o hadoop para que possam ser utilizados posteriormente.
Hoje eu trabalho em conjunto com os times de engenharia de dados, inclusive eu também atuo como engenheiro de dados, então nós pegamos os dados de qualquer fonte que seja, transformamos o dado em formato de tabela, na maioria dos casos em parquet e colocamnos em um diretório no Data Lake para que os cientistas de dados possam executar as queries neles.
Explicando os passos da pipeline:
1 - Landing Folder: Essa etapa tem esse nome, pois é o diretório temporário onde os arquivos que serão recebidos devem ser trabalhados antes de serem mandados para o Data Lake.
2 - Copying Logs: Essa etapa é basicamente o download dos arquivos dentro do Landing Folder, criado na etapa anterior.
3 - Uncompress: Sabe-se que por boas práticas logs antigos são compactados no servidor para ocupar menos espaço, então aqui eu faço o descompactamento dos arquivos para poder manda-los em texto plano.
4 - Send To Datalake: Essa é a etapa onde executo os comandos do hadoop para enviar os dados para o diretório Raw, pois como os dados não foram transformados nem nada, precisamos terminar essa pipeline logo e depois acionar uma segunda para trabalhar esse dado, pegando por exemplo todos os logs e agregando em um unico arquivo no formato parquet e compactando no formato snappy.
Note que no final do código do airflow, temos essa linha:
landing_folder >> copying_logs >> uncompress >> send_to_datalake
Que indica a sequencia em que os passos devem ser executados.
Então vamos criar o banco de dados iniciado do airflow:
root@airflow:/opt# airflow initdb
Ao usar o airflow, o diretório home é sempre o ~/airflow, então dentro dele precisamos criar um diretório chamado dags ( Directed Acyclic Graph ) e copiar a nossa pipeline pra dentro dele.
root@airflow:~/airflow# mkdir dags
root@airflow:~/airflow# cd dags/
root@airflow:~/airflow/dags# cp /vagrant/dataops.py .
Para criar a pipeline no airflow execute o comando:
root@airflow:~/airflow/dags# python3 dataops.py
E pronto, a pipeline já foi criada.
Para ver a interface do airflow precisamos executar o seguinte comando:
root@airflow:/opt# airflow webserver -D
____________ _____________
____ |__( )_________ __/__ /________ __
____ /| |_ /__ ___/_ /_ __ /_ __ \_ | /| / /
___ ___ | / _ / _ __/ _ / / /_/ /_ |/ |/ /
_/_/ |_/_/ /_/ /_/ /_/ \____/____/|__/
[2020-08-03 19:43:54,111] {__init__.py:51} INFO - Using executor SequentialExecutor
[2020-08-03 19:43:54,112] {dagbag.py:396} INFO - Filling up the DagBag from /root/airflow/dags
Running the Gunicorn Server with:
Workers: 4 sync
Host: 0.0.0.0:8080
Timeout: 120
Logfiles: - -
=================================================================
Execute também o scheduler:
root@airflow:/opt# airflow scheduler
____________ _____________
____ |__( )_________ __/__ /________ __
____ /| |_ /__ ___/_ /_ __ /_ __ \_ | /| / /
___ ___ | / _ / _ __/ _ / / /_/ /_ |/ |/ /
_/_/ |_/_/ /_/ /_/ /_/ \____/____/|__/
[2020-08-03 19:45:48,470] {__init__.py:51} INFO - Using executor SequentialExecutor
[2020-08-03 19:45:48,484] {scheduler_job.py:1346} INFO - Starting the scheduler
[2020-08-03 19:45:48,484] {scheduler_job.py:1354} INFO - Running execute loop for -1 seconds
[2020-08-03 19:45:48,485] {scheduler_job.py:1355} INFO - Processing each file at most -1 times
[2020-08-03 19:45:48,485] {scheduler_job.py:1358} INFO - Searching for files in /root/airflow/dags
[2020-08-03 19:45:48,489] {scheduler_job.py:1360} INFO - There are 24 files in /root/airflow/dags
[2020-08-03 19:45:48,489] {scheduler_job.py:1411} INFO - Resetting orphaned tasks for active dag runs
[2020-08-03 19:45:48,541] {dag_processing.py:556} INFO - Launched DagFileProcessorManager with pid: 12056
[2020-08-03 19:45:48,545] {settings.py:54} INFO - Configured default timezone
[2020-08-03 19:45:48,556] {dag_processing.py:758} WARNING - Because we cannot use more than 1 thread (max_threads = 2) when using sqlite. So we set parallelism to 1.
Na interface web você já poderá ver a sua pipeline criada:
http://192.168.33.111:8080/admin/
Veja que a pipeline cahamada dataops já está lá criada.
Na coluna links, o primeiro botão se chama Trigger Dag, clique nele para executar a sua pipeline.
E veja a executação dela na tela.
Note que todas as etapas declaradas no script estão ai presentes.
Como esse é o meu primeiro post sobre DataOps, vou para por aqui, pois é só uma introdução do que ainda está por vir.
Obrigado galera =)