Exécuter des charges de travail Dataproc sans serveur avec Cloud Composer

Cloud Composer 1 | Cloud Composer 2 | Cloud Composer 3

Cette page explique comment utiliser Cloud Composer 2 pour exécuter les charges de travail Dataproc sans serveur sur Google Cloud.

Les exemples des sections suivantes vous montrent comment utiliser des 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, Supprimez, répertoriez et obtenez une charge de travail par lot Spark sans serveur pour Dataproc:

• Créer des DAG pour les opérateurs compatibles avec les charges de travail par lot Dataproc sans serveur:

  • DataprocCreateBatchOperator
  • DataprocDeleteBatchOperator
  • DataprocListBatchesOperator
  • DataprocGetBatchOperator

• Créer des DAG qui utilisent des conteneurs personnalisés, et Dataproc Metastore.

• Configurez le serveur d'historique persistant pour ces DAG.

Avant de commencer

1. Activez l'API Dataproc:

   Console

   Enable the Dataproc API.

   Enable the API

   gcloud

   Enable the Dataproc API:

   gcloud services enable dataproc.googleapis.com

2. Sélectionnez l'emplacement de votre fichier de charge de travail par lot. Vous pouvez utiliser l'une des options suivantes :

   • Créez un bucket Cloud Storage qui stocke ce fichier.
   • Utilisez le bucket de votre environnement. Comme vous n'avez pas besoin de synchroniser ce fichier avec Airflow, vous pouvez créer un sous-dossier distinct en dehors de /dags ou /data dossiers. Par exemple, /batches.
   • Utilisez un bucket existant.

Configurer les fichiers et les variables Airflow

Cette section explique comment configurer les fichiers et les 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:

1. Enregistrez tous les scripts pyspark dans un fichier local nommé spark-job.py. Par exemple, vous pouvez utiliser la méthode exemple de script pyspark.

2. Importez le fichier à l'emplacement que vous avez sélectionné dans 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 y accéder.

Les exemples de ce tutoriel utilisent les variables Airflow suivantes. Vous pouvez les configurer au besoin, en fonction de l'exemple que vous utilisez.

Définissez les variables Airflow suivantes à utiliser dans votre code DAG :

• project_id: ID du projet
• bucket_name: URI d'un bucket où se trouve le fichier Python principal de où se trouve la charge de travail (spark-job.py). Vous avez sélectionné cet emplacement dans 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 balise de l'image de conteneur personnalisée (image:tag). Vous définissez cette variable lorsque vous utilisez une image de conteneur personnalisée 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.

Utiliser la console Google Cloud et l'interface utilisateur Airflow pour définir chaque variable Airflow

1. Dans la console Google Cloud, accédez à la page Environnements.

   Accéder à la page Environnements

2. Dans la liste des environnements, cliquez sur le lien Airflow correspondant à votre environnement. L'interface utilisateur d'Airflow s'ouvre.

3. Dans l'interface utilisateur d'Airflow, sélectionnez Admin > Variables.

4. Cliquez sur Ajouter un enregistrement.

5. Spécifiez le nom de la variable dans le champ Clé, puis définissez sa valeur dans le champ Val.

6. Cliquez sur Enregistrer.

Créer un serveur d'historique persistant

Utiliser un serveur d'historique persistant (PHS) pour afficher les fichiers d'historique Spark de votre lot charges de travail:

1. Créez un serveur d'historique persistant.
2. Assurez-vous d'avoir spécifié le nom du cluster PHS dans le fichier phs_cluster Variable Airflow.

DataprocCreateBatchOperator

Le DAG suivant démarre une charge de travail par lot Dataproc sans serveur.

Pour en savoir plus sur les arguments DataprocCreateBatchOperator, consultez code source de l'opérateur.

Pour en savoir plus sur les attributs que vous pouvez transmettre dans batch de DataprocCreateBatchOperator, consultez les description de la classe Batch.

