Aller au contenu principal
Dernière mise à jour le

Orchestrer les Lakeflow Jobs avec Apache Airflow

Utilisez Apache Airflow pour orchestrer les pipelines de données avec Databricks via le fournisseur Databricks open source. Installez et configurez Airflow localement, puis déployez et exécutez un Job Databricks à partir d'un DAG Airflow.

remarque

Le support Apache Airflow décrit sur cette page utilise plusieurs packages open source. Cela inclut le fournisseur Databricks pour Airflow (y compris les opérateurs Databricks Airflow). Ces packages ne sont pas directement pris en charge par Databricks. Pour des informations sur le fournisseur Databricks pour Airflow, consultez apache-airflow-providers-databricks sur Apache.org.

Orchestration de Job dans une pipeline de données​

Le développement et le déploiement d'un pipeline de traitement de données nécessitent souvent la gestion de dépendances complexes entre les tâches. Par exemple, un pipeline peut lire des données à partir d'une source, nettoyer les données, transformer les données nettoyées et écrire les données transformées vers une cible. Vous avez également besoin d'un support pour les tests, la planification et le dépannage des erreurs lorsque vous opérationnalisez un pipeline.

Les systèmes de workflow relèvent ces défis en vous permettant de définir les dépendances entre les tâches, de planifier l'exécution des pipelines et de surveiller les workflows. Apache Airflow est une solution open source pour la gestion et la planification des pipelines de données. Airflow représente les pipelines de données sous forme de graphes orientés acycliques (DAG) d'opérations. Vous définissez un flux de travail dans un fichier Python, et Airflow gère la planification et l'exécution. La connexion Airflow Databricks vous permet de tirer parti du moteur Spark optimisé offert par Databricks avec les fonctionnalités de planification d'Airflow.

Exigences​

  • L'intégration entre Airflow et Databricks nécessite Airflow version 2.5.0 et ultérieure. Les exemples de cet article sont testés avec Airflow version 2.6.1.
  • Airflow nécessite Python 3.8, 3.9, 3.10 ou 3.11. Les exemples de cet article sont testés avec Python 3.8.
  • Les instructions de cet article pour installer et exécuter Airflow nécessitent pipenv pour créer un environnement virtuel Python.

Opérateurs Airflow pour Databricks​

Un DAG Airflow est composé de tâches, où chaque tâche exécute un opérateur Airflow. Les opérateurs Airflow prenant en charge l’intégration à Databricks sont implémentés dans le fournisseur Databricks.

Le fournisseur Databricks inclut des opérateurs pour exécuter plusieurs tâches sur un Workspace Databricks, notamment l'importation de données dans une table, l'exécution de requêtes SQL et l'utilisation de dossiers Git Databricks.

Le fournisseur Databricks implémente deux opérateurs pour déclencher des jobs :

Pour créer un nouveau Job Databricks ou Reset un Job existant, le fournisseur Databricks implémente le DatabricksCreateJobsOperator. Le DatabricksCreateJobsOperator utilise l'API POST /api/2.1/jobs/create et POST /api/2.1/jobs/reset Requêtes API. Vous pouvez utiliser le DatabricksCreateJobsOperator avec le DatabricksRunNowOperator pour créer et exécuter un job.

remarque

L'utilisation des opérateurs Databricks pour Trigger un Job nécessite de fournir des identifiants dans la configuration de connexion Databricks. Voir Créez un jeton d'accès personnel Databricks pour Airflow.

Les opérateurs Databricks Airflow écrivent l'URL de la page d'exécution du Job dans les logs Airflow toutes les polling_period_seconds (la valeur par default est de 30 secondes). Pour plus d'informations, consultez la page du package apache-airflow-providers-databricks sur le site web d'Airflow.

Installez l'intégration Airflow Databricks localement​

Pour installer Airflow et le fournisseur Databricks localement pour les tests et le développement, utilisez les étapes suivantes. Pour les autres options d'installation d'Airflow, y compris la création d'une installation de production, consultez l'installation dans la documentation Airflow.

Ouvrez un terminal et exécutez les commandes suivantes :

Bash
mkdir airflow
cd airflow
pipenv --python 3.8
pipenv shell
export AIRFLOW_HOME=$(pwd)
pipenv install apache-airflow
pipenv install apache-airflow-providers-databricks
mkdir dags
airflow db init
airflow users create --username admin --firstname <firstname> --lastname <lastname> --role Admin --email <email>

Remplacez <firstname>, <lastname> et <email> par votre nom d'utilisateur et votre e-mail. Il vous est demandé de saisir un mot de passe pour l'utilisateur administrateur. Veillez à enregistrer ce mot de passe, car il est requis pour se connecter à l'interface utilisateur d'Airflow.

