🥇 Como acionar um DAG no Airflow usando a API Experimental

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 um post no Habr e também apresentamos no Moscow Spark Meetup.

Antecedentes

Desta vez, falaremos sobre outro programa – Data Engineer. 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 API Experimental. 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.0

Se 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 airflow

A 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 initdb

O 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.0

Se o comando executou corretamente, o Airflow criou seu arquivo de configuração airflow.cfg em AIRFLOW_HOME:

$ tree
.
├── airflow.cfg
└── unittests.cfg

O Airflow possui uma interface web. Você pode iniciá-la executando o comando:

(venv) $ airflow webserver --port 8081

Agora 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_auth

Em 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