"""
Examples below show how to use operators for managing Dataproc Serverless batch workloads.
 You use these operators in DAGs that create, delete, list, and get a Dataproc Serverless Spark batch workload.
https://airflow.apache.org/docs/apache-airflow/stable/concepts/variables.html
* project_id is the Google Cloud Project ID to use for the Cloud Dataproc Serverless.
* bucket_name is the URI of a bucket where the main python file of the workload (spark-job.py) is located.
* phs_cluster is the Persistent History Server cluster name.
* image_name is the name and tag of the custom container image (image:tag).
* metastore_cluster is the Dataproc Metastore service name.
* region_name is the region where the Dataproc Metastore service is located.
"""

import datetime

from airflow import models
from airflow.providers.google.cloud.operators.dataproc import (
    DataprocCreateBatchOperator,
    DataprocDeleteBatchOperator,
    DataprocGetBatchOperator,
    DataprocListBatchesOperator,
)
from airflow.utils.dates import days_ago

PROJECT_ID = "{{ var.value.project_id }}"
REGION = "{{ var.value.region_name}}"
BUCKET = "{{ var.value.bucket_name }}"
PHS_CLUSTER = "{{ var.value.phs_cluster }}"
METASTORE_CLUSTER = "{{var.value.metastore_cluster}}"
DOCKER_IMAGE = "{{var.value.image_name}}"

PYTHON_FILE_LOCATION = "gs://{{var.value.bucket_name }}/spark-job.py"
# for e.g.  "gs//my-bucket/spark-job.py"
# Start a single node Dataproc Cluster for viewing Persistent History of Spark jobs
PHS_CLUSTER_PATH = "projects/{{ var.value.project_id }}/regions/{{ var.value.region_name}}/clusters/{{ var.value.phs_cluster }}"
# for e.g. projects/my-project/regions/my-region/clusters/my-cluster"
SPARK_BIGQUERY_JAR_FILE = "gs://spark-lib/bigquery/spark-bigquery-latest_2.12.jar"
# use this for those pyspark jobs that need a spark-bigquery connector
# https://cloud.google.com/dataproc/docs/tutorials/bigquery-connector-spark-example
# Start a Dataproc MetaStore Cluster
METASTORE_SERVICE_LOCATION = "projects/{{var.value.project_id}}/locations/{{var.value.region_name}}/services/{{var.value.metastore_cluster }}"
# for e.g. projects/my-project/locations/my-region/services/my-cluster
CUSTOM_CONTAINER = "us.gcr.io/{{var.value.project_id}}/{{ var.value.image_name}}"
# for e.g. "us.gcr.io/my-project/quickstart-image",

default_args = {
    # Tell airflow to start one day ago, so that it runs as soon as you upload it
    "start_date": days_ago(1),
    "project_id": PROJECT_ID,
    "region": REGION,
}
with models.DAG(
    "dataproc_batch_operators",  # The id you will see in the DAG airflow page
    default_args=default_args,  # The interval with which to schedule the DAG
    schedule_interval=datetime.timedelta(days=1),  # Override to match your needs
) as dag:
    create_batch = DataprocCreateBatchOperator(
        task_id="batch_create",
        batch={
            "pyspark_batch": {
                "main_python_file_uri": PYTHON_FILE_LOCATION,
                "jar_file_uris": [SPARK_BIGQUERY_JAR_FILE],
            },
            "environment_config": {
                "peripherals_config": {
                    "spark_history_server_config": {
                        "dataproc_cluster": PHS_CLUSTER_PATH,
                    },
                },
            },
        },
        batch_id="batch-create-phs",
    )
    list_batches = DataprocListBatchesOperator(
        task_id="list-all-batches",
    )

    get_batch = DataprocGetBatchOperator(
        task_id="get_batch",
        batch_id="batch-create-phs",
    )
    delete_batch = DataprocDeleteBatchOperator(
        task_id="delete_batch",
        batch_id="batch-create-phs",
    )
    create_batch >> list_batches >> get_batch >> delete_batch

Utiliser une image de conteneur personnalisée avec DataprocCreateBatchOperator

L'exemple suivant montre comment utiliser une image de conteneur personnalisée pour exécuter vos charges de travail. Vous pouvez utiliser un conteneur personnalisé, par exemple, pour ajouter des dépendances Python non fournies par l'image de conteneur par défaut.

