Test dei DAG di Airflow

Airflow gestito (terza generazione) | Airflow gestito (seconda generazione) | Airflow gestito (prima generazione legacy)

Prima di eseguire il deployment dei DAG in produzione, puoi eseguire i sottocomandi della CLI di Airflow per analizzare il codice DAG nello stesso contesto in cui viene eseguito il DAG.

Test durante la creazione dei DAG

Puoi eseguire una singola istanza di attività in locale e visualizzare l'output dei log. La visualizzazione dell'output ti consente di verificare la presenza di errori di sintassi e di attività. Il test in locale non controlla le dipendenze né comunica lo stato al database.

Ti consigliamo di inserire i DAG in una cartella data/test nell'ambiente di test.

Creare una directory di test

Nel bucket dell'ambiente, crea una directory di test e copia i DAG al suo interno.

gcloud storage cp BUCKET_NAME/dags \
  BUCKET_NAME/data/test --recursive

Sostituisci quanto segue:

  • BUCKET_NAME: il nome del bucket associato all'ambiente Airflow gestito.

Esempio:

gcloud storage cp gs://us-central1-example-environment-a12bc345-bucket/dags \
  gs://us-central1-example-environment-a12bc345-bucket/data/test --recursive

Per ulteriori informazioni sul caricamento dei DAG, vedi Aggiungere e aggiornare i DAG.

Controllare la presenza di errori di sintassi

Per verificare la presenza di errori di sintassi nei DAG caricati nella cartella /data/test, inserisci il seguente comando gcloud:

Airflow 2 

gcloud composer environments run \
  ENVIRONMENT_NAME \
  --location ENVIRONMENT_LOCATION \
  dags list -- --subdir /home/airflow/gcs/data/test

Airflow 1 

gcloud composer environments run \
  ENVIRONMENT_NAME \
  --location ENVIRONMENT_LOCATION \
  list_dags -- -sd /home/airflow/gcs/data/test

Sostituisci quanto segue:

  • ENVIRONMENT_NAME: il nome dell'ambiente.
  • ENVIRONMENT_LOCATION: la regione in cui si trova l'ambiente.

Verificare la presenza di errori nelle attività

Per verificare la presenza di errori specifici delle attività nei DAG caricati nella cartella /data/test, esegui il seguente comando gcloud:

Airflow 2 

gcloud composer environments run \
  ENVIRONMENT_NAME \
  --location ENVIRONMENT_LOCATION \
  tasks test -- --subdir /home/airflow/gcs/data/test \
  DAG_ID TASK_ID \
  DAG_EXECUTION_DATE

Airflow 1 

gcloud composer environments run \
  ENVIRONMENT_NAME \
  --location ENVIRONMENT_LOCATION \
  test -- -sd /home/airflow/gcs/data/test DAG_ID \
  TASK_ID DAG_EXECUTION_DATE

Sostituisci quanto segue:

  • ENVIRONMENT_NAME: il nome dell'ambiente.
  • ENVIRONMENT_LOCATION: la regione in cui si trova l'ambiente.
  • DAG_ID: l'ID del DAG.
  • TASK_ID: l'ID dell'attività.
  • DAG_EXECUTION_DATE: la data di esecuzione del DAG. Questa data viene utilizzata per la creazione di modelli. Indipendentemente dalla data specificata qui, il DAG viene eseguito immediatamente.

Esempio:

Airflow 2 

gcloud composer environments run \
  example-environment \
  --location us-central1 \
  tasks test -- --subdir /home/airflow/gcs/data/test \
  hello_world print_date 2021-04-22

Airflow 1 

gcloud composer environments run example-environment \
  --location us-central1 \
  test -- -sd /home/airflow/gcs/data/test \
  hello_world print_date 2021-04-22

Aggiornare e testare un DAG di cui è stato eseguito il deployment

