Flusso di lavoro con Managed Airflow

Obiettivo:

1. Crea un modello di flusso di lavoro Managed Service for Apache Spark che esegue un job Spark PI
2. Crea un DAG di Apache Airflow che Managed Airflow utilizzerà per avviare il workflow a un orario specifico.

In questo documento vengono utilizzati i seguenti componenti fatturabili di Google Cloud:

• Managed Service for Apache Spark
• Compute Engine
• Managed Airflow

Per generare una stima dei costi in base all'utilizzo previsto, utilizza il calcolatore prezzi.

I nuovi utenti di Google Cloud potrebbero avere diritto a una prova senza costi.

Prima di iniziare

Configura il progetto

Accedi al tuo account Google Cloud . Se non conosci Google Cloud, crea un account per valutare le prestazioni dei nostri prodotti in scenari reali. I nuovi clienti ricevono anche 300 $di crediti senza costi per l'esecuzione, il test e il deployment dei carichi di lavoro.

In the Google Cloud console, on the project selector page, select or create a Google Cloud project.

Roles required to select or create a project

• Select a project: Selecting a project doesn't require a specific IAM role—you can select any project that you've been granted a role on.
• Create a project: To create a project, you need the Project Creator role (roles/resourcemanager.projectCreator), which contains the resourcemanager.projects.create permission. Learn how to grant roles.

Note: If you don't plan to keep the resources that you create in this procedure, create a project instead of selecting an existing project. After you finish these steps, you can delete the project, removing all resources associated with the project.

Go to project selector

Verify that billing is enabled for your Google Cloud project.

Enable the Dataproc, Compute Engine, and Managed Airflow APIs.

Roles required to enable APIs

To enable APIs, you need the Service Usage Admin IAM role (roles/serviceusage.serviceUsageAdmin), which contains the serviceusage.services.enable permission. Learn how to grant roles.

Enable the APIs

Installa Google Cloud CLI.

Se utilizzi un provider di identità (IdP) esterno, devi prima accedere a gcloud CLI con la tua identità federata.

Per inizializzare gcloud CLI, esegui questo comando:

gcloud init

In the Google Cloud console, on the project selector page, select or create a Google Cloud project.

Roles required to select or create a project

• Select a project: Selecting a project doesn't require a specific IAM role—you can select any project that you've been granted a role on.
• Create a project: To create a project, you need the Project Creator role (roles/resourcemanager.projectCreator), which contains the resourcemanager.projects.create permission. Learn how to grant roles.

Note: If you don't plan to keep the resources that you create in this procedure, create a project instead of selecting an existing project. After you finish these steps, you can delete the project, removing all resources associated with the project.

Go to project selector

Verify that billing is enabled for your Google Cloud project.

Enable the Dataproc, Compute Engine, and Managed Airflow APIs.

Roles required to enable APIs

To enable APIs, you need the Service Usage Admin IAM role (roles/serviceusage.serviceUsageAdmin), which contains the serviceusage.services.enable permission. Learn how to grant roles.

Enable the APIs

Installa Google Cloud CLI.

Se utilizzi un provider di identità (IdP) esterno, devi prima accedere a gcloud CLI con la tua identità federata.

Per inizializzare gcloud CLI, esegui questo comando:

gcloud init

Crea un modello di flusso di lavoro Managed Service for Apache Spark

Copia ed esegui i seguenti comandi in una finestra del terminale locale o in Cloud Shell per creare e definire un modello di workflow.

1. Crea il modello di workflow sparkpi.

   gcloud dataproc workflow-templates create sparkpi \
       --region=us-central1
         

2. Aggiungi il job Spark al template di workflow sparkpi. Il flag "compute" step-id identifica il job SparkPi.

   gcloud dataproc workflow-templates add-job spark \
       --workflow-template=sparkpi \
       --step-id=compute \
       --class=org.apache.spark.examples.SparkPi \
       --jars=file:///usr/lib/spark/examples/jars/spark-examples.jar \
       --region=us-central1 \
       -- 1000
         

3. Utilizza un cluster gestito, a un singolo nodo per eseguire il flusso di lavoro. Managed Service for Apache Spark creerà il cluster, eseguirà il flusso di lavoro e poi lo eliminerà al termine.

   gcloud dataproc workflow-templates set-managed-cluster sparkpi \
       --cluster-name=sparkpi \
       --single-node \
       --region=us-central1
         

4. Conferma la creazione del modello di workflow.

   Console

   Fai clic sul nome di sparkpi nella pagina Workflow di Managed Service for Apache Spark nella console Google Cloud per aprire la pagina Dettagli modello di workflow. Fai clic sul nome del modello di workflow per confermare gli attributi del modello sparkpi.

   

   Comando g-cloud

   Esegui questo comando:

   gcloud dataproc workflow-templates describe sparkpi --region=us-central1
       

Crea e carica un DAG in Cloud Storage