Remarque :Pour exécuter des DAG exécutant Dataproc sans serveur pour Spark s'exécute avec des conteneurs Docker personnalisés, votre environnement Cloud Composer vous devez utiliser Cloud Composer 2.0.6 et Airflow 2.2.3 ou version ultérieure ; versions.

Pour utiliser une image de conteneur personnalisée :

1. Créez une image de conteneur personnalisée et importez-la dans Container Registry.

2. Spécifiez l'image dans la variable Airflow image_name.

3. Utilisez DataprocCreateBatchOperator avec votre image personnalisée:

create_batch_with_custom_container = DataprocCreateBatchOperator(
    task_id="dataproc_custom_container",
    batch={
        "pyspark_batch": {
            "main_python_file_uri": PYTHON_FILE_LOCATION,
            "jar_file_uris": [SPARK_BIGQUERY_JAR_FILE],
        },
        "environment_config": {
            "peripherals_config": {
                "spark_history_server_config": {
                    "dataproc_cluster": PHS_CLUSTER_PATH,
                },
            },
        },
        "runtime_config": {
            "container_image": CUSTOM_CONTAINER,
        },
    },
    batch_id="batch-custom-container",
)
get_batch_custom = DataprocGetBatchOperator(
    task_id="get_batch_custom",
    batch_id="batch-custom-container",
)
delete_batch_custom = DataprocDeleteBatchOperator(
    task_id="delete_batch_custom",
    batch_id="batch-custom-container",
)
create_batch_with_custom_container >> get_batch_custom >> delete_batch_custom

Utiliser le service Dataproc Metastore avec DataprocCreateBatchOperator

Pour utiliser un service Dataproc Metastore à partir d'un DAG :

1. Vérifiez que votre service de métadonnées est déjà démarré.

   Pour savoir comment démarrer un service de métastore, consultez la section Activer et désactiver Dataproc Metastore.

   Pour en savoir plus sur l'opérateur de traitement par lot pour la création des configuration, consultez PeripheralsConfig.

2. Une fois que le service de métastore est opérationnel, spécifiez son nom dans La variable metastore_cluster et sa région dans la variable Airflow region_name.

3. Utilisez le service Metastore dans DataprocCreateBatchOperator :

create_batch_with_metastore = DataprocCreateBatchOperator(
    task_id="dataproc_metastore",
    batch={
        "pyspark_batch": {
            "main_python_file_uri": PYTHON_FILE_LOCATION,
            "jar_file_uris": [SPARK_BIGQUERY_JAR_FILE],
        },
        "environment_config": {
            "peripherals_config": {
                "metastore_service": METASTORE_SERVICE_LOCATION,
                "spark_history_server_config": {
                    "dataproc_cluster": PHS_CLUSTER_PATH,
                },
            },
        },
    },
    batch_id="dataproc-metastore",
)
get_batch_metastore = DataprocGetBatchOperator(
    task_id="get_batch_metatstore",
    batch_id="dataproc-metastore",
)
delete_batch_metastore = DataprocDeleteBatchOperator(
    task_id="delete_batch_metastore",
    batch_id="dataproc-metastore",
)

create_batch_with_metastore >> get_batch_metastore >> delete_batch_metastore

DataprocDeleteBatchOperator

Vous pouvez utiliser DataprocDeleteBatchOperator pour supprimer un lot en fonction de l'ID de lot de la charge de travail.

delete_batch = DataprocDeleteBatchOperator(
    task_id="delete_batch",
    batch_id="batch-create-phs",
)

DataprocListBatchesOperator

DataprocDeleteBatchOperator répertorie les lots qui existent dans un project_id donné. et la région.

list_batches = DataprocListBatchesOperator(
    task_id="list-all-batches",
)

DataprocGetBatchOperator

DataprocGetBatchOperator récupère une charge de travail par lot particulière.

get_batch = DataprocGetBatchOperator(
    task_id="get_batch",
    batch_id="batch-create-phs",
)

Étape suivante

• Présentation de Dataproc sans serveur
• Déclencher des DAG avec des fonctions Cloud Run