Ce script effectue les étapes suivantes :

  1. Crée un répertoire nommé airflow et se place dans ce répertoire.
  2. Utilise pipenv pour créer et générer un environnement virtuel Python. Databricks recommande d'utiliser un environnement virtuel Python pour isoler les versions de package et les dépendances de code à cet environnement. Cette isolation aide à réduire les incohérences inattendues de versions de package et les collisions de dépendances de code.
  3. Initialise une variable d'environnement nommée AIRFLOW_HOME définie sur le chemin du répertoire airflow.
  4. Installe Airflow et les packages du fournisseur Airflow Databricks.
  5. Crée un répertoire airflow/dags. Airflow utilise le répertoire dags pour stocker les définitions de DAG.
  6. Initialise une base de données SQLite qu'Airflow utilise pour suivre les métadonnées. Dans un déploiement Airflow de production, vous configurerez Airflow avec une base de données standard. La base de données SQLite et la configuration par default de votre déploiement Airflow sont initialisées dans le répertoire airflow.
  7. Crée un utilisateur admin pour Airflow.
astuce

Pour confirmer l'installation du fournisseur Databricks, exécutez la commande suivante dans le répertoire d'installation d'Airflow :

Bash
airflow providers list

Start le serveur web et l'ordonnanceur Airflow​

Le serveur web Airflow est requis pour afficher l'interface utilisateur Airflow. Pour start le serveur web, ouvrez un terminal dans le répertoire d'installation d'Airflow et exécutez les commandes suivantes :

remarque

Si le serveur web Airflow ne start pas en raison d'un conflit de ports, vous pouvez modifier le port default dans la configuration Airflow.

Bash
pipenv shell
export AIRFLOW_HOME=$(pwd)
airflow webserver

L'ordonnanceur est le composant Airflow qui planifie les DAGs. Pour start le planificateur, ouvrez un nouveau terminal dans le répertoire d'installation d'Airflow et exécutez les commandes suivantes :

Bash
pipenv shell
export AIRFLOW_HOME=$(pwd)
airflow scheduler

Tester l'installation d'Airflow​

Pour vérifier l'installation d'Airflow, vous pouvez exécuter l'un des DAGs d'exemple inclus avec Airflow :

  1. Dans une fenêtre de navigateur, ouvrez http://localhost:8080/home. Connexion à l'interface utilisateur d'Airflow avec le nom d'utilisateur et le mot de passe que vous avez créés lors de l'installation d'Airflow. La page DAGs d'Airflow apparaît.
  2. Cliquez sur le bouton **Pause/Reprendre DAG** pour réactiver l'un des DAG d'exemple, par exemple,example_python_operator le.
  3. Trigger l'exemple de DAG en cliquant sur le bouton Déclencher le DAG .
  4. Cliquez sur le nom du DAG pour afficher les détails, y compris l'état d'exécution du DAG.

Créez un jeton d'accès personnel Databricks pour Airflow​

Airflow se connecte à Databricks à l'aide d'un jeton d'accès personnel Databricks (PAT). Pour créer un PAT, suivez les étapes de Créer des jetons d'accès personnels pour les utilisateurs du workspace.

remarque

En tant que bonne pratique de sécurité lorsque vous vous authentifiez avec des outils, des systèmes, des scripts et des applications automatisés, Databricks vous recommande d'utiliser des jetons OAuth.

Si vous utilisez l'authentification par jeton d'accès personnel, Databricks recommande d'utiliser des jetons d'accès personnels appartenant aux Service Principal plutôt qu'aux utilisateurs du Workspace. Pour créer des jetons pour les Service Principals, consultez Gérer les jetons pour un Service Principal.

Vous pouvez également vous authentifier auprès de Databricks à l’aide de l’OAuth Databricks pour les Service Principal. Voir Databricks Connection dans la documentation Airflow.

Configurez une connexion Databricks​

Votre installation Airflow contient une connexion default pour Databricks. Pour mettre à jour la connexion afin de vous connecter à votre Workspace à l'aide du jeton d'accès personnel que vous avez créé ci-dessus :

  1. Dans une fenêtre de navigateur, ouvrez http://localhost:8080/connection/list/. Si vous êtes invité à vous connecter, entrez votre nom d’utilisateur et mot de passe d’administrateur.
  2. Sous Conn ID , localisez databricks_default et cliquez sur le bouton Edit record .
  3. Remplacez la valeur du champ **Hôte** par le nom d'instance du Workspace de votre déploiement Databricks, parhttps://adb-123456789.cloud.databricks.com exemple,.
  4. Dans le champ Mot de passe , saisissez votre jeton d’accès personnel Databricks.
  5. Cliquez sur Enregistrer .

Exemple : Créez un DAG Airflow pour exécuter un Job Databricks.​

