Esegui un DAG Apache Airflow in Cloud Composer 3

Crea il account di servizio di un ambiente

Quando crei un ambiente, specifichi un account di servizio. Questo service account è chiamato service account dell'ambiente. Il tuo ambiente utilizza questo account di servizio per eseguire la maggior parte delle operazioni.

Il account di servizio per il tuo ambiente non è un account utente. Un account di servizio è un particolare tipo di account utilizzato da un'applicazione o da un'istanza di macchina virtuale (VM), non da una persona fisica.

Per creare un account di servizio per il tuo ambiente:

1. Crea un nuovo service account, come descritto nella documentazione di Identity and Access Management.

2. Assegna un ruolo, come descritto nella documentazione di Identity and Access Management. Il ruolo richiesto è Worker Composer (composer.worker).

Creazione di un ambiente

1. Nella console Google Cloud , vai alla pagina Crea ambiente.

   Vai a Crea ambiente

2. Nel campo Nome, inserisci example-environment.

3. Nell'elenco a discesa Località, seleziona una regione per l'ambiente Cloud Composer. Questa guida utilizza la regione us-central1.

4. Per le altre opzioni di configurazione dell'ambiente, utilizza i valori predefiniti forniti.

5. Fai clic su Crea e attendi la creazione dell'ambiente.

6. Al termine, accanto al nome dell'ambiente viene visualizzato un segno di spunta verde.

Creare un file DAG

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

Questa guida utilizza un DAG di Airflow di esempio 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 la configurazione dell'esecuzione del DAG utilizzando l'operatore bash.

Salva una copia del file quickstart.py sulla tua macchina 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 dell'ambiente

Ogni ambiente Cloud Composer ha un bucket Cloud Storage associato. Airflow in Cloud Composer pianifica solo i DAG che si trovano nella cartella /dags di questo bucket.

Per pianificare il DAG, carica quickstart.py dalla tua macchina locale nella cartella /dags del tuo ambiente:

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

   Vai ad Ambienti

2. Nell'elenco degli ambienti, fai clic sul nome del tuo ambiente, example-environment. Viene visualizzata la pagina Dettagli ambiente.

3. Fai clic su Apri cartella DAG. Viene visualizzata la pagina Dettagli bucket.

4. Fai clic su Carica file e seleziona la tua copia di quickstart.py.

5. Per caricare il file, fai clic su Apri.

Visualizzare il DAG

Dopo aver caricato il file DAG, Airflow esegue le seguenti operazioni:

1. Analizza il file DAG che hai caricato. Potrebbero essere necessari alcuni minuti prima che il DAG diventi disponibile 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 sia disponibile in Airflow visualizzandolo nell'interfaccia utente DAG. La UI DAG è l'interfaccia di Cloud Composer per visualizzare le informazioni sui DAG nella console Google Cloud . Cloud Composer fornisce anche l'accesso alla UI di Airflow, un'interfaccia web nativa di Airflow.

1. Attendi circa cinque minuti per dare ad Airflow il tempo di elaborare il file DAG che hai caricato in precedenza e di completare la prima esecuzione del DAG (spiegata in seguito).

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

   Vai ad Ambienti

3. Nell'elenco degli ambienti, fai clic sul nome del tuo ambiente, example-environment. Viene visualizzata la pagina Dettagli ambiente.

4. Vai alla scheda DAG.

5. Verifica che il DAG composer_quickstart sia presente nell'elenco dei DAG.

   

Visualizzare i dettagli dell'esecuzione del DAG

Una singola esecuzione di un DAG è detta esecuzione di DAG. Airflow esegue immediatamente un'esecuzione del DAG di esempio perché la data di inizio nel file DAG è impostata su ieri. In questo modo, Airflow recupera la pianificazione del DAG specificato.

Il DAG di esempio contiene un'attività, print_dag_run_conf, che esegue il comando echo nella console. Questo comando restituisce i metadati del DAG (l'identificatore numerico dell'esecuzione del DAG).

1. Nella scheda DAG, fai clic su composer_quickstart. Si apre la scheda Esecuzioni per il DAG.

2. Nell'elenco delle esecuzioni DAG, fai clic sulla prima voce.

   

3. Vengono visualizzati i dettagli dell'esecuzione del DAG, che forniscono informazioni sulle singole attività del DAG di esempio.

   

4. La sezione Log per l'esecuzione del DAG elenca i log per tutte le attività nell'esecuzione del DAG. Puoi visualizzare l'output del comando echo nei log.