1. Crea o utilizza un ambiente Airflow gestito esistente.
2. Imposta le variabili di ambiente.

   UI di Airflow

   1. Nella barra degli strumenti, fai clic su Amministrazione > Variabili.
   2. Fai clic su Crea.
   3. Inserisci le seguenti informazioni:
      • Chiave:project_id
      • Val: PROJECT_ID: l'ID progetto Google Cloud
   4. Fai clic su Salva.

   Comando g-cloud

   Inserisci i seguenti comandi:

   • ENVIRONMENT è il nome dell'ambiente Managed Airflow
   • LOCATION è la regione in cui si trova l'ambiente Managed Airflow
   • PROJECT_ID è l'ID progetto del progetto che contiene l'ambiente Managed Airflow

       gcloud composer environments run ENVIRONMENT --location LOCATION variables set -- project_id PROJECT_ID
       

3. Copia il seguente codice DAG localmente in un file denominato "composer-dataproc-dag.py", che utilizza DataprocInstantiateWorkflowTemplateOperator.

   Airflow 2

   
   """Example Airflow DAG that kicks off a Cloud Dataproc Template that runs a
   Spark Pi Job.
   
   This DAG relies on an Airflow variable
   https://airflow.apache.org/docs/apache-airflow/stable/concepts/variables.html
   * project_id - Google Cloud Project ID to use for the Cloud Dataproc Template.
   """
   
   import datetime
   
   from airflow import models
   from airflow.providers.google.cloud.operators.dataproc import (
       DataprocInstantiateWorkflowTemplateOperator,
   )
   from airflow.utils.dates import days_ago
   
   project_id = "{{var.value.project_id}}"
   
   
   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,
   }
   
   # Define a DAG (directed acyclic graph) of tasks.
   # Any task you create within the context manager is automatically added to the
   # DAG object.
   with models.DAG(
       # The id you will see in the DAG airflow page
       "dataproc_workflow_dag",
       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:
       start_template_job = DataprocInstantiateWorkflowTemplateOperator(
           # The task id of your job
           task_id="dataproc_workflow_dag",
           # The template id of your workflow
           template_id="sparkpi",
           project_id=project_id,
           # The region for the template
           region="us-central1",
       )
   

   Airflow 1

   
   """Example Airflow DAG that kicks off a Cloud Dataproc Template that runs a
   Spark Pi Job.
   
   This DAG relies on an Airflow variable
   https://airflow.apache.org/docs/apache-airflow/stable/concepts/variables.html
   * project_id - Google Cloud Project ID to use for the Cloud Dataproc Template.
   """
   
   import datetime
   
   from airflow import models
   from airflow.contrib.operators import dataproc_operator
   from airflow.utils.dates import days_ago
   
   project_id = "{{var.value.project_id}}"
   
   
   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,
   }
   
   # Define a DAG (directed acyclic graph) of tasks.
   # Any task you create within the context manager is automatically added to the
   # DAG object.
   with models.DAG(
       # The id you will see in the DAG airflow page
       "dataproc_workflow_dag",
       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:
       start_template_job = dataproc_operator.DataprocWorkflowTemplateInstantiateOperator(
           # The task id of your job
           task_id="dataproc_workflow_dag",
           # The template id of your workflow
           template_id="sparkpi",
           project_id=project_id,
           # The region for the template
           # For more info on regions where Dataflow is available see:
           # https://cloud.google.com/dataflow/docs/resources/locations
           region="us-central1",
       )
   

4. Carica il DAG nella cartella dell'ambiente in Cloud Storage. Una volta completato il caricamento, fai clic sul link Cartella DAG nella pagina dell'ambiente Managed Airflow.
   

Visualizzare lo stato di un'attività

UI di Airflow

1. Apri l'interfaccia web di Airflow.
2. Nella pagina DAG, fai clic sul nome del DAG (ad esempio dataproc_workflow_dag).
3. Nella pagina dei dettagli dei DAG, fai clic su Visualizzazione grafico.
4. Controlla lo stato:
   • Non riuscita: l'attività è racchiusa in un riquadro rosso. Puoi anche passare il puntatore sopra l'attività e cercare Stato: non riuscita.
   • Riuscita: l'attività è racchiusa in un riquadro verde. Puoi anche passare il puntatore sopra l'attività e verificare che lo stato sia riuscito.

Console

Fai clic sulla scheda Workflows per visualizzare lo stato del flusso di lavoro.

Comando g-cloud

gcloud dataproc operations list \
    --region=us-central1 \
    --filter="labels.goog-dataproc-workflow-template-id=sparkpi"
    

Esegui la pulizia

Per evitare che al tuo account Google Cloud vengano addebitati costi, puoi eliminare le risorse utilizzate in questo tutorial:

1. Elimina l'ambiente Managed Airflow.

2. Elimina il modello di workflow.

Passaggi successivi

• Consulta la panoramica dei modelli di workflow di Managed Service for Apache Spark.
• Vedi Soluzioni di pianificazione del flusso di lavoro.