Pianifica e attiva i DAG Airflow

Cloud Composer 1 | Cloud Composer 2 | Cloud Composer 3

Anteprima - Cloud Composer 3

Questa funzionalità è soggetta ai "Termini delle Offerte pre-GA" nella sezione Termini generali del servizio dei Termini specifici dei servizi. Le funzionalità pre-GA sono disponibili "così come sono" e potrebbero avere un supporto limitato. Per saperne di più, consulta le descrizioni della fase di lancio.

Questa pagina spiega come funzionano la pianificazione e l'attivazione DAG in Airflow, come definisci una pianificazione per un DAG e come attivare un DAG manualmente o metterlo in pausa.

Informazioni sui DAG Airflow in Cloud Composer

I DAG di Airflow in Cloud Composer vengono eseguiti in uno o più ambienti Cloud Composer nel progetto. Carichi i file di origine dei tuoi DAG Airflow in in un bucket Cloud Storage associato a un ambiente. L'istanza di Airflow dell'ambiente analizza quindi questi file e pianifica le esecuzioni dei DAG, come definito dalla pianificazione di ciascun DAG. Durante l'esecuzione di un DAG, Airflow pianifica ed esegue le singole attività che compongono un DAG in una sequenza definita dal DAG.

Per saperne di più sui concetti principali di Airflow come i DAG e le esecuzioni dei DAG, o operatori, consulta la pagina Concetti principali nella documentazione di Airflow.

Informazioni sulla pianificazione dei DAG in Airflow

Airflow fornisce i seguenti concetti per il suo meccanismo di pianificazione:

Data logica

Rappresenta una data in cui viene eseguita una determinata esecuzione di DAG.

Non si tratta della data effettiva in cui Airflow esegue un DAG, ma di un periodo di tempo che una determinata esecuzione del DAG deve elaborare. Ad esempio, per un DAG che programmato per essere eseguito ogni giorno alle 12:00, la data logica sarebbe anche le 12:00 in un giorno specifico. Poiché viene eseguito due volte al giorno, il periodo di tempo da elaborare è costituito dalle ultime 12 ore. Allo stesso tempo, la logica definita nel DAG stesso potrebbe non utilizzare affatto la data logica o l'intervallo di tempo. Ad esempio, un DAG potrebbe eseguire lo stesso script una volta al giorno senza utilizzare valore della data logica.

Nelle versioni di Airflow precedenti alla 2.2, questa data è chiamata data di esecuzione.

Data esecuzione

Rappresenta una data in cui viene eseguita una determinata esecuzione di DAG.

Ad esempio, per un DAG pianificato per l'esecuzione ogni giorno alle 12:00 l'esecuzione effettiva del DAG potrebbe avvenire alle 12:05, qualche tempo dopo passa una data logica.

Intervallo pianificazione

Rappresenta quando e con quale frequenza deve essere eseguito un DAG, in termini di date logiche.

Ad esempio, una pianificazione giornaliera indica che un DAG viene eseguito una volta al giorno e le date logiche per le relative esecuzioni hanno intervalli di 24 ore.

Data di inizio

Specifica quando vuoi che Airflow inizi a pianificare il DAG.

Le attività nel DAG possono avere date di inizio individuali oppure puoi specificare un'unica data di inizio per tutte le attività. In base alla data di inizio minima delle attività nel DAG e nell'intervallo pianificato, Airflow pianifica le esecuzioni dei DAG.

Recupero, backfill e nuovi tentativi

Meccanismi per eseguire le esecuzioni di DAG per date passate.

Il recupero esegue le esecuzioni dei DAG che non sono ancora state eseguite, ad esempio: se il DAG è stato messo in pausa per un lungo periodo di tempo e poi è stato riattivato. Puoi utilizzare il backfill per eseguire le esecuzioni dei DAG in un determinato intervallo di date. I tentativi specificano il numero di tentativi che Airflow deve eseguire durante l'esecuzione delle attività da un DAG.

La pianificazione funziona nel seguente modo:

1. Una volta trascorsa la data di inizio, Airflow attende la successiva occorrenza del intervallo di pianificazione.

