Cloud Composer 1 | Cloud Composer 2
Cette page explique comment utiliser Cloud Composer 2 pour exécuter des charges de travail Dataproc sans serveur sur Google Cloud.
Les exemples figurant dans les sections suivantes vous expliquent comment utiliser les opérateurs pour gérer les charges de travail par lot Dataproc sans serveur. Vous utilisez ces opérateurs dans les DAG qui créent, suppriment, répertorient et obtiennent une charge de travail par lot Spark sans serveur Dataproc:
Créez des DAG pour les opérateurs qui fonctionnent avec les charges de travail par lot sans serveur Dataproc:
Créez des DAG qui utilisent des conteneurs personnalisés et Dataproc Metastore.
Configurez le serveur d'historique persistant pour ces DAG.
Avant de commencer
Activez l'API Dataproc:
Console
Activez l'API Dataproc
gcloud
Activer l'API Dataproc :
gcloud services enable dataproc.googleapis.com
Sélectionnez l'emplacement du fichier de votre charge de travail Batch. Vous pouvez utiliser l'une des options suivantes:
- Créez un bucket Cloud Storage dans lequel stocker ce fichier.
- Utilisez le bucket de votre environnement. Étant donné que vous n'avez pas besoin de synchroniser ce fichier avec Airflow, vous pouvez créer un sous-dossier distinct en dehors des dossiers
/dags
ou/data
. Exemple :/batches
- Utiliser un bucket existant
Configurer des fichiers et des variables Airflow
Cette section explique comment configurer des fichiers et des variables Airflow pour ce tutoriel.
Importer un fichier de charge de travail Spark ML sans serveur Dataproc dans un bucket
La charge de travail de ce tutoriel exécute un script pyspark:
Enregistrez le script pyspark dans un fichier local nommé
spark-job.py
. Vous pouvez utiliser l'exemple de script pyspark.Importez le fichier à l'emplacement que vous avez sélectionné à la section Avant de commencer.
Définir les variables Airflow
Les exemples des sections suivantes utilisent des variables Airflow. Vous définissez des valeurs pour ces variables dans Airflow, puis votre code DAG peut accéder à ces valeurs.
Les exemples de ce tutoriel utilisent les variables Airflow suivantes. Vous pouvez les définir si nécessaire, selon l'exemple que vous utilisez.
Définissez les variables Airflow suivantes à utiliser dans le code de votre DAG:
project_id
: ID du projet.bucket_name
: URI d'un bucket où se trouve le fichier Python principal de la charge de travail (spark-job.py
). Vous avez sélectionné cet emplacement à la section Avant de commencer.phs_cluster
: nom du cluster du serveur d'historique persistant. Vous définissez cette variable lorsque vous créez un serveur d'historique persistant.image_name
: nom et tag de l'image de conteneur personnalisé (image:tag
). Vous définissez cette variable lorsque vous utilisez une image de conteneur personnalisé avec DataprocCreateBatchOperator.metastore_cluster
: nom du service Dataproc Metastore. Vous définissez cette variable lorsque vous utilisez le service Dataproc Metastore avec DataprocCreateBatchOperator.region_name
: région dans laquelle se trouve le service Dataproc Metastore. Vous définissez cette variable lorsque vous utilisez le service Dataproc Metastore avec DataprocCreateBatchOperator.
Définir chaque variable Airflow à l'aide de la console Google Cloud et de l'interface utilisateur Airflow
Dans la console Google Cloud, accédez à la page Environnements.
Dans la liste des environnements, cliquez sur le lien Airflow correspondant à votre environnement. L'interface utilisateur Airflow s'ouvre.
Dans l'interface utilisateur Airflow, sélectionnez Admin > Variables.
Cliquez sur Ajouter un enregistrement.
Indiquez le nom de la variable dans le champ Key (Clé) et définissez sa valeur dans le champ Val.
Cliquez sur Enregistrer.
Créer un serveur d'historique persistant
Utilisez un serveur d'historique persistant (PHS) pour afficher les fichiers d'historique Spark de vos charges de travail par lot:
- Créez un serveur d'historique persistant.
- Assurez-vous d'avoir spécifié le nom du cluster PHS dans la variable Airflow
phs_cluster
.
DataprocCreateBatchOperator
Le DAG suivant démarre une charge de travail par lot Dataproc sans serveur.
Pour en savoir plus sur les arguments DataprocCreateBatchOperator
, consultez le code source de l'opérateur.
Pour en savoir plus sur les attributs que vous pouvez transmettre dans le paramètre batch
de DataprocCreateBatchOperator
, consultez la description de la classe Batch.
Utiliser une image de conteneur personnalisé avec DataprocCreateBatchOperator
L'exemple suivant montre comment exécuter vos charges de travail à l'aide d'une image de conteneur personnalisé. Vous pouvez utiliser un conteneur personnalisé, par exemple, pour ajouter des dépendances Python non fournies par l'image de conteneur par défaut.
Pour utiliser une image de conteneur personnalisé:
Créez une image de conteneur personnalisé et importez-la dans Container Registry.
Spécifiez l'image dans la variable Airflow
image_name
.Utilisez DataprocCreateBatchOperator avec votre image personnalisée:
Utiliser le service Dataproc Metastore avec DataprocCreateBatchOperator
Pour utiliser un service Dataproc Metastore à partir d'un DAG, procédez comme suit:
Vérifiez que votre service de métastore est déjà démarré.
Pour en savoir plus sur le démarrage d'un service de métastore, consultez la page Activer et désactiver Dataproc Metastore.
Pour en savoir plus sur l'opérateur de traitement par lot pour la création de la configuration, consultez PeripheralsConfig.
Une fois le service de métastore opérationnel, spécifiez son nom dans la variable
metastore_cluster
et sa région dans la variable Airflowregion_name
.Utilisez le service de métastore dans DataprocCreateBatchOperator:
DataprocDeleteBatchOperator
Vous pouvez utiliser DataprocDeleteBatchOperator pour supprimer un lot en fonction de l'ID de lot de la charge de travail.
DataprocListBatchesOperator
DataprocDeleteBatchOperator répertorie les lots qui existent dans un ID de projet et une région donnés.
DataprocGetBatchOperator
DataprocGetBatchOperator récupère une charge de travail par lot particulière.