Esegui un DAG Apache Airflow in Cloud Composer 2 (Google Cloud CLI)

Esegui un DAG Apache Airflow in Cloud Composer 2 (Google Cloud CLI)

Cloud Composer 1 | Cloud Composer 2 | Cloud Composer 3

Questa guida rapida mostra come creare un ambiente Cloud Composer ed eseguire un DAG Apache Airflow in Cloud Composer 2.

• Se non hai mai utilizzato Airflow, consulta le Tutorial sui concetti di Airflow in Apache Airflow documentazione di Airflow per ulteriori informazioni su concetti, oggetti e il loro utilizzo.

• Se vuoi utilizzare la console Google Cloud, consulta Esegui un DAG Apache Airflow in Cloud Composer.

• Se vuoi creare un ambiente utilizzando Terraform, consulta Crea ambienti (Terraform).

Prima di iniziare

1. 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 gratuiti per l'esecuzione, il test e il deployment dei carichi di lavoro.
2. Installa Google Cloud CLI.

3. Per initialize gcloud CLI, esegui questo comando:

   gcloud init

4. Create or select a Google Cloud project.

   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.

   • Create a Google Cloud project:

     gcloud projects create PROJECT_ID

     Replace PROJECT_ID with a name for the Google Cloud project you are creating.

   • Select the Google Cloud project that you created:

     gcloud config set project PROJECT_ID

     Replace PROJECT_ID with your Google Cloud project name.

5. Assicurati che la fatturazione sia attivata per il tuo progetto Google Cloud.

6. Installa Google Cloud CLI.

7. Per initialize gcloud CLI, esegui questo comando:

   gcloud init

8. Create or select a Google Cloud project.

   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.

   • Create a Google Cloud project:

     gcloud projects create PROJECT_ID

     Replace PROJECT_ID with a name for the Google Cloud project you are creating.

   • Select the Google Cloud project that you created:

     gcloud config set project PROJECT_ID

     Replace PROJECT_ID with your Google Cloud project name.

9. Assicurati che la fatturazione sia attivata per il tuo progetto Google Cloud.

10. Attiva l'API Cloud Composer.

    gcloud services enable composer.googleapis.com

11. Per ottenere le autorizzazioni necessarie per completare questa guida rapida, chiedi all'amministratore di concederti i seguenti ruoli IAM sul tuo progetto:

    • Per visualizzare, creare e gestire l'ambiente Cloud Composer:
      • Amministratore oggetti di ambiente e archiviazione (roles/composer.environmentAndStorageObjectAdmin)
      • Utente account di servizio (roles/iam.serviceAccountUser)
    • Per visualizzare i log: Visualizzatore log (roles/logging.viewer)

    Per saperne di più sulla concessione dei ruoli, consulta Gestire l'accesso.

    Potresti anche riuscire a ottenere le autorizzazioni richieste tramite la ruoli o altri ruoli predefiniti ruoli.

Creazione di un ambiente

Se questo è il primo ambiente del progetto, aggiungi l'account dell'agente di servizio Cloud Composer come nuova entità sull'account di servizio del tuo ambiente e concedere ruolo roles/composer.ServiceAgentV2Ext.

Per impostazione predefinita, il tuo ambiente utilizza l'account di servizio predefinito di Compute Engine e nell'esempio seguente viene mostrato come aggiungere questa autorizzazione.

# Get current project's project number
PROJECT_NUMBER=$(gcloud projects list \
  --filter="$(gcloud config get-value project)" \
  --format="value(PROJECT_NUMBER)" \
  --limit=1)

# Add the Cloud Composer v2 API Service Agent Extension role
gcloud iam service-accounts add-iam-policy-binding \
    $[e-mail] \
    --member serviceAccount:service-$[e-mail] \
    --role roles/composer.ServiceAgentV2Ext

Crea un nuovo ambiente denominato example-environment in us-central1 regione, con la versione più recente di Cloud Composer 2.

gcloud composer environments create example-environment \
    --location us-central1 \
    --image-version composer-2.8.6-airflow-2.7.3

Crea un file DAG

Un DAG Airflow è una raccolta di attività organizzate che che vuoi pianificare ed eseguire. I DAG sono definiti in file Python standard.

