Comment écrire et distribuer des fournisseurs personnalisés dans Airflow ?

Lors de la création d'un fournisseur personnalisé pour Airflow, une organisation claire et une structuration cohérente du code sont essentielles pour garantir non seulement une bonne intégration avec le système, mais aussi pour faciliter sa distribution et sa maintenance. Voici une vue d'ensemble de la façon dont vous pouvez structurer et développer votre propre fournisseur Airflow en suivant des conventions bien établies.

Les modules peuvent inclure les éléments suivants :

• Hooks : Ce dossier est dédié à l’écriture des hooks pour le service avec lequel vous vous intégrez. Ces hooks permettent d’établir des connexions avec des services externes et d’en gérer les interactions.

• Operators : C’est ici que vous implémentez les opérateurs qui seront utilisés dans vos DAGs. Les opérateurs définissent les tâches à exécuter dans Airflow.

• Sensors : Ces opérateurs spécialisés sont conçus pour interagir avec des services externes, attendre des événements spécifiques ou vérifier des conditions avant d’exécuter des tâches.

• Triggers : Utilisés par des opérateurs différables, ces déclencheurs gèrent les événements qui doivent se produire avant qu’une tâche ne commence.

Dans cet exemple, un fichier module "provider" est également inclus, avec une fonction permettant d’enregistrer votre fournisseur auprès d’Airflow lorsque celui-ci est installé.

Le dossier dev contient des fichiers utilisés pour la création d’un environnement de test et de démonstration localisé. Il peut comprendre un fichier Dockerfile ainsi qu’un fichier docker-compose.yaml pour décrire cet environnement.

Le dossier Example DAGs sert à stocker des exemples de DAGs qui démontrent le fonctionnement du fournisseur dans votre environnement de développement. Ces exemples sont cruciaux pour permettre aux utilisateurs de voir concrètement comment intégrer et utiliser votre fournisseur.

Pour emballer votre fournisseur et le rendre distribué, deux fichiers principaux sont nécessaires : setup.py et setup.cfg. Le fichier setup.cfg contient les métadonnées de votre package, ce qui permet aux outils de distribution Python de comprendre comment empaqueter et installer le fournisseur. Le fichier setup.py est optionnel, mais il peut être utile lorsque vous souhaitez installer votre package en mode "édition".

Bien que la structure que nous venons de décrire ne soit pas strictement obligatoire pour que votre fournisseur fonctionne dans Airflow, elle suit des conventions largement adoptées dans la communauté. Ces standards existent depuis suffisamment longtemps pour être considérés comme des bonnes pratiques, ce qui facilite la compréhension et l’intégration par d’autres utilisateurs d’Airflow.

Une fois la structure du projet en place, vous pouvez commencer à modifier vos fichiers de configuration et votre code pour vous assurer que votre fournisseur s’intègre bien à Airflow. Il est crucial de garder en tête que le code doit pouvoir fonctionner sans connexion Internet. Le planificateur d'Airflow (Scheduler) analyse les DAGs à intervalles réguliers, et pendant cette analyse, tout ce qui se trouve dans la méthode __init__ d'une classe sera exécuté, y compris les appels aux API externes. Cette restriction doit être respectée pour éviter des erreurs d'importation et garantir un fonctionnement fluide.

Lors de la création d’opérateurs, la méthode execute est impérativement requise. C’est cette méthode qui définit le cœur de l’opération à réaliser pour chaque tâche dans un DAG. Pour que votre fournisseur fonctionne correctement avec Airflow, vous devrez aussi gérer son enregistrement au démarrage du système via un point d'entrée dans la configuration du package. En utilisant un fichier provider.py, vous pouvez définir une fonction get_provider_info qui fournira des informations à Airflow pour l'affichage de détails dans l'interface utilisateur, la ligne de commande et l'API d’Airflow.

La création du hook pour notre théière intelligente