L'exemple suivant montre comment créer un déploiement Airflow simple qui s'exécute sur votre machine locale et déploie un DAG d'exemple pour Trigger des exécutions dans Databricks. Dans cet exemple, vous allez :

  1. Créez un nouveau Notebook et ajoutez du code pour afficher un message de bienvenue basé sur un parameter configuré.
  2. Créez un job Databricks avec une seule tâche qui exécute le notebook.
  3. Configurez une connexion Airflow à votre workspace Databricks.
  4. Créez un DAG Airflow pour Trigger le Job Notebook. Vous définissez le DAG dans un script Python en utilisant DatabricksRunNowOperator.
  5. Utilisez l'interface utilisateur d'Airflow pour Trigger le DAG et afficher l'état de l'exécution.

Créer un Notebook​

Cet exemple utilise un Notebook contenant deux cellules :

  • La première cellule contient un widget de texte des utilitaires Databricks définissant une variable nommée greeting définie sur la valeur par default world.
  • La deuxième cellule imprime la valeur de la variable greeting préfixée par hello.

Pour créer le Notebook :

  1. Accédez à votre Workspace Databricks, cliquez Nouvelle icône sur **Nouveau** dans la barre latérale, et sélectionnez **Notebook**.

  2. Donnez un nom à votre Notebook, tel que Hello Airflow, et assurez-vous que la langue default est définie sur Python.

  3. Copiez le code Python suivant et collez-le dans la première cellule du notebook.

    Python
    dbutils.widgets.text("greeting", "world", "Greeting")
    greeting = dbutils.widgets.get("greeting")

  4. Ajoutez une nouvelle cellule sous la première cellule et copiez-collez le code Python suivant dans la nouvelle cellule :

    Python
    print("hello {}".format(greeting))

Créer un Job​

  1. Dans votre Workspace, cliquez Icône Workflows. sur **Tâches et pipelines** dans la barre latérale.
  2. Cliquez sur Créer , puis sur Job .
  3. Cliquez sur la vignette Notebook pour configurer la première tâche. Si la vignette Notebook n'est pas disponible, cliquez sur Ajouter un autre type de tâche et recherchez Notebook .
  4. En option, remplacez le nom du Job, qui est par default New Job <date-time> , par le nom de votre Job.
  5. Dans Nom de la tâche , entrez un nom pour la tâche, par exemple greeting-task.
  6. Dans le menu déroulant Source , sélectionnez Workspace .
  7. Cliquez sur la zone de texte Chemin et utilisez l'explorateur de fichiers pour trouver le Notebook que vous avez créé, cliquez sur le nom du Notebook, et cliquez sur Confirmer .
  8. Click Add under parameter . Dans le champ Clé , saisissez greeting. Dans le champ Valeur , saisissez Airflow user.
  9. Cliquez sur Enregistrer la tâche .

Dans le panneau **Détails du Job**, copiez la valeur **ID du Job**. Cette valeur est requise pour trigger le Job depuis Airflow.

Exécuter le job​

Pour tester votre nouveau job dans l'interface utilisateur de Lakeflow Jobs, cliquez sur Bouton Exécuter maintenant dans le coin supérieur droit. Une fois l'exécution terminée, vous pouvez vérifier le résultat en consultant les détails d'exécution du Job.

Créer un nouveau DAG Airflow​

Vous définissez un DAG Airflow dans un fichier Python. Pour créer un DAG pour Trigger le Job du Notebook d'exemple :

  1. Dans un éditeur de texte ou un IDE, créez un nouveau fichier nommé databricks_dag.py avec le contenu suivant :

    Python
    from airflow import DAG
    from airflow.providers.databricks.operators.databricks import DatabricksRunNowOperator
    from airflow.utils.dates import days_ago

    default_args = {
      'owner': 'airflow'
    }

    with DAG('databricks_dag',
      start_date = days_ago(2),
      schedule_interval = None,
      default_args = default_args
      ) as dag:

      opr_run_now = DatabricksRunNowOperator(
        task_id = 'run_now',
        databricks_conn_id = 'databricks_default',
        job_id = JOB_ID
      )

    Remplacez JOB_ID par la valeur de l’ID du Job enregistré précédemment.

  2. Enregistrez le fichier dans le répertoire airflow/dags. Airflow lit et installe automatiquement les fichiers DAG stockés dans airflow/dags/.

Installer et vérifier le DAG dans Airflow​

Pour Trigger et vérifier le DAG dans l'interface utilisateur d'Airflow :

  1. Dans une fenêtre de navigateur, ouvrez http://localhost:8080/home. L’écran **DAGs** Airflow apparaît.
  2. Localisez databricks_dag et cliquez sur le bouton bascule Suspendre/Reprendre le DAG pour reprendre le DAG.
  3. Déclenchez le DAG en cliquant sur le bouton Trigger DAG .
  4. Cliquez sur une exécution dans la colonne Exécutions pour afficher le statut et les détails de l'exécution.
Sur cette page