Cloud Composer 1 | Cloud Composer 2
Ce tutoriel explique comment utiliser Cloud Composer pour créer un DAG (graphe orienté acyclique) Apache Airflow qui exécute une tâche de décompte de mots Apache Hadoop sur un cluster Dataproc.
Objectifs
- Accédez à votre environnement Cloud Composer et utilisez l'interface utilisateur Airflow.
- Créer et afficher des variables d'environnement Airflow
- Créez et exécutez un DAG comprenant les tâches suivantes :
- crée un cluster Dataproc ;
- Exécute une tâche de décompte de mots Apache Hadoop sur le cluster.
- Génère les résultats du décompte de mots dans un bucket Cloud Storage.
- Supprime le cluster.
Coûts
Dans ce document, vous utilisez les composants facturables suivants de Google Cloud :
- Cloud Composer
- Dataproc
- Cloud Storage
Obtenez une estimation des coûts en fonction de votre utilisation prévue à l'aide du simulateur de coût.
Avant de commencer
Assurez-vous que les API suivantes sont activées dans votre projet:
Console
Activer les API Dataproc, Cloud Storage.
gcloud
Activer les API Dataproc, Cloud Storage :
gcloud services enable dataproc.googleapis.com
storage-component.googleapis.com Dans votre projet, créez un bucket Cloud Storage de n'importe quelle classe de stockage et région pour stocker les résultats de la tâche Hadoop de décompte de mots.
Notez le chemin d'accès au bucket que vous avez créé, par exemple
gs://example-bucket
. Vous définirez une variable Airflow pour ce chemin et l'utiliserez dans l'exemple de DAG plus loin dans ce tutoriel.Créez un environnement Cloud Composer avec les paramètres par défaut. Attendez la fin de la création de l'environnement. Lorsque vous avez terminé, une coche verte s'affiche à gauche du nom de l'environnement.
Notez la région dans laquelle vous avez créé votre environnement, par exemple
us-central
. Vous allez définir une variable Airflow pour cette région et l'utiliser dans l'exemple de DAG afin d'exécuter un cluster Dataproc dans la même région.
Définir les variables Airflow
Définissez les variables Airflow à utiliser ultérieurement dans l'exemple de DAG. Par exemple, vous pouvez définir des variables Airflow dans l'interface utilisateur Airflow.
Variable Airflow | Valeur |
---|---|
gcp_project
|
L'ID du projet que vous utilisez pour ce tutoriel, par exemple example-project . |
gcs_bucket
|
L'URI du bucket Cloud Storage que vous avez créé pour ce tutoriel, tel que gs://example-bucket |
gce_region
|
Région dans laquelle vous avez créé votre environnement, par exemple us-central1 .
Il s'agit de la région dans laquelle votre cluster Dataproc sera créé. |
Afficher l'exemple de workflow
Un DAG Airflow est un ensemble de tâches organisées que vous souhaitez planifier et exécuter. Les DAG sont définis dans des fichiers Python standards. Le code affiché dans hadoop_tutorial.py
est le code du workflow.
Airflow 2
Airflow 1
Opérateurs
Pour orchestrer les trois tâches de l'exemple de workflow, le DAG importe les trois opérateurs Airflow suivants:
DataprocClusterCreateOperator
: crée un cluster Dataproc.DataProcHadoopOperator
: envoie un job de décompte de mots Hadoop et écrit les résultats dans un bucket Cloud Storage.DataprocClusterDeleteOperator
: supprime le cluster pour éviter que des frais Compute Engine continuent de vous être facturés.
Dépendances
Vous organisez les tâches que vous souhaitez exécuter de manière à refléter leurs relations et leurs dépendances. Les tâches de ce DAG s'exécutent de manière séquentielle.
Airflow 2
Airflow 1
Planification
Le nom du DAG est composer_hadoop_tutorial
, et il s'exécute une fois par jour. Étant donné que la valeur start_date
transmise à default_dag_args
est définie sur yesterday
, Cloud Composer planifie le démarrage du workflow immédiatement après l'importation du DAG dans le bucket de l'environnement.
Airflow 2
Airflow 1
Importer le DAG dans le bucket de l'environnement
Cloud Composer stocke les DAG dans le dossier /dags
du bucket de votre environnement.
Pour importer le DAG :
Sur votre ordinateur local, enregistrez
hadoop_tutorial.py
.Dans la console Google Cloud, accédez à la page Environnements.
Dans la liste des environnements, dans la colonne Dossier des DAG de votre environnement, cliquez sur le lien DAG.
Cliquez sur Importer des fichiers.
Sélectionnez
hadoop_tutorial.py
sur votre machine locale, puis cliquez sur Ouvrir.
Cloud Composer ajoute le DAG à Airflow et le planifie automatiquement. Les modifications sont appliquées au DAG après trois à cinq minutes.
Explorer les exécutions du DAG
Afficher l'état de la tâche
Lorsque vous importez votre fichier DAG dans le dossier dags/
de Cloud Storage, Cloud Composer l'analyse. Une fois l'opération terminée, le nom du workflow apparaît dans la liste des DAG, et le workflow est mis en file d'attente pour s'exécuter immédiatement.
Pour connaître l'état des tâches, accédez à l'interface Web Airflow, puis cliquez sur DAGs (DAG) dans la barre d'outils.
Pour ouvrir la page de détails des DAG, cliquez sur
composer_hadoop_tutorial
. Cette page comprend une représentation graphique des tâches et des dépendances du workflow.Pour connaître l'état de chaque tâche, cliquez sur Graph View (Vue graphique), puis sur le graphique de chaque tâche.
Ajouter à nouveau le workflow à la file d'attente
Pour exécuter de nouveau le workflow à partir de la vue graphique, procédez comme suit :
- Dans la vue graphique de l'interface utilisateur Airflow, cliquez sur le graphique
create_dataproc_cluster
. - Pour réinitialiser les trois tâches, cliquez sur Clear (Effacer), puis sur OK pour confirmer.
- Cliquez de nouveau sur
create_dataproc_cluster
dans la vue graphique. - Pour remettre le workflow en file d'attente, cliquez sur Run (Exécuter).
Afficher les résultats des tâches
Vous pouvez également vérifier l'état et les résultats du workflow composer_hadoop_tutorial
en accédant aux pages suivantes de la console Google Cloud:
Clusters Dataproc: surveiller la création et la suppression des clusters. Notez que le cluster créé par le workflow est éphémère: il n'existe que pour la durée du workflow et est supprimé lors de la dernière tâche.
Tâches Dataproc: afficher ou surveiller le job de décompte de mots Apache Hadoop Cliquez sur l'ID de job pour afficher la sortie du journal associée au job.
Navigateur Cloud Storage: affichez les résultats du décompte de mots dans le dossier
wordcount
du bucket Cloud Storage que vous avez créé pour ce tutoriel.
Nettoyage
Supprimez les ressources utilisées dans ce tutoriel:
Supprimez l'environnement Cloud Composer, y compris en supprimant manuellement le bucket de l'environnement.
Supprimez le bucket Cloud Storage qui stocke les résultats de la tâche Hadoop de décompte de mots.