Ao preparar nossos programas educacionais, ocasionalmente enfrentamos dificuldades ao trabalhar com algumas ferramentas. E na época em que nos deparamos com essas dificuldades, nem sempre há documentação e artigos suficientes que ajudem a resolver o problema.
Isso aconteceu, por exemplo, em 2015, quando no programa "Especialista em Big Data" utilizamos um cluster Hadoop com Spark para 35 usuários simultâneos. Como prepará-lo para esse caso de uso com o uso do YARN, era incerto. No final, depois de entender e passar pelo processo por conta própria, criamos e também apresentamos no .
Antecedentes
Desta vez, falaremos sobre outro programa – . Nele, nossos participantes constroem dois tipos de arquitetura: lambda e kappa. E na arquitetura lambda, no âmbito do processamento em lote, usamos Airflow para transferir logs do HDFS para o ClickHouse.
Tudo está, em geral, bem. Que eles construam seus pipelines. Mas há um porém: todos os nossos programas são tecnológicos do ponto de vista do processo de ensino. Para verificar os labs, usamos checadores automáticos: o participante precisa acessar seu painel, clicar no botão “Verificar”, e após algum tempo ele verá um feedback aprofundado sobre o que fez. E é exatamente nesse momento que começamos a nos deparar com nosso problema.
A verificação desse lab é organizada da seguinte forma: enviamos um pacote de dados de controle para o Kafka do participante, depois o Gobblin transfere esse pacote de dados para o HDFS, em seguida, o Airflow pega esse pacote de dados e o coloca no ClickHouse. A questão é que o Airflow não deve fazer isso em tempo real, ele faz isso com uma programação: a cada 15 minutos, pega um lote de arquivos e os envia.
Portanto, precisamos de alguma forma para acionar seu DAG manualmente, sob nosso requisito, durante o funcionamento do verificador, aqui e agora. Após pesquisar, descobrimos que para versões mais recentes do Airflow existe a chamada . A palavra experimental, claro, soa assustadora, mas o que fazer... Talvez de certo.
A seguir, descreveremos todo o caminho: desde a instalação do Airflow até a formação de um POST request que aciona o DAG, utilizando a API Experimental. Trabalharemos com Ubuntu 16.04.
1. Instalação do Airflow
Vamos verificar se temos Python 3 e virtualenv instalados.
$ python3 --version
Python 3.6.6
$ virtualenv --version
15.2.0Se algo disso não estiver instalado, faça a instalação.
Agora, criamos um diretório onde iremos trabalhar com o Airflow.
$ mkdir
$ cd /path/to/seu/novo/diretório
$ virtualenv -p which python3 venv
$ source venv/bin/activate
(venv) $Vamos instalar o Airflow:
(venv) $ pip install airflowA versão que usamos foi: 1.10.
Agora precisamos criar um diretório airflow_home, onde estarão os arquivos DAG e plugins do Airflow. Após criar o diretório, configuraremos a variável de ambiente AIRFLOW_HOME.
(venv) $ cd /path/to/my/airflow/workspace
(venv) $ mkdir airflow_home
(venv) $ export AIRFLOW_HOME=O próximo passo é executar o comando que criará e inicializará o banco de dados do fluxo de dados no SQLite:
(venv) $ airflow initdbO banco de dados será criado em airflow.db padrão.
Vamos verificar se o Airflow foi instalado:
$ airflow version
[2018-11-26 19:38:19,607] {__init__.py:57} INFO - Usando executor SequentialExecutor
[2018-11-26 19:38:19,745] {driver.py:123} INFO - Gerando tabelas de gramática a partir de /usr/lib/python3.6/lib2to3/Grammar.txt
[2018-11-26 19:38:19,771] {driver.py:123} INFO - Gerando tabelas de gramática a partir de /usr/lib/python3.6/lib2to3/PatternGrammar.txt
____________ _____________
____ |__( )_________ __/__ /________ __
____ /| |_ /__ ___/_ /_ __ /_ __ _ | /| /
___ ___ | / _ / _ __/ _ / / \/_\/ |_ |/ |/
_/_/_ |_/_/_/ /_/_/ /_/_/ /_/_/ ____/____/|__/
v1.10.0Se o comando executou corretamente, o Airflow criou seu arquivo de configuração airflow.cfg em AIRFLOW_HOME:
$ tree
.
├── airflow.cfg
└── unittests.cfgO Airflow possui uma interface web. Você pode iniciá-la executando o comando:
(venv) $ airflow webserver --port 8081Agora você pode acessar a interface web no navegador na porta 8081 na máquina em que o Airflow foi iniciado, por exemplo: <hostname:8081>.
2. Trabalhando com o Experimental API
Com isso, o Airflow está configurado e pronto para uso. No entanto, precisamos também iniciar o Experimental API. Nossos verificadores estão escritos em Python, então a partir de agora, todas as requisições serão realizadas com ele usando a biblioteca requests.
Na verdade, a API já está funcionando para requisições simples. Por exemplo, essa requisição permite testar seu funcionamento:
>>> import requests
>>> host =
>>> airflow_port = 8081 # neste caso usamos esse, mas por padrão é 8080
>>> requests.get('http://{}:{} /{}'.format(host, airflow_port, 'api/experimental/test')).text
'OK'Se você recebeu essa mensagem em resposta, isso significa que tudo está funcionando.
No entanto, quando tentarmos acionar o DAG, teremos que enfrentar o fato de que esse tipo de requisição não pode ser feito sem autenticação.
Para isso, será necessário realizar mais algumas etapas.
Primeiro, adicione isso ao seu arquivo de configuração:
[api]
auth_backend = airflow.contrib.auth.backends.password_authEm seguida, você precisará criar seu usuário com privilégios de administrador:
>>> import airflow
>>> from airflow import models, settings
>>> from airflow.contrib.auth.backends.password_auth import PasswordUser
>>> user = PasswordUser(models.Admin())
>>> user.username = 'new_user_name'
>>> user.password = 'set_the_password'
>>> session = settings.Session()
>>> session.add(user)
>>> session.commit()
>>> session.close()
>>> exit()Em seguida, é necessário criar um usuário com permissões comuns, que será permitido acionar o DAG.
>>> import airflow
>>> from airflow import models, settings
>>> from airflow.contrib.auth.backends.password_auth import PasswordUser
>>> user = PasswordUser(models.User())
>>> user.username = 'newprolab'
>>> user.password = 'Newprolab2019!'
>>> session = settings.Session()
>>> session.add(user)
>>> session.commit()
>>> session.close()
>>> exit()Agora tudo está pronto.
3. Iniciando uma solicitação POST
A solicitação POST será assim:
>>> dag_id = newprolab
>>> url = 'http://{}:{} /{} /{} /{}'.format(host, airflow_port, 'api/experimental/dags', dag_id, 'dag_runs')
>>> data = {"conf":"{"key":"value"}"}
>>> headers = {'Content-type': 'application/json'}
>>> auth = ('newprolab', 'Newprolab2019!')
>>> uri = requests.post(url, data=json.dumps(data), headers=headers, auth=auth)
>>> uri.text
'{n "message": "Created "n}n'A solicitação foi processada com sucesso.
Portanto, em seguida, damos algum tempo para o DAG processar e fazemos uma solicitação na tabela ClickHouse, tentando capturar o pacote de dados de controle.
Verificação concluída.
Fonte: habr.com