2. Airflow pianifica la prima esecuzione del DAG alla fine di questo intervallo di pianificazione.

   Ad esempio, se un DAG è pianificato per l'esecuzione ogni ora e la data di inizio è alle 12:00 di oggi, la prima esecuzione del DAG avviene alle 13:00 di oggi.

La sezione Pianifica un DAG Airflow in questo documento descrive come configurare la pianificazione dei DAG utilizzando questi concetti. Per ulteriori informazioni su esecuzioni e pianificazione dei DAG, consulta Esecuzioni DAG nella documentazione di Airflow.

Informazioni sui modi per attivare un DAG

Airflow offre i seguenti modi per attivare un DAG:

• Attiva in base a una pianificazione. Airflow attiva il DAG automaticamente in base alla pianificazione specificata nel file DAG.

• Attivazione manuale. Puoi attivare manualmente un DAG da Console Google Cloud, UI di Airflow o esecuzione di un comando dell'interfaccia a riga di comando di Airflow da Google Cloud CLI.

• Si attiva in risposta agli eventi. Il modo standard per attivare un DAG in risposta agli eventi è utilizzare un sensore.

Altri modi per attivare i DAG:

• Si attiva in modo programmatico. Puoi attivare un DAG utilizzando l'API REST di Airflow. Ad esempio da uno script Python.

• Attivare ed eseguire il codice in modo programmatico in risposta agli eventi. Puoi attivare i DAG in la risposta agli eventi utilizzando Funzioni Cloud Run e l'API REST Airflow.

Prima di iniziare

• Assicurati che il tuo account abbia un ruolo che possa gestire gli oggetti nel di ambiente e visualizzare e attivare i DAG. Per ulteriori informazioni, consulta Controllo dell'accesso.

Pianifica un DAG Airflow

Devi definire una pianificazione per un DAG nel file DAG. Modifica la definizione del DAG in nel seguente modo:

1. Individua e modifica il file DAG sul computer. Se non hai il file DAG, puoi scaricare la relativa copia dal bucket dell'ambiente. Per un nuovo DAG, puoi definire tutti i parametri quando crei il file DAG.

2. Nel parametro schedule_interval, definisci la pianificazione. Puoi utilizzare un'espressione CRON, ad esempio 0 0 * * *, o un valore preimpostato, ad esempio @daily. Per Per ulteriori informazioni, consulta la sezione Cron e intervalli di tempo nella documentazione di Airflow.

   Airflow determina le date logiche per le esecuzioni dei DAG in base alla pianificazione impostata.

3. Nel parametro start_date, definisci la data di inizio.

   Airflow determina la data logica della prima esecuzione di DAG utilizzando questo .

4. (Facoltativo) Nel parametro catchup, definisci se Airflow deve eseguire tutte le esecuzioni precedenti di questo DAG dalla data di inizio alla data corrente che non sono state ancora eseguite.

   Le esecuzioni dei DAG eseguite durante il recupero avranno la data logica nei e la relativa data di esecuzione rifletterà l'ora in cui l'esecuzione del DAG è stata effettivamente eseguito.

   Nota: per eseguire il backfill delle esecuzioni di DAG per una determinata data anziché recuperare, esegui dags backfill Comando dell'interfaccia a riga di comando Airflow utilizzando Google Cloud CLI.

5. (Facoltativo) Nel parametro retries, definisci quante volte Airflow deve riprovare le attività non riuscite (ogni DAG è costituito da una o più attività individuali). Per impostazione predefinita, le attività in Cloud Composer vengono riprovate due volte.

6. Carica la nuova versione del DAG nel set di dati dell'ambiente di sincronizzare la directory di una VM con un bucket.

7. Attendi fino a quando Airflow non analizza correttamente il DAG. Ad esempio, puoi controllare l'elenco dei DAG nel tuo ambiente nella console Google Cloud o nell'interfaccia utente di Airflow.

La definizione di DAG di esempio seguente viene eseguita due volte al giorno alle 00:00 e alle 12:00. La data di inizio è impostata su 1° gennaio 2024, ma Airflow non la esegue per le date passate dopo il caricamento o la messa in pausa perché il recupero è disattivato.