Questa guida utilizza un esempio di DAG Airflow definito nel file quickstart.py. Il codice Python in questo file esegue le seguenti operazioni:

1. Crea un DAG, composer_sample_dag. Questo DAG viene eseguito ogni giorno.
2. Esegue un'attività, print_dag_run_conf. L'attività stampa l'esecuzione del DAG configurazione utilizzando l'operatore bash.

Salva una copia del file quickstart.py sul computer locale:

import datetime

from airflow import models
from airflow.operators import bash

# If you are running Airflow in more than one time zone
# see https://airflow.apache.org/docs/apache-airflow/stable/timezone.html
# for best practices
YESTERDAY = datetime.datetime.now() - datetime.timedelta(days=1)

default_args = {
    "owner": "Composer Example",
    "depends_on_past": False,
    "email": [""],
    "email_on_failure": False,
    "email_on_retry": False,
    "retries": 1,
    "retry_delay": datetime.timedelta(minutes=5),
    "start_date": YESTERDAY,
}

with models.DAG(
    "composer_quickstart",
    catchup=False,
    default_args=default_args,
    schedule_interval=datetime.timedelta(days=1),
) as dag:
    # Print the dag_run id from the Airflow logs
    print_dag_run_conf = bash.BashOperator(
        task_id="print_dag_run_conf", bash_command="echo {{ dag_run.id }}"
    )

Carica il file DAG nel bucket del tuo ambiente

Ogni ambiente Cloud Composer dispone di un associato a questo bucket. Airflow solo in pianificazioni di Cloud Composer I DAG che si trovano nella cartella /dags di questo bucket.

Per pianificare il DAG, carica quickstart.py dalla macchina locale al tuo cartella /dags dell'ambiente:

Per caricare quickstart.py con Google Cloud CLI, esegui questo comando in la cartella in cui si trova il file quickstart.py:

gcloud composer environments storage dags import \
--environment example-environment --location us-central1 \
--source quickstart.py

Visualizza il DAG

Dopo aver caricato il file DAG, Airflow fa quanto segue:

1. Analizza il file DAG che hai caricato. Potrebbero essere necessari alcuni minuti i DAG per renderli disponibili per Airflow.
2. Aggiunge il DAG all'elenco dei DAG disponibili.
3. Esegue il DAG in base alla pianificazione fornita nel file DAG.

Verifica che il DAG venga elaborato senza errori e che sia disponibile in Airflow e lo visualizzi nella UI dei DAG. La UI di DAG è l'interfaccia di Cloud Composer per la visualizzazione Informazioni sui DAG nella console Google Cloud. Cloud Composer offre inoltre accesso alla UI di Airflow, che è un ambiente web di Airflow nativo a riga di comando.

1. Attendi circa cinque minuti per concedere ad Airflow il tempo di elaborare il file DAG caricato in precedenza e per completare la prima esecuzione del DAG (come spiegato più avanti).

2. Esegui questo comando in Google Cloud CLI. Questo comando esegue dags list Comando dell'interfaccia a riga di comando di Airflow che elenca i DAG completamente gestito di Google Cloud.

   gcloud composer environments run example-environment \
   --location us-central1 \
   dags list
   

3. Verifica che il DAG composer_quickstart sia elencato nell'output del comando.

   Output di esempio:

   Executing the command: [ airflow dags list ]...
   Command has been started. execution_id=d49074c7-bbeb-4ee7-9b26-23124a5bafcb
   Use ctrl-c to interrupt the command
   dag_id              | filepath              | owner            | paused
   ====================+=======================+==================+=======
   airflow_monitoring  | airflow_monitoring.py | airflow          | False
   composer_quickstart | dag-quickstart-af2.py | Composer Example | False
   

   Nota: il DAG airflow_monitoring viene utilizzato da Cloud Composer per per monitorare l'integrità del tuo ambiente.

Visualizza dettagli esecuzione DAG

Una singola esecuzione di un DAG è chiamata esecuzione di DAG. Airflow immediatamente esegue un'esecuzione di DAG per il DAG di esempio perché la data di inizio nel file DAG è impostato su ieri. In questo modo, Airflow raggiunge i DAG specificati programmazione.

Il DAG di esempio contiene un'attività, print_dag_run_conf, che esegue echo nella console. Questo comando restituisce metadati sul DAG (Identificatore numerico dell'esecuzione di DAG).

