Airflow auto-hébergé : révolution dans la collecte de données

Airflow et l'auto-hébergement : une nouvelle ère pour la collecte de données financières

L'auto-hébergement d'Airflow à domicile ouvre de nouvelles perspectives pour les passionnés de technologie cherchant à maîtriser les opérations de machine learning (MLOps). L'objectif principal est de comprendre comment déployer des flux de travail en intelligence artificielle (IA) dans un environnement réel. Les MLOps consistent à transférer des modèles d'IA de la phase de recherche à la production, tout en automatisant les processus et en gérant les erreurs de manière efficace.

Dans mes précédents articles, j'ai décrit la mise en place d'un serveur PostgreSQL et d'un serveur Airflow. Ces serveurs agissent comme des bases de données essentielles pour collecter les ensembles de données nécessaires à l'entraînement des modèles d'IA. La prochaine étape consiste à remplir ces bases de données PostgreSQL avec des données pertinentes. Airflow joue ici un rôle crucial en orchestrant les pipelines de données, garantissant que les informations les plus récentes sont intégrées dans notre base de données. Cette étape est fondamentale dans le processus de machine learning, car elle fournit les données indispensables à l'entraînement des modèles.

L'une des motivations derrière la création de mon laboratoire à domicile est de démontrer que l'ensemble du processus de machine learning peut être auto-hébergé à l'aide de machines virtuelles et de conteneurs. Étant donné mes récents investissements personnels, j'ai choisi de travailler avec des données financières. Ces données sont précieuses pour analyser les tendances, corréler les prix et même tenter des prévisions, ce qui les rend extrêmement utiles dans divers scénarios.

Configuration d'Airflow

Exécution de votre serveur Airflow

Pour optimiser l'utilisation d'Airflow, j'ai découvert et mis en œuvre plusieurs astuces. Lors de la configuration initiale de mon serveur Airflow, j'ai utilisé les commandes suivantes pour lancer les différents composants :

• nohup airflow scheduler > scheduler.log 2>&1 &
• nohup airflow dag-processor > dag-processor.log 2>&1 &
• nohup airflow triggerer > triggerer.log 2>&1 &
• nohup airflow api-server --port 8080 > api-server.log 2>&1 &

La commande nohup permet de maintenir un processus actif même après la fermeture du terminal. Cependant, j'ai rencontré des problèmes de plantage des composants d'Airflow, nécessitant un redémarrage manuel. De plus, le démarrage de chaque composant demandait quatre commandes distinctes, ce qui était fastidieux.

Pour simplifier ce processus, j'ai créé un script pour démarrer ou redémarrer Airflow plus facilement.

nano airflow_restart.sh

Dans le fichier airflow_restart.sh, j'ai inséré le code suivant :

#!/bin/bash
pkill -f "airflow" --ignore-ancestors
sleep 2
echo "Démarrage du planificateur..."
nohup airflow scheduler > scheduler.log 2>&1 &
echo "Démarrage du processeur DAG..."
nohup airflow dag-processor > dag-processor.log 2>&1 &
echo "Démarrage du déclencheur..."
nohup airflow triggerer > triggerer.log 2>&1 &
echo "Démarrage du serveur API..."
nohup airflow api-server --port 8080 > api-server.log 2>&1 &
echo "Airflow redémarré"

Ce script permet de tuer tous les processus liés à "airflow" et de redémarrer les quatre composants. Ainsi, au lieu d'exécuter quatre commandes distinctes, il suffit désormais de lancer ce script.

Pour résoudre le problème des plantages fréquents, j'ai configuré Airflow en tant que processus daemon système. Un daemon est un processus en arrière-plan qui démarre automatiquement au démarrage du système, redémarre en cas de défaillance et reste indépendant du terminal.

Pour cela, j'ai créé un fichier de service :

nano /etc/systemd/system/airflow-scheduler.service

Ensuite, j'ai inséré le texte suivant :

[Unit]
Description=Airflow Scheduler
After=network.target postgresql.service
Wants=postgresql.service

[Service]
User=<USER>
Group=<GROUP>
Environment=PATH=<AIRFLOW_PATH>:/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin
Environment=AIRFLOW_HOME=<AIRFLOW_FOLDER>
ExecStart=<AIRFLOW_PATH> scheduler
Restart=on-failure
RestartSec=5s
StandardOutput=journal
StandardError=journal

[Install]
WantedBy=multi-user.target

J'ai remplacé <USER>, <GROUP>, <AIRFLOW_PATH> et <AIRFLOW_FOLDER> par les valeurs spécifiques à ma configuration. J'ai répété ce processus pour les autres composants d'Airflow : dag-processor, triggerer, api-server.

Après avoir enregistré les fichiers, j'ai exécuté la commande suivante pour recharger les nouveaux processus daemon :

sudo systemctl daemon-reload

Puis, j'ai activé et démarré tous les services :

# Activer les services au démarrage
sudo systemctl enable airflow-scheduler
sudo systemctl enable airflow-dag-processor
sudo systemctl enable airflow-triggerer
sudo systemctl enable airflow-api-server

# Démarrer les services
sudo systemctl start airflow-scheduler
sudo systemctl start airflow-dag-processor
sudo systemctl start airflow-triggerer
sudo systemctl start airflow-api-server

Désormais, Airflow fonctionne de manière plus stable sur mon serveur. Vous pouvez utiliser systemctl status pour vérifier si chaque composant fonctionne et journalctl pour voir les informations de journalisation.

Ajout de votre connexion PostgreSQL

Une étape cruciale est la configuration de la connexion PostgreSQL sur le serveur Airflow pour permettre une communication fluide avec la base de données PostgreSQL. Airflow offre une connexion PostgreSQL pratique qui permet de définir les paramètres de connexion une seule fois, facilitant ainsi leur réutilisation.

Pour configurer cela, j'ai modifié le fichier airflow.cfg :

nano airflow.cfg

J'ai cherché l'entrée test_connection et l'ai définie sur Enabled. Ensuite, j'ai redémarré Airflow, soit via le script, soit en redémarrant les processus daemon. Cela permet de tester les connexions via l'interface utilisateur d'Airflow.

Dans l'interface utilisateur d'Airflow, j'ai ajouté la connexion PostgreSQL en naviguant vers Admin > Connections. J'ai sélectionné Add Connection en haut à droite. Sous Connection Type, j'ai choisi Postgres. Si l'option Postgres n'apparaît pas, il peut être nécessaire d'exécuter pip install apache-airflow-providers-postgres sur le serveur Airflow.

J'ai ensuite renseigné les détails de connexion : PostgreSQL Host (adresse IP), Login, Password, Port et Database. Une fois la connexion ajoutée, j'ai testé sa validité en cliquant sur l'icône de graphique, qui s'est transformée en icône de wi-fi. Un message vert est apparu, confirmant le succès du test de connexion.

Création du code du pipeline Airflow

Avec Airflow correctement configuré, il est temps de développer le code pour le pipeline Airflow. Airflow fonctionne à travers des graphes acycliques dirigés (DAGs), qui structurent les flux de travail en plusieurs tâches de traitement de données.

Dans un DAG, les tâches peuvent être exécutées dans un ordre spécifique et avec des dépendances, permettant d'exécuter certaines tâches après l'achèvement d'autres, ou en parallèle.

Pour obtenir des données financières, j'ai utilisé yfinance, un package Python qui exploite les API de Yahoo! Finance pour récupérer des données de marché. Ce package est idéal pour extraire et écrire des données de prix de ticker d'actions dans notre base de données PostgreSQL.