Il DAG contiene un'attività denominata insert_query_job, che inserisce una riga in un con l'operatore BigQueryInsertJobOperator. Questo operatore è uno di Operatori BigQuery di Google Cloud, che puoi utilizzare per gestire set di dati e tabelle, eseguire query e convalidare i dati. Se una determinata esecuzione di questa attività non va a buon fine, Airflow ne riprova altre quattro volte con l'intervallo predefinito per i nuovi tentativi. La data logica per questi nuovi tentativi rimane la stessa.

La query SQL per questa riga utilizza Modelli Airflow per scrivere la data logica del DAG e il nome della riga.

import datetime

from airflow.models.dag import DAG
from airflow.providers.google.cloud.operators.bigquery import BigQueryInsertJobOperator

with DAG(
  "bq_example_scheduling_dag",
  start_date=datetime.datetime(2024, 1, 1),
  schedule_interval='0 */12 * * *',
  catchup=False
  ) as dag:

  insert_query_job = BigQueryInsertJobOperator(
    task_id="insert_query_job",
    retries=4,
    configuration={
        "query": {
            # schema: date (string), description (string)
            # example row: "20240101T120000", "DAG run: <DAG: bq_example_scheduling_dag>"
            "query": "INSERT example_dataset.example_table VALUES ('{{ ts_nodash }}', 'DAG run: {{ dag }}' )",
            "useLegacySql": False,
            "priority": "BATCH",
        }
    },
    location="us-central1"
  )

  insert_query_job

Per testare questo DAG, puoi attivarlo manualmente e e poi visualizza i log di esecuzione delle attività.

Altri esempi di parametri di pianificazione

I seguenti esempi di parametri di pianificazione illustrano il funzionamento della pianificazione con diverse combinazioni di parametri:

• Se start_date è datetime(2024, 4, 4, 16, 25) e schedule_interval è 30 16 * * *, la prima esecuzione del DAG verrà eseguita alle 16:30 del 5 aprile 2024.

• Se start_date è datetime(2024, 4, 4, 16, 35) e schedule_interval è 30 16 * * *, la prima esecuzione del DAG avviene alle 16:30 del 6 aprile 2024. Poiché la data di inizio è successiva all'intervallo di pianificazione del 4 aprile 2024, la relativa esecuzione non avviene il 5 aprile 2024. Invece, la programmazione l'intervallo termina alle 16:35 del 5 aprile 2024, quindi la prossima esecuzione del DAG è pianificata alle 16:30 del giorno successivo.

• Se start_date è datetime(2024, 4, 4) e schedule_interval è @daily, la prima esecuzione del DAG è programmata per le 00:00 del 5 aprile 2024.

• Se start_date è datetime(2024, 4, 4, 16, 30) e schedule_interval è 0 * * * *, quindi la prima esecuzione del DAG programmato per le 18:00 del 4 aprile 2024. Una volta trascorsi la data e l'ora specificati, Airflow pianifica un'esecuzione del DAG al minuto 0 di ogni ora. Il momento più vicino in cui si verifica è alle 17:00. A questo punto, Airflow pianifica un'esecuzione del DAG alla fine dell'intervallo pianificato, ovvero alle 18:00.

Attivare manualmente un DAG

Quando attivi manualmente un DAG Airflow, Airflow esegue il DAG una volta, indipendentemente dalla pianificazione specificata nel file DAG.

Console

Per attivare un DAG dalla console Google Cloud:

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

   Vai ad Ambienti

2. Seleziona un ambiente per visualizzarne i dettagli.

3. Nella pagina Dettagli dell'ambiente, vai alla scheda DAG.

4. Fai clic sul nome di un DAG.

5. Nella pagina Dettagli DAG, fai clic su Attiva DAG. Una nuova esecuzione di DAG è stato creato.

   Nota: non puoi specificare una configurazione per l'esecuzione del DAG da console Google Cloud.

Interfaccia utente di Airflow

Per attivare un DAG dall'interfaccia utente di Airflow:

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

   Vai ad Ambienti