Pour intégrer cette API dans Airflow, nous devons écrire un "hook" permettant de gérer la connexion et les échanges avec la théière. Cela commence par la sous-classe d’un hook de base d’Airflow. Ce hook contiendra des méthodes pour définir et gérer les connexions au service et facilitera les interactions avec l'API.

python
Copy code
class TeaPotHook(BaseHook):

    conn_name_attr = "tea_pot_conn_id"
    default_conn_name = "tea_pot_default"
    conn_type = "teapot"
    hook_name = "TeaPot"
    def __init__(self, tea_pot_conn_id: str = default_conn_name) -> None:
        super().__init__()
        self.tea_pot_conn_id = tea_pot_conn_id

Par exemple, avec la méthode get_connection_form_widgets, vous pouvez spécifier les champs à remplir pour chaque connexion de théière. Cette méthode permet de personnaliser l'interface d'Airflow pour collecter les informations nécessaires au fonctionnement du hook, comme le designateur de la théière ou les ajouts à la préparation du thé.

Comment concevoir un fournisseur personnalisé dans Airflow ?

La création d'un fournisseur personnalisé dans Apache Airflow implique plusieurs étapes clés permettant de définir des connexions, des opérations et des événements spécifiques à votre environnement. Cela permet d'intégrer facilement des services externes tout en garantissant une bonne gestion des ressources et une logique métier propre et évolutive. L'un des exemples les plus simples et révélateurs d'une telle intégration est l'utilisation de la notion de « théière » (teapot) dans un système de gestion de connexions et d'opérations.

Tout d'abord, il est essentiel de définir deux nouveaux champs dans votre modèle de connexion — le « pot_designator » et « additions ». Ces champs seront stockés comme des clés dans un dictionnaire extra, qui est lié à chaque objet de connexion. La méthode statique suivante montre comment définir le comportement de l'interface utilisateur (UI) pour ces champs, de manière à simplifier l'entrée des utilisateurs dans l'interface graphique d'Airflow :

python
Copy code
@staticmethod
def get_ui_field_behaviour():
    return {

        "hidden_fields": ["password", "login", "schema", "extra"],

        "placeholders": {
            "pot_designator": "1",
            "additions": "sugar",
            "host": "tea-pot",
            "port": "8083",
        },
        "relabeling": {}
    }

Cette méthode permet de modifier les éléments de l'interface utilisateur existants : les champs listés dans hidden_fields seront cachés, et des exemples de valeurs (comme "1" pour le désignateur de la théière ou "sugar" pour les ajouts) seront affichés dans les champs correspondants pour guider l'utilisateur. Bien que le champ relabeling soit nécessaire, il est déconseillé de l'utiliser fréquemment, car cela nuit à la lisibilité et à la compréhension du code par les autres développeurs.

Une fois cette configuration définie, nous devons créer un accès aux connexions. Cela peut se faire par la méthode suivante, qui permet de récupérer les informations de connexion à partir d'une base de données via un ID spécifique, et d'affecter ces informations aux attributs de la classe correspondante :

python
Copy code
@cached_property

def get_conn(self):
    conn = self.get_connection(self.tea_pot_conn_id)
    self.url = f"{conn.host}:{conn.port}"
    self.pot_designator = conn.extra_dejson.get("pot_designator", None)
    self.additions = conn.extra_dejson.get("additions", None)
    return self

Cela permet de construire dynamiquement une URL complète basée sur l'hôte et le port, tout en récupérant les valeurs spécifiques à la théière et aux ajouts, si elles sont définies. Cette approche vous assure une intégration fluide des valeurs de configuration dans votre classe.

python
Copy code
def test_connection(self):
    """Test a connection"""
    if self.is_ready():
        return (True, "Alive")
    return (False, "Not Alive")

Cela garantit que la connexion est correctement établie avant toute interaction plus poussée avec le service externe.

python
Copy code
"connection-types": [
    {
        "connection-type": "teapot",
        "hook-class-name": "airflow_provider_tea_pot.hooks.TeaPotHook"
    }
]