Per testare gli aggiornamenti dei DAG nell'ambiente di test:

  1. Copia il DAG di cui è stato eseguito il deployment che vuoi aggiornare in data/test.
  2. Aggiorna il DAG.
  3. Testa il DAG.
    1. Controlla la presenza di errori di sintassi.
    2. Verifica la presenza di errori specifici delle attività.
  4. Assicurati che il DAG venga eseguito correttamente.
  5. Disattiva il DAG nell'ambiente di test.
    1. Vai alla pagina Interfaccia utente di Airflow > DAG.
    2. Se il DAG che stai modificando viene eseguito costantemente, disattivalo.
    3. Per velocizzare le attività in sospeso, fai clic sull'attività e su Segna come riuscita.
  6. Esegui il deployment del DAG nell'ambiente di produzione.
    1. Disattiva il DAG nell'ambiente di produzione.
    2. Carica il DAG aggiornato nella cartella dags/ nell'ambiente di produzione.

Domande frequenti sui test dei DAG

Come faccio a isolare le esecuzioni dei DAG negli ambienti di produzione e di test?

Ad esempio, Airflow ha un repository globale di codice sorgente nella cartella dags/ che tutte le esecuzioni dei DAG condividono. Vuoi aggiornare il codice sorgente in produzione o in test senza interferire con i DAG in esecuzione.

Airflow non fornisce un isolamento DAG efficace. Ti consigliamo di mantenere ambienti Managed Airflow di produzione e di test separati per evitare che i DAG di test interferiscano con i DAG di produzione.

Come faccio a evitare interferenze tra i DAG quando eseguo test di integrazione da rami GitHub diversi?

Utilizza nomi di attività univoci per evitare interferenze. Ad esempio, puoi anteporre l'ID del ramo agli ID delle attività.

Qual è una best practice per i test di integrazione con Airflow?

Ti consigliamo di utilizzare un ambiente dedicato per i test di integrazione con Airflow. Un modo per segnalare la riuscita dell'esecuzione del DAG è scrivere in un file in una cartella Cloud Storage e poi controllare il contenuto nei tuoi scenari di test di integrazione.

Come faccio a collaborare in modo efficiente con altri contributori di DAG?

Ogni contributore può avere una sottodirectory nella cartella data/ per lo sviluppo.

I DAG aggiunti alla cartella data/ non vengono rilevati automaticamente dallo scheduler o dal server web di Airflow

I contributori di DAG possono creare esecuzioni manuali di DAG utilizzando il comando gcloud composer environments run e il sottocomando test con il flag --subdir per specificare la directory di sviluppo del contributore.

Ad esempio:

Airflow 2 

gcloud composer environments run test-environment-name \
  tasks test -- dag-id task-id execution-date \
  --subdir /home/airflow/gcs/data/alice_dev

Airflow 1 

gcloud composer environments run test-environment-name \
  test -- dag-id task-id execution-date \
  --subdir /home/airflow/gcs/data/alice_dev

Come faccio a mantenere sincronizzati gli ambienti di deployment e di produzione?

Per gestire l'accesso:

  • Per l'autenticazione, utilizza gli account di servizio .

  • Per il controllo dell'accesso, utilizza Identity and Access Management e i ruoli e le autorizzazioni di Managed Airflow .

Per eseguire il deployment dallo sviluppo alla produzione:

  • Garantisci una configurazione coerente, ad esempio variabili di ambiente e pacchetti PyPI.

  • Garantisci argomenti DAG coerenti. Per evitare di codificare in modo rigido, ti consigliamo di utilizzare macro e variabili di Airflow.

    Ad esempio:

    Airflow 2 

    gcloud composer environments run test-environment-name \
  variables set -- DATA_ENDPOINT_KEY DATA_ENDPOINT_VALUE

    Airflow 1 

    gcloud composer environments run test-environment-name \
  variables -- --set DATA_ENDPOINT_KEY DATA_ENDPOINT_VALUE

Passaggi successivi