2. Nella colonna Server web Airflow, segui il link Airflow per il tuo completamente gestito di Google Cloud.

3. Accedi con l'Account Google che dispone delle autorizzazioni appropriate.

4. Nell'interfaccia web di Airflow, nella pagina DAG, nella colonna Collegamenti per il DAG, fai clic sul pulsante Attiva DAG.

5. (Facoltativo) Specifica la configurazione dell'esecuzione del DAG.

6. Fai clic su Attiva.

gcloud

Esegui il comando dell'interfaccia a riga di comando di Airflow dags trigger:

  gcloud composer environments run ENVIRONMENT_NAME \
    --location LOCATION \
    dags trigger -- DAG_ID

Sostituisci quanto segue:

• ENVIRONMENT_NAME: il nome dell'ambiente.
• LOCATION: la regione in cui si trova l'ambiente.
• DAG_ID: il nome del DAG.

Per ulteriori informazioni sull'esecuzione dei comandi dell'interfaccia a riga di comando di Airflow in per gli ambienti Cloud Composer, consulta Esecuzione dei comandi dell'interfaccia a riga di comando di Airflow.

Per ulteriori informazioni sui comandi dell'interfaccia a riga di comando di Airflow disponibili, consulta consulta il riferimento per il comando gcloud composer environments run.

Visualizza i log di esecuzione dei DAG e i relativi dettagli

Nella console Google Cloud puoi:

• Visualizza gli stati delle esecuzioni di DAG passate e i dettagli dei DAG.
• Esplora i log dettagliati di tutte le esecuzioni e di tutte le attività di questi DAG.
• Visualizza le statistiche DAG.

Inoltre, Cloud Composer fornisce accesso UI di Airflow, che è l'interfaccia web di Airflow.

Nota: la pagina Orchestrazione in BigQuery combina le informazioni di pianificazione per tutti i DAG Airflow da tutti ambienti Cloud Composer 3 nel tuo progetto, oltre ad altri tipi di disponibili in BigQuery.

Metti in pausa un DAG

Attenzione: se hai definito il catchup per il DAG che viene eseguito su pianificazione, poi Airflow pianifica tutte le esecuzioni dei DAG non eseguite quando riattivi il DAG.

Console

Per mettere in pausa un DAG dalla console Google Cloud:

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

   Vai ad Ambienti

2. Seleziona un ambiente per visualizzarne i dettagli.

3. Nella pagina Dettagli dell'ambiente, vai alla scheda DAG.

4. Fai clic sul nome di un DAG.

5. Nella pagina Dettagli DAG, fai clic su Metti in pausa il DAG.

UI di Airflow

Per mettere in pausa un DAG dall'interfaccia utente di Airflow:

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

Vai ad Ambienti

1. Nella colonna Server web Airflow, segui il link Airflow per il tuo completamente gestito di Google Cloud.

2. Accedi con l'Account Google che dispone delle autorizzazioni appropriate.

3. Nell'interfaccia web di Airflow, nella pagina DAG, fai clic sul pulsante di attivazione/disattivazione accanto al nome del DAG.

gcloud

Esegui il comando dags pause dell'interfaccia a riga di comando Airflow:

  gcloud composer environments run ENVIRONMENT_NAME \
    --location LOCATION \
    dags pause -- DAG_ID

Sostituisci quanto segue:

• ENVIRONMENT_NAME: il nome dell'ambiente.
• LOCATION: la regione in cui si trova l'ambiente.
• DAG_ID: il nome del DAG.

Per ulteriori informazioni sull'esecuzione dei comandi dell'interfaccia a riga di comando di Airflow in per gli ambienti Cloud Composer, consulta Esecuzione dei comandi dell'interfaccia a riga di comando di Airflow.

Per ulteriori informazioni sui comandi dell'interfaccia a riga di comando di Airflow disponibili, consulta consulta il riferimento per il comando gcloud composer environments run.

Passaggi successivi

• Scrivi i DAG Airflow
• Testare i DAG di Airflow
• Risolvere i problemi dei DAG di Airflow