Ensuite, pour interagir avec le service de théière, nous pouvons définir des méthodes pour chaque point de terminaison exposé, comme la préparation du thé ou la vérification du niveau d'eau. Un exemple simple de ces méthodes est le suivant, où chaque méthode appelle le service externe via une requête HTTP et traite la réponse :

python
Copy code
def make_tea(self):

    self.get_conn()
    response = requests.get(f"http://{self.url}/ready")
    return self.__check_and_return(response)
def get_water_level(self):
    self.get_conn()
    response = requests.get(f"http://{self.url}/ready")
    return self.__check_and_return(response)

L'approche utilisée ici est très basique, avec une vérification minimale de la réponse HTTP. Dans un cas réel, il serait préférable d'ajouter des mécanismes de gestion des erreurs plus robustes et des pratiques défensives pour garantir une haute disponibilité et une résilience face aux pannes.

Une fois les « hooks » mis en place, nous pouvons créer des opérateurs pour effectuer des actions concrètes. Par exemple, un opérateur de préparation de thé peut ressembler à ceci :

python
Copy code
class MakeTeaOperator(BaseOperator):

    def __init__(self, tea_pot_conn_id, additions=None, pot_designator=None, **kwargs) -> None:
        super().__init__(**kwargs)
        self.tea_pot_conn_id = tea_pot_conn_id
        self.pot_designator = pot_designator
        self.additions = additions
    def execute(self, context) -> typing.Any:
        self.hook = TeaPotHook(tea_pot_conn_id=self.tea_pot_conn_id)
        if self.pot_designator:
            self.hook.pot_designator = self.pot_designator
        if self.additions:
            self.hook.additions = self.additions
        return self.hook.make_tea()

En parallèle, les capteurs (ou « sensors ») jouent un rôle essentiel en surveillant des ressources externes jusqu'à ce qu'une condition particulière soit remplie, déclenchant alors une action. Cela permet de gérer des flux de travail basés sur des événements externes.

Dans les scénarios plus complexes, vous devrez également envisager la possibilité d'utiliser des opérateurs différables (Deferrable Operators), qui permettent à l'opérateur de suspendre son exécution jusqu'à ce que des conditions spécifiques soient remplies, libérant ainsi les ressources du planificateur pendant ce temps.

La création de ces composants dans Airflow vous permettra non seulement d'intégrer des services externes mais aussi de garantir une gestion optimale des ressources, tout en permettant une extensibilité future de vos processus de travail.

Comment optimiser le flux de travail de l’apprentissage automatique avec Airflow et Kubernetes

Le premier défi consiste à préparer et pré-traiter les données. Par exemple, dans le cas de la recommandation de films, le fichier CSV contenant les informations sur les utilisateurs et les films doit être converti en un jeu de données tabulaire, avec des représentations vectorielles adaptées pour l'algorithme de filtrage collaboratif. Cette étape nécessite des outils comme PythonOperator dans Airflow pour orchestrer la génération des données sous forme de DataFrames. Une fois les données traitées, elles sont stockées dans un espace objet (comme Amazon S3), et des valeurs XCom sont poussées pour informer les étapes en aval sur l’emplacement des fichiers et la structure des données.

Une fois la phase de pré-traitement achevée, le pipeline passe à la création des caractéristiques pour le modèle de filtrage collaboratif. Traditionnellement, un modèle KNN (K-nearest neighbors) serait formé à l’aide de bibliothèques comme Scikit-learn. Cependant, dans ce contexte, les vecteurs bruts sont directement insérés dans une base de données vectorielle, comme une extension Postgres vector, permettant des requêtes directes pour trouver des correspondances similaires sans devoir re-former constamment le modèle. L’utilisation d’extensions vectorielles dans une base de données relationnelle optimise non seulement la gestion des données mais améliore également les performances des requêtes de recommandation.

Cette gestion des données est suivie d’une autre étape importante : la formation du modèle d’apprentissage profond. Bien que cette étape puisse nécessiter une infrastructure de calcul spécialisée, la combinaison d'Airflow et de Kubernetes offre une solution scalable et flexible. En déléguant l'exécution de l’entraînement du modèle à un cluster Kubernetes via KubernetesPodOperator, on peut profiter d'une capacité de calcul dynamique, choisissant les nœuds optimisés pour le traitement lourd des données.

