Comment créer un déclencheur DAG dans Airflow en utilisant l'API expérimentale

Lors de la préparation de nos programmes éducatifs, nous rencontrons périodiquement des difficultés liées à certains outils. Et au moment où nous sommes confrontés à ces problèmes, il n'y a pas toujours suffisamment de documentation et d'articles pour nous aider à les résoudre.

C'était le cas, par exemple, en 2015, lorsque dans le programme "Spécialiste des Big Data", nous utilisions un cluster Hadoop avec Spark pour 35 utilisateurs simultanés. Comment le préparer pour ce cas d'utilisation avec YARN n'était pas clair. Finalement, après avoir compris et fait le chemin par nous-même, nous avons créé un post sur Habr et nous avons également présenté lors du Moscow Spark Meetup.

Contexte

Cette fois, nous parlerons d'un autre programme – Data Engineer. Nos participants y construisent deux types d'architectures : lambda et kappa. Dans l'architecture lambda, dans le cadre du traitement batch, nous utilisons Airflow pour transférer les logs d'HDFS vers ClickHouse.

Tout va plutôt bien. Qu'ils construisent leurs pipelines. Cependant, il y a un mais : tous nos programmes sont technologiques en ce qui concerne le processus d'apprentissage lui-même. Pour vérifier les labs, nous utilisons des vérificateurs automatiques : le participant doit se connecter à son espace personnel, cliquer sur le bouton "Vérifier", et après un certain temps, il reçoit des retours détaillés sur ce qu'il a fait. Et c'est précisément à ce moment-là que nous commençons à nous approcher de notre problème.

La vérification de ce lab est organisée comme suit : nous envoyons un paquet de données de contrôle dans Kafka du participant, ensuite Gobblin dépose ce paquet de données dans HDFS, puis Airflow prend ce paquet de données et le place dans ClickHouse. L'astuce est qu'Airflow ne doit pas le faire en temps réel, il le fait selon un calendrier : toutes les 15 minutes, il prend un lot de fichiers et les traite.

Il s'avère qu'il nous faut d'une manière ou d'une autre déclencher leur DAG nous-mêmes selon notre demande à ce moment-là lors de l'exécution du vérificateur. En cherchant sur Google, nous avons découvert que pour les versions récentes d'Airflow, il existe ce qu'on appelle un API Expérimental. Le mot expérimental, bien sûr, fait peur, mais que faire... Peut-être que ça marchera.

Nous allons maintenant décrire tout le processus : de l'installation d'Airflow à la formation de la requête POST qui déclenche le DAG en utilisant l'API Expérimentale. Nous travaillerons avec Ubuntu 16.04.

1. Installation d'Airflow

Vérifions que nous avons Python 3 et virtualenv installés.

$ python3 --version
Python 3.6.6
$ virtualenv --version
15.2.0

Si l'un d'eux n'est pas présent, veillez à les installer.

Nous allons maintenant créer un répertoire dans lequel nous allons travailler avec Airflow.

$ mkdir 
$ cd /chemin/vers/votre/nouveau/répertoire
$ virtualenv -p which python3 venv
$ source venv/bin/activate
(venv) $

Installons Airflow :

(venv) $ pip install airflow

Version que nous avons utilisée : 1.10.

Nous devons maintenant créer un répertoire airflow_home, où seront situés les fichiers DAG et les plugins Airflow. Après avoir créé le répertoire, nous définirons la variable d'environnement AIRFLOW_HOME.

(venv) $ cd /chemin/vers/mon/airflow/espace_de_travail
(venv) $ mkdir airflow_home
(venv) $ export AIRFLOW_HOME=

L'étape suivante consiste à exécuter la commande qui créera et initialisera la base de données du flux dans SQLite :

(venv) $ airflow initdb

La base de données sera créée dans airflow.db par défaut.

Vérifions si Airflow est bien installé :

$ airflow version
[2018-11-26 19:38:19,607] {__init__.py:57} INFO - Utilisation de l'exécuteur SequentialExecutor
[2018-11-26 19:38:19,745] {driver.py:123} INFO - Génération des tables de grammaire à partir de /usr/lib/python3.6/lib2to3/Grammar.txt
[2018-11-26 19:38:19,771] {driver.py:123} INFO - Génération des tables de grammaire à partir de /usr/lib/python3.6/lib2to3/PatternGrammar.txt
  ____________       _____________
 ____    |__( )_________  __/__  /________      __
____  /| |_  /__  ___/_  /_ __  /_  __ _ | /| / \\
___  ___ |  / _  /   _  __/ _  / / /_/_/ \_|/|/ \
 _/_/_/  |_/_/_/  _/_/    _/_/    _/_/  ____/____/|__/
   v1.10.0

Si la commande a réussi, Airflow a créé son fichier de configuration airflow.cfg dans AIRFLOW_HOME:

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

Airflow dispose d'une interface web. Vous pouvez la lancer en exécutant la commande :

(venv) $ airflow webserver --port 8081

Vous pouvez maintenant accéder à l'interface web dans votre navigateur à l'adresse http://:8081, par exemple : <hostname:8081>.

2. Travailler avec l'API Expérimentale

À ce stade, Airflow est configuré et prêt à l'emploi. Cependant, nous devons également démarrer l'API Expérimentale. Nos contrôleurs sont écrits en Python, donc toutes les requêtes suivantes seront réalisées en utilisant la bibliothèque requests.

En réalité, l'API fonctionne déjà pour des requêtes simples. Par exemple, cette requête permet de tester son fonctionnement :

>>> import requests
>>> host = 
>>> airflow_port = 8081 # dans notre cas, c'est 8081, par défaut c'est 8080
>>> requests.get('http://{}:{} /{}'.format(host, airflow_port, 'api/experimental/test').text
'OK'

Si vous avez reçu ce message en réponse, cela signifie que tout fonctionne.

Cependant, lorsque nous souhaitons déclencher un DAG, nous constaterons que ce type de requête ne peut être effectué sans authentification.

À cet effet, il sera nécessaire d'effectuer quelques autres actions.

Tout d'abord, ajoutez ceci dans la configuration :

[api]
auth_backend = airflow.contrib.auth.backends.password_auth

Ensuite, vous devez créer un utilisateur avec des droits d'administration :

>>> 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()

Ensuite, il faut créer un utilisateur avec des droits standard, qui sera autorisé à déclencher le 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()

Maintenant tout est prêt.

3. Lancement de la requête POST

La requête POST ressemblera à ceci :

>>> 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": "Créé "n}n'

La requête a été traitée avec succès.

Nous donnons ensuite un certain temps au DAG pour traiter et faisons une requête à la table ClickHouse, en essayant d'attraper le paquet de contrôle de données.

La vérification est terminée.

Source : habr.com