Aggiungi e aggiorna i DAG

Cloud Composer 3 | Cloud Composer 2 | Cloud Composer 1

Questa pagina descrive come gestire i DAG nell'ambiente Cloud Composer.

Cloud Composer utilizza un bucket Cloud Storage per archiviare i DAG del tuo ambiente Cloud Composer. Il tuo ambiente sincronizza i DAG da questo bucket con i componenti Airflow come i worker e gli scheduler di Airflow.

Prima di iniziare

• Poiché Apache Airflow non fornisce un forte isolamento dei DAG, ti consigliamo di mantenere ambienti di produzione e test separati per evitare interferenze tra i DAG. Per ulteriori informazioni, consulta Testare i DAG.
• Assicurati che il tuo account disponga di autorizzazioni sufficienti per gestire i DAG.
• Le modifiche al DAG vengono propagate ad Airflow entro 3-5 minuti. Puoi vedere lo stato dell'attività nell'interfaccia web di Airflow.

Accedere al bucket del tuo ambiente

Per accedere al bucket associato al tuo ambiente:

gcloud composer environments describe ENVIRONMENT_NAME \     --location LOCATION \     --format="get(config.dagGcsPrefix)" 

gcloud beta composer environments describe example-environment \     --location us-central1 \     --format="get(config.dagGcsPrefix)" 

GET https://composer.googleapis.com/v1/projects/example-project/ locations/us-central1/environments/example-environment 

import google.auth import google.auth.transport.requests  # Authenticate with Google Cloud. # See: https://cloud.google.com/docs/authentication/getting-started credentials, _ = google.auth.default(     scopes=["https://www.googleapis.com/auth/cloud-platform"] ) authed_session = google.auth.transport.requests.AuthorizedSession(credentials)  # project_id = 'YOUR_PROJECT_ID' # location = 'us-central1' # composer_environment = 'YOUR_COMPOSER_ENVIRONMENT_NAME'  environment_url = (     "https://composer.googleapis.com/v1beta1/projects/{}/locations/{}"     "/environments/{}" ).format(project_id, location, composer_environment) response = authed_session.request("GET", environment_url) environment_data = response.json()  # Print the bucket name from the response body. print(environment_data["config"]["dagGcsPrefix"])

Aggiungere o aggiornare un DAG

Per aggiungere o aggiornare un DAG, sposta il file Python .py del DAG nella cartella /dags del bucket dell'ambiente.

gcloud composer environments storage dags import \     --environment ENVIRONMENT_NAME \     --location LOCATION \     --source="LOCAL_FILE_TO_UPLOAD" 

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

Aggiorna un DAG con esecuzioni di DAG attive

Se aggiorni un DAG con esecuzioni attive:

• Tutte le attività attualmente in esecuzione vengono completate utilizzando il file DAG originale.
• Tutte le attività pianificate ma non attualmente in esecuzione utilizzano il file DAG aggiornato.
• Tutte le attività non più presenti nel file DAG aggiornato vengono contrassegnate come rimosse.

Aggiornare i DAG eseguiti con una pianificazione frequente

Dopo aver caricato un file DAG, Airflow impiega un po' di tempo per caricare il file e aggiornare il DAG. Se il DAG viene eseguito con una pianificazione frequente, ti consigliamo di assicurarti che utilizzi la versione aggiornata del file DAG. Ecco come fare:

1. Metti in pausa il DAG nell'interfaccia utente di Airflow.

2. Carica un file DAG aggiornato.

3. Attendi finché non visualizzi gli aggiornamenti nell'interfaccia utente di Airflow. Ciò significa che il DAG è stato analizzato correttamente dallo scheduler e aggiornato nel database Airflow.

   Se l'interfaccia utente di Airflow mostra i DAG aggiornati, ciò non garantisce che i worker di Airflow dispongano della versione aggiornata del file DAG. Ciò accade perché i file DAG vengono sincronizzati in modo indipendente per gli scheduler e i worker.

4. Potresti voler estendere il tempo di attesa per assicurarti che il file DAG venga sincronizzato con tutti i worker del tuo ambiente. La sincronizzazione avviene diverse volte al minuto. In un ambiente sano, l'attesa di circa 20-30 secondi è sufficiente per la sincronizzazione di tutti i worker.

5. (Facoltativo) Se vuoi assicurarti che tutti i worker abbiano la nuova versione del file DAG, esamina i log di ogni singolo worker. Ecco come fare:

   1. Apri la scheda Log per il tuo ambiente nella console Google Cloud .

   2. Vai a Log di Composer > Infrastruttura > Sincronizzazione di Cloud Storage e controlla i log di ogni worker nel tuo ambiente. Cerca la voce di log Syncing dags directory più recente con un timestamp successivo al caricamento del nuovo file DAG. Se vedi un elemento Finished syncing che lo segue, significa che i DAG sono stati sincronizzati correttamente su questo worker.

6. Riattiva il DAG.

Eliminare un DAG nell'ambiente

Per eliminare un DAG, rimuovi il file Python .py per il DAG dalla cartella /dags dell'ambiente nel bucket dell'ambiente.

gcloud composer environments storage dags delete \     --environment ENVIRONMENT_NAME \     --location LOCATION \     DAG_FILE 

gcloud composer environments storage dags delete \     --environment example-environment \     --location us-central1 \     example_dag.py 

Rimuovere un DAG dall'UI di Airflow

Per rimuovere i metadati di un DAG dall'interfaccia web di Airflow:

  gcloud composer environments run ENVIRONMENT_NAME \     --location LOCATION \     delete_dag -- DAG_NAME 

  gcloud composer environments run ENVIRONMENT_NAME \     --location LOCATION \     dags delete -- DAG_NAME 

Passaggi successivi

• Scrivere DAG
• Testare, sincronizzare ed eseguire il deployment dei DAG da GitHub
• Risoluzione dei problemi dei DAG