Lorsque le script d'entraînement est conçu, il doit être entièrement indépendant de l'infrastructure sous-jacente. Les paramètres de configuration sont extraits des variables d'environnement ou des commandes Docker, garantissant une gestion sécurisée des secrets. Une fois l’entraînement du modèle terminé, les résultats peuvent être facilement renvoyés vers Airflow à l'aide d’un fichier JSON compatible avec XCom. Cela permet de rendre le modèle disponible pour d’autres étapes du pipeline sans redondance.

Enfin, l’optimisation de ce flux de travail est possible grâce à la modularité d'Airflow et la gestion efficace des ressources avec Kubernetes. L'approche consiste à séparer les différentes étapes du pipeline (comme le pré-traitement des données, la création des vecteurs, et l'entraînement du modèle) en tâches indépendantes, ce qui permet non seulement de tester chaque étape individuellement, mais aussi d’assurer une plus grande fiabilité et une scalabilité du système dans son ensemble.

Le succès d’un tel système repose non seulement sur la configuration adéquate des différents composants, mais aussi sur une compréhension approfondie de l'architecture distribuée et de la gestion des tâches asynchrones. L’utilisation de Kubernetes permet, par exemple, de tirer parti de ressources spécifiques comme les processeurs graphiques ou d’autres matériels spécialisés pour l’entraînement des modèles d’apprentissage profond, ce qui représente un avantage décisif lorsque l’échelle du projet augmente.

En somme, la combinaison d'Airflow, des bases de données vectorielles et de Kubernetes représente une solution moderne et performante pour la gestion et l’orchestration des flux de travail dans l'apprentissage automatique. La mise en place de telles architectures exige cependant une attention particulière à l'optimisation des ressources, à la gestion des erreurs et à la sécurité des données tout au long du pipeline.

Quelle structure de dépôt choisir pour un déploiement Airflow optimal ?

La structure de dépôt est une décision qui semble triviale, mais qui a un impact considérable sur le succès d’un projet. Avant de prendre cette décision, il est essentiel de bien comprendre comment votre équipe fonctionne actuellement, quels types de modèles opérationnels vous souhaitez soutenir et comment vos équipes interagiront avec la structure choisie. Prenez le temps de réfléchir à la manière dont votre équipe va travailler et comment elle interagira avec cette structure avant de finaliser vos choix. Il est important de garder à l’esprit que vous pouvez toujours changer d’avis.

Le Mono-repo

Cependant, ce modèle présente des inconvénients notables. Tout d’abord, la taille du dépôt devient un problème, entraînant des temps de téléchargement plus longs. Ensuite, la coordination des releases et des promotions devient plus complexe et nécessite davantage de gestion. Enfin, les pipelines CI/CD peuvent devenir plus complexes à maintenir, notamment lorsqu’on déploie selon une architecture microservices.

Le Multi-repo

Le modèle multi-repo, quant à lui, divise l’ensemble du déploiement Airflow en plusieurs dépôts. Dans sa forme la plus extrême, chaque élément du système pourrait avoir son propre dépôt : un dépôt pour Airflow de base, des dépôts pour chaque fournisseur ou plugin utilisé, et des dépôts pour chaque équipe déployant des DAGs. Ce modèle est intéressant si vous voulez que les équipes puissent développer et déployer des parties spécifiques du système de manière indépendante, avec des workflows simples et isolés.

Mono-repo ou multi-repo ? Cette question est souvent un faux dilemme. Ces deux modèles représentent les deux extrêmes d’un spectre qui décrit différentes possibilités. Dans la réalité, vous finirez probablement par adopter une combinaison des deux. Une solution fréquemment recommandée consiste à utiliser un mono-repo pour la description de votre déploiement Airflow et des fournisseurs internes, tandis que des dépôts séparés sont utilisés pour les DAGs.

La gestion des objets de connexion et des variables

Les objets de connexion et de variable dans Airflow sont essentiels au bon fonctionnement des opérateurs et des DAGs. Ils servent à définir les systèmes avec lesquels Airflow doit interagir et à ajuster le comportement des DAGs en fonction des valeurs définies dans les variables. Ces objets jouent un rôle similaire à des configurations dans d’autres logiciels et sont susceptibles de changer en fonction de l’environnement dans lequel ils sont déployés. Par ailleurs, ces objets peuvent contenir des informations sensibles, telles que des identifiants de connexion à des systèmes externes.

Dans de nombreux cas, les développeurs créent des objets de connexion et de variables via l'interface Web d'Airflow. Bien que cette méthode soit parfaitement adaptée à des exemples de démonstration, elle souffre des limitations typiques des opérations manuelles, à savoir l'absence de méthode cohérente pour gérer les objets de configuration à travers plusieurs environnements. Heureusement, Airflow propose des mécanismes plus adaptés aux pratiques DevOps modernes.

Les variables d’environnement

Les objets de connexion et de variable peuvent être définis en tant que variables d’environnement dans Airflow. Par exemple, les variables peuvent être définies comme suit :

bash
Copy code
export AIRFLOW_VAR_NAME=1
export AIRFLOW_VAR_NAME_JSON='{"hello":"world"}'
export AIRFLOW_CONN_MY_DATABASE='conn-type://login:password@host:port/schema?param1=val1¶m2=val2'

Airflow enregistre ces variables et les rend accessibles dans les DAGs. Toutefois, cette méthode présente des risques de sécurité si les variables sont stockées sous forme de chaînes de texte brutes, ce qui nécessite souvent l’utilisation d’un système de gestion des secrets. Ce système permet de stocker les variables de manière cryptographiquement sécurisée, et de les décrypter lors du déploiement, ou de les injecter à partir d'un service externe au démarrage du processus.

Les backends de secrets

Les backends de secrets constituent un autre moyen de gérer les objets de connexion et de variables de manière sécurisée dans Airflow. Grâce à l'adaptateur Secrets Backend, Airflow peut récupérer des secrets depuis des gestionnaires de secrets externes tels que AWS Secrets Manager, Google Cloud Secret Manager, HashiCorp Vault, ou encore Microsoft Azure Key Vault. Cette approche nécessite la configuration adéquate de la section "secrets" dans le fichier de configuration d'Airflow, avec le nom de classe du fournisseur utilisé et les arguments nécessaires pour initialiser ce fournisseur.

Méthodes de déploiement d’Airflow

Le choix de la méthode de déploiement dépend en grande partie des capacités et de l’infrastructure de votre organisation. Voici quelques-unes des méthodes les plus courantes.

Kubernetes

Kubernetes simplifie grandement le processus de construction et de gestion d’applications complexes comme Airflow. En offrant une syntaxe déclarative simple pour décrire l’application et abstrait les besoins auxiliaires liés à la gestion des services interconnectés, Kubernetes permet une gestion efficace de l’orchestration de services. Grâce à l’existence d'un chart Helm bien développé pour Airflow, le déploiement dans un cluster Kubernetes devient une tâche relativement facile. Si vous avez déjà une expérience avec Kubernetes (directement ou via un service fourni par une autre équipe), cette méthode est fortement recommandée.

Machines virtuelles

Si vous préférez travailler sur des machines physiques ou virtuelles, vous pouvez déployer Airflow sur des machines virtuelles (ou sur des machines physiques directement). Cependant, cette méthode est plus complexe et nécessite des outils de gestion de configuration comme Terraform, Chef, Puppet, Ansible ou CloudFormation pour gérer l'approvisionnement et l'opération des infrastructures.

L’une ou l’autre de ces méthodes dépendra des ressources et des connaissances disponibles au sein de votre équipe, ainsi que des budgets alloués pour le déploiement et la gestion de l’infrastructure.