Esegui questo comando in Google Cloud CLI. Questo comando elenca le esecuzioni di DAG per il DAG composer_quickstart:

gcloud composer environments run example-environment \
--location us-central1 \
dags list-runs -- --dag-id composer_quickstart

Output di esempio:

dag_id              | run_id                                      | state   | execution_date                   | start_date                       | end_date
====================+=============================================+=========+==================================+==================================+=================================
composer_quickstart | scheduled__2024-02-17T15:38:38.969307+00:00 | success | 2024-02-17T15:38:38.969307+00:00 | 2024-02-18T15:38:39.526707+00:00 | 2024-02-18T15:38:42.020661+00:00

Nota: puoi utilizzare il comando tasks states-for-dag-run per visualizzare lo stato delle singole attività.

L'interfaccia a riga di comando di Airflow non fornisce un comando per visualizzare i log delle attività. Puoi utilizzare la modalità Altri metodi per visualizzare i log delle attività di Airflow: UI DAG di Cloud Composer, UI di Airflow o Cloud Logging. Questa guida mostra un modo per eseguire query su Cloud Logging per i log di una specifica esecuzione del DAG.

Esegui questo comando in Google Cloud CLI. Questo comando legge i log da Cloud Logging per un'esecuzione specifica del DAG del DAG composer_quickstart. La L'argomento --format formatta l'output in modo che solo il testo del messaggio di log .

gcloud logging read \
--format="value(textPayload)" \
--order=asc \
"resource.type=cloud_composer_environment \
resource.labels.location=us-central1 \
resource.labels.environment_name=example-environment \
labels.workflow=composer_quickstart \
(labels.\"execution-date\"=\"RUN_ID\")"

Sostituisci:

• RUN_ID con il valore run_id dell'output del tasks states-for-dag-run che hai eseguito in precedenza. Per ad esempio 2024-02-17T15:38:38.969307+00:00.

Output di esempio:

...

Starting attempt 1 of 2
Executing <Task(BashOperator): print_dag_run_conf> on 2024-02-17
15:38:38.969307+00:00
Started process 22544 to run task

...

Running command: ['/usr/bin/bash', '-c', 'echo 115746']
Output:
115746

...

Command exited with return code 0
Marking task as SUCCESS. dag_id=composer_quickstart,
task_id=print_dag_run_conf, execution_date=20240217T153838,
start_date=20240218T153841, end_date=20240218T153841
Task exited with return code 0
0 downstream tasks scheduled from follow-on schedule check

Esegui la pulizia

Per evitare che al tuo account Google Cloud vengano addebitati costi per le risorse utilizzate in questa pagina, elimina il progetto Google Cloud Google Cloud.

Elimina le risorse utilizzate in questo tutorial:

1. Elimina l'ambiente Cloud Composer:

   1. Nella console Google Cloud, vai alla pagina Ambienti.

      Vai ad Ambienti

   2. Seleziona example-environment e fai clic su Elimina.

   3. Attendi fino a quando l'ambiente non viene eliminato.

2. Elimina il bucket dell'ambiente. Eliminazione di Cloud Composer di rete non elimina il proprio bucket.

   1. Nella console Google Cloud, vai a Storage > Pagina Browser.

      Vai ad Spazio di archiviazione > Browser

   2. Seleziona il bucket dell'ambiente e fai clic su Elimina. Ad esempio, questo il bucket può essere denominato us-central1-example-environ-c1616fe8-bucket.

3. Elimina il disco permanente dalla coda Redis del tuo ambiente. L'eliminazione del L'ambiente Cloud Composer non elimina il disco permanente.

   1. Nella console Google Cloud, vai a Compute Engine > Dischi.

      Vai a dischi

   2. Seleziona il disco permanente della coda Redis dell'ambiente e fai clic su Elimina.

      Ad esempio, questo disco può essere denominato pvc-02bc4842-2312-4347-8519-d87bdcd31115. Dischi per Cloud Composer 2 ha sempre il tipo Balanced persistent disk di 2 GB.

Passaggi successivi

• Aggiunta e aggiornamento dei DAG
• Scrittura dei DAG
• Ignorare le configurazioni Airflow
• Accesso alla UI di Airflow