Cloud Composer 1 | Cloud Composer 2
Questo tutorial illustra i passaggi per eseguire il debug di un DAG Airflow non riuscito in Cloud Composer e diagnosticare i problemi relativi alle risorse dei worker, come la mancanza di memoria o di spazio di archiviazione del worker, con l'aiuto dei log e del monitoraggio dell'ambiente.
Introduzione
Questo tutorial è incentrato sui problemi relativi alle risorse per dimostrare come eseguire il debug di un DAG.
La mancanza di risorse worker allocate causa errori dei DAG. Se un'attività Airflow esaurisce la memoria o lo spazio di archiviazione, potresti vedere un'eccezione di Airflow, ad esempio:
WARNING airflow.exceptions.AirflowException: Task received SIGTERM signal
INFO - Marking task as FAILED.
o
Task exited with return code Negsignal.SIGKILL
In questi casi, si consiglia di aumentare le risorse worker di Airflow o ridurre il numero di attività per worker. Tuttavia, poiché le eccezioni di Airflow possono essere generiche, potrebbe essere difficile identificare la risorsa specifica che causa il problema.
Questo tutorial spiega come diagnosticare il motivo di un errore dei DAG e identificare il tipo di risorsa che causa i problemi eseguendo il debug di due DAG di esempio che non funzionano a causa della mancanza di memoria e spazio di archiviazione dei worker.
Obiettivi
Esegui DAG di esempio che non riescono per i seguenti motivi:
- Mancanza di memoria dei lavoratori
- Mancanza di spazio di archiviazione del worker
Diagnosi dei motivi dell'errore
Aumenta le risorse worker allocate
Testare i DAG con nuovi limiti di risorse
Costi
Questo tutorial utilizza i seguenti componenti fatturabili di Google Cloud:
Al termine di questo tutorial, puoi evitare di continuare la fatturazione eliminando le risorse che hai creato. Per maggiori dettagli, vedi Pulizia.
Prima di iniziare
Questa sezione descrive le azioni necessarie prima di iniziare il tutorial.
Creare e configurare un progetto
Per questo tutorial è necessario un progetto Google Cloud. Configura il progetto nel seguente modo:
Nella console Google Cloud, seleziona o crea un progetto:
Verifica che la fatturazione sia attivata per il tuo progetto. Scopri come verificare se la fatturazione è abilitata in un progetto.
Assicurati che l'utente del progetto Google Cloud disponga dei seguenti ruoli per creare le risorse necessarie:
- Amministratore ambienti e oggetti Storage
(
roles/composer.environmentAndStorageObjectAdmin) - Amministratore Compute (
roles/compute.admin) - Editor di Monitoring (
roles/monitoring.editor)
- Amministratore ambienti e oggetti Storage
(
Abilita le API per il tuo progetto
Attiva l'API Cloud Composer.
Crea il tuo ambiente Cloud Composer
Crea un ambiente Cloud Composer 2.
Durante la creazione dell'ambiente, concedi il ruolo Estensione agente di servizio API Cloud Composer v2 (roles/composer.ServiceAgentV2Ext) all'account agente di servizio Composer. Cloud Composer utilizza questo account per eseguire operazioni
nel tuo progetto Google Cloud.
Controlla i limiti delle risorse worker
Controlla i limiti delle risorse worker di Airflow nel tuo ambiente:
Nella console Google Cloud, vai alla pagina Ambienti.
Nell'elenco degli ambienti, fai clic sul nome del tuo ambiente. Viene visualizzata la pagina Dettagli ambiente.
Vai alla scheda Configurazione dell'ambiente.
Vai a Risorse > Configurazione dei carichi di lavoro > Worker.
Verifica che i valori siano 0,5 vCPU, 1,875 GB di memoria e 1 GB di spazio di archiviazione. Questi sono i limiti delle risorse worker di Airflow con cui lavorerai nei passaggi successivi di questo tutorial.
Esempio: diagnostica i problemi di esaurimento della memoria
Carica il DAG di esempio riportato di seguito nell'ambiente creato nei passaggi precedenti. In questo tutorial, questo DAG è denominato
create_list_with_many_strings.
Questo DAG contiene un'attività che esegue i seguenti passaggi:
- Crea un elenco vuoto
s. - Esegue un ciclo per aggiungere la stringa
Moreall'elenco. - Stampa la quantità di memoria consumata dall'elenco e attende 1 secondo in ogni iterazione di 1 minuto.
import time
import airflow
from airflow import DAG
from airflow.operators.python_operator import PythonOperator
import sys
from datetime import timedelta
default_args = {
'start_date': airflow.utils.dates.days_ago(0),
'retries': 0,
'retry_delay': timedelta(minutes=10)
}
dag = DAG(
'create_list_with_many_strings',
default_args=default_args,
schedule_interval=None)
def consume():
s = []
for i in range(120):
for j in range(1000000):
s.append("More")
print(f"i={i}; size={sys.getsizeof(s) / (1000**3)}GB")
time.sleep(1)
t1 = PythonOperator(
task_id='task0',
python_callable=consume,
dag=dag,
depends_on_past=False,
retries=0
)
Attiva il DAG di esempio
Attiva il DAG di esempio, create_list_with_many_strings:
Nella console Google Cloud, vai alla pagina Ambienti.
Nella colonna Airflow webserver, segui il link Airflow per il tuo ambiente.
Nell'interfaccia web di Airflow, nella colonna Link del DAG della pagina DAG, fai clic sul pulsante Evento di attivazione.
Fai clic su Attivatore.
Nella pagina DAG, fai clic sull'attività attivata ed esamina i log di output per assicurarti che il DAG sia in esecuzione.
Mentre l'attività è in esecuzione, i log di output stamperanno le dimensioni della memoria in GB utilizzate dal DAG.
Dopo diversi minuti, l'attività non riuscirà perché supera il limite di memoria worker di Airflow di 1,875 GB.
Diagnosi del DAG non riuscito
Se eseguivi più attività al momento dell'errore, valuta la possibilità di eseguire una sola attività e diagnosticare la pressione delle risorse durante quel periodo per identificare quali attività causano la pressione delle risorse e quali devi aumentare.
Esamina i log delle attività Airflow
Osserva che l'attività del DAG create_list_with_many_strings ha uno stato Failed.
Esamina i log dell'attività. Verrà visualizzata la seguente voce di log:
```none
{local_task_job.py:102} INFO - Task exited with return code
Negsignal.SIGKILL
```
`Netsignal.SIGKILL` might be an indication of your task using more memory
than the Airflow worker is allocated. The system sends
the `Negsignal.SIGKILL` signal to avoid further memory consumption.
Rivedi carichi di lavoro
Esamina i carichi di lavoro per verificare che il carico dell'attività non causi il superamento del limite di consumo di memoria da parte del nodo in cui viene eseguito il pod:
Nella console Google Cloud, vai alla pagina Ambienti.
Nell'elenco degli ambienti, fai clic sul nome del tuo ambiente. Viene visualizzata la pagina Dettagli ambiente.
Vai alla scheda Configurazione dell'ambiente.
In Risorse > Cluster GKE > Carichi di lavoro, fai clic su Visualizza carichi di lavoro del cluster.
Verifica se alcuni pod dei carichi di lavoro hanno stati simili ai seguenti:
Error with exit code 137 and 1 more issue. ContainerStatusUnknown with exit code 137 and 1 more issueExit code 137indica che un container o un pod sta cercando di utilizzare più memoria di quella consentita. Il processo viene interrotto per impedire l'utilizzo della memoria.
Esamina l'integrità dell'ambiente e il monitoraggio del consumo di risorse
Esamina l'integrità dell'ambiente e il monitoraggio del consumo di risorse:
Nella console Google Cloud, vai alla pagina Ambienti.
Nell'elenco degli ambienti, fai clic sul nome del tuo ambiente. Viene visualizzata la pagina Dettagli ambiente.
Vai alla scheda Monitoring e seleziona Panoramica.
Nel riquadro Panoramica dell'ambiente, individua il grafico Environment Health (DAG Monitoring Airflow Monitoring). Contiene un'area rossa che corrisponde all'ora in cui sono stati avviati gli errori di stampa dei log.
Seleziona Worker e trova il grafico Utilizzo memoria totale dei worker. Osserva che la riga Utilizzo memoria ha un picco nel momento in cui l'attività era in esecuzione.
Anche se la linea di utilizzo della memoria sul grafico non raggiunge il limite, quando si diagnostica i motivi dell'errore, è necessario considerare solo la memoria allocabile, mentre la riga Limite di memoria sul grafico rappresenta la memoria totale disponibile (inclusa la capacità prenotata da GKE).
In questo esempio, il limite di memoria del worker è impostato su 1,875 GB. GKE riserva il 25% dei primi 4 GiB di memoria. GKE riserva inoltre un'ulteriore soglia di rimozione: 100 MiB di memoria su ciascun nodo per l'eliminazione di kubelet.
La memoria allocabile viene calcolata nel seguente modo:
ALLOCATABLE = CAPACITY - RESERVED - EVICTION-THRESHOLD
Se il limite di memoria è 1,875 GB, la memoria allocabile effettiva è:
1.75 GiB (1.875GB) - 0.44 (25% GiB reserved) - 0.1 = 1.21 GiB (~1.3 GB).
Quando aggiungi questo limite effettivo al grafico di utilizzo della memoria, vedrai che il picco di utilizzo della memoria dell'attività raggiunge il limite di memoria effettivo e puoi concludere che l'attività non è riuscita a causa di memoria worker insufficiente.
Aumenta il limite di memoria del worker
Alloca memoria aggiuntiva del worker in modo che il DAG di esempio abbia esito positivo:
Nella console Google Cloud, vai alla pagina Ambienti.
Nell'elenco degli ambienti, fai clic sul nome del tuo ambiente. Viene visualizzata la pagina Dettagli ambiente.
Vai alla scheda Configurazione dell'ambiente.
Trova la configurazione Risorse > Carichi di lavoro e fai clic su Modifica.
Nel campo Memoria della sezione Worker, specifica il nuovo limite di memoria per i worker di Airflow. In questo tutorial, utilizza 3 GB.
Salva le modifiche e attendi qualche minuto per il riavvio dei worker di Airflow.
Testa il DAG con il nuovo limite di memoria
Attiva di nuovo il DAG create_list_with_many_strings e attendi che venga completato.
Nei log di output dell'esecuzione dei DAG, vedrai
Marking task as SUCCESSe lo stato dell'attività indicherà Success.Consulta la sezione Panoramica dell'ambiente nella scheda Monitoraggio e assicurati che non ci siano aree rosse.
Fai clic sulla sezione Worker e trova il grafico Utilizzo memoria totale dei worker. Vedrai che la riga Limite di memoria riflette la modifica del limite di memoria, mentre la riga Utilizzo memoria è di gran lunga inferiore al limite di memoria allocabile effettivo.
Esempio: diagnosticare i problemi di esaurimento dello spazio di archiviazione
In questo passaggio devi caricare due DAG che creano file di grandi dimensioni. Il primo DAG crea un file di grandi dimensioni. Il secondo DAG crea un file di grandi dimensioni e imita un'operazione a lunga esecuzione.
Le dimensioni del file in entrambi i DAG superano il limite di archiviazione predefinito del worker Airflow di 1 GB, ma il secondo DAG ha un'attività di attesa aggiuntiva per estenderne artificialmente la durata.
Nei passaggi successivi esaminerai le differenze nel comportamento di entrambi i DAG.
Carica un DAG che crea un file di grandi dimensioni
Carica il DAG di esempio riportato di seguito nell'ambiente creato nei passaggi precedenti. In questo tutorial, questo DAG è denominato
create_large_txt_file_print_logs.
Questo DAG contiene un'attività che esegue i seguenti passaggi:
- Scrive un file
localfile.txtda 1,5 GB nello spazio di archiviazione worker di Airflow. - Stampa le dimensioni del file creato utilizzando il modulo Python
os. - Stampa la durata dell'esecuzione di DAG ogni minuto.
import airflow
from airflow import DAG
from airflow.operators.python_operator import PythonOperator
import os
from datetime import timedelta
import time
default_args = {
'start_date': airflow.utils.dates.days_ago(0),
'retries': 0,
'retry_delay': timedelta(minutes=10)
}
dag = DAG(
'create_large_txt_file_print_logs',
default_args=default_args,
schedule_interval=None)
def consume():
size = 1000**2 # bytes in 1 MB
amount = 100
def create_file():
print(f"Start creating a huge file")
with open("localfile.txt", "ab") as f:
for j in range(15):
f.write(os.urandom(amount) * size)
print("localfile.txt size:", os.stat("localfile.txt").st_size / (1000**3), "GB")
create_file()
print("Success!")
t1 = PythonOperator(
task_id='create_huge_file',
python_callable=consume,
dag=dag,
depends_on_past=False,
retries=0)
Carica un DAG che crea un file di grandi dimensioni in un'operazione a lunga esecuzione
Per imitare un DAG di lunga durata ed esaminare l'impatto della durata dell'attività sullo stato di fine, carica il secondo DAG di esempio nel tuo ambiente. In questo tutorial, questo DAG è denominato
long_running_create_large_txt_file_print_logs.
Questo DAG contiene un'attività che esegue i seguenti passaggi:
- Scrive un file
localfile.txtda 1,5 GB nello spazio di archiviazione worker di Airflow. - Stampa le dimensioni del file creato utilizzando il modulo Python
os. - Attende 1 ora e 15 minuti per imitare il tempo necessario per le operazioni con il file, ad esempio la lettura del file.
- Stampa la durata dell'esecuzione di DAG ogni minuto.
import airflow
from airflow import DAG
from airflow.operators.python_operator import PythonOperator
import os
from datetime import timedelta
import time
default_args = {
'start_date': airflow.utils.dates.days_ago(0),
'retries': 0,
'retry_delay': timedelta(minutes=10)
}
dag = DAG(
'long_running_create_large_txt_file_print_logs',
default_args=default_args,
schedule_interval=None)
def consume():
size = 1000**2 # bytes in 1 MB
amount = 100
def create_file():
print(f"Start creating a huge file")
with open("localfile.txt", "ab") as f:
for j in range(15):
f.write(os.urandom(amount) * size)
print("localfile.txt size:", os.stat("localfile.txt").st_size / (1000**3), "GB")
create_file()
for k in range(75):
time.sleep(60)
print(f"{k+1} minute")
print("Success!")
t1 = PythonOperator(
task_id='create_huge_file',
python_callable=consume,
dag=dag,
depends_on_past=False,
retries=0)
Attiva DAG di esempio
Attiva il primo DAG, create_large_txt_file_print_logs:
Nella console Google Cloud, vai alla pagina Ambienti.
Nella colonna Airflow webserver, segui il link Airflow per il tuo ambiente.
Nell'interfaccia web di Airflow, nella colonna Link del DAG della pagina DAG, fai clic sul pulsante Evento di attivazione.
Fai clic su Attivatore.
Nella pagina DAG, fai clic sull'attività attivata ed esamina i log di output per assicurarti che il DAG sia in esecuzione.
Attendi il completamento dell'attività creata con il DAG
create_large_txt_file_print_logs. Questa operazione potrebbe richiedere alcuni minuti.Nella pagina DAG, fai clic sull'esecuzione di DAG. Vedrai che l'attività è in stato
Success, anche se è stato superato il limite di spazio di archiviazione.
Esamina i log di Airflow dell'attività:
Nella console Google Cloud, vai alla pagina Ambienti.
Nell'elenco degli ambienti, fai clic sul nome del tuo ambiente. Viene visualizzata la pagina Dettagli ambiente.
Vai alla scheda Log, quindi vai a Tutti i log > Log di Airflow > Worker > Visualizza in Esplora log.
Filtra i log per tipo: mostra solo i messaggi di tipo Errore.
Nei log vedrai messaggi simili ai seguenti:
Worker: warm shutdown (Main Process)
o
A worker pod was evicted at 2023-12-01T12:30:05Z with message: Pod ephemeral
local storage usage exceeds the total limit of containers 1023Mi.
Questi log indicano che il pod ha avviato il processo di "arresto a caldo" perché lo spazio di archiviazione utilizzato ha superato il limite ed è stato rimosso dopo un'ora. Tuttavia, l'esecuzione del DAG non è riuscita perché è stata completata durante il periodo di tolleranza per la terminazione di Kubernetes, spiegato più dettagliatamente in questo tutorial.
Per illustrare il concetto di periodo di tolleranza per la risoluzione, esamina il risultato del secondo DAG di esempio, long_running_create_large_txt_file_print_logs.
Attiva il secondo DAG, long_running_create_large_txt_file_print_logs:
Nella console Google Cloud, vai alla pagina Ambienti.
Nella colonna Airflow webserver, segui il link Airflow per il tuo ambiente.
Nell'interfaccia web di Airflow, nella colonna Link del DAG della pagina DAG, fai clic sul pulsante Evento di attivazione.
Fai clic su Attivatore.
Nella pagina DAG, fai clic sull'attività attivata ed esamina i log di output per assicurarti che il DAG sia in esecuzione.
Attendi che l'esecuzione del DAG
long_running_create_large_txt_file_print_logsnon vada a buon fine. L'operazione richiederà circa un'ora.
Esamina i risultati dell'esecuzione di DAG:
Nella pagina DAG, fai clic sull'esecuzione
long_running_create_large_txt_file_print_logsdi DAG. Vedrai che l'attività ha uno statoFailede che la durata dell'esecuzione è stata esattamente 1 ora e 5 minuti, ovvero inferiore al periodo di attesa dell'attività di 1 ora e 15 minuti.Esamina i log dell'attività. Dopo che il DAG ha creato il file
localfile.txtnel container del worker Airflow, il log mostra che il DAG è stato avviato in attesa e la durata di esecuzione viene stampata nei log delle attività ogni minuto. In questo esempio, il DAG stampa il loglocalfile.txt size:e la dimensione del filelocalfile.txtsarà di 1,5 GB.
Quando il file scritto nel container del worker Airflow supera il limite di archiviazione, l'esecuzione del DAG dovrebbe non riuscire. Tuttavia, l'attività non viene completata immediatamente e continua a essere eseguita finché la sua durata non raggiunge 1 ora e 5 minuti. Questo accade perché Kubernetes non termina immediatamente l'attività e continua a essere in esecuzione per consentire 1 ora di tempo per il ripristino, noto come "periodo di tolleranza della terminazione". Quando un nodo esaurisce le risorse, Kubernetes non termina immediatamente il pod per gestire agevolmente la terminazione, il che riduce l'impatto sull'utente finale.
Il periodo di tolleranza per la terminazione consente agli utenti di recuperare i file dopo errori delle attività, ma può creare confusione durante la diagnosi dei DAG. Quando il limite di spazio di archiviazione del worker di Airflow viene superato, lo stato finale dell'attività dipende dalla durata dell'esecuzione dell'DAG:
Se l'esecuzione del DAG supera il limite di spazio di archiviazione del worker, ma viene completata in meno di un'ora, l'attività viene completata con lo stato
Successperché è stata completata entro il periodo di tolleranza per la terminazione. Tuttavia, Kubernetes termina il pod e il file scritto viene eliminato immediatamente dal container.Se il DAG supera il limite di spazio di archiviazione del worker e viene eseguito per più di un'ora, il DAG rimane in esecuzione per un'ora e può superare il limite di archiviazione di migliaia di percentuale prima che Kubernetes elimini il pod e Airflow contrassegna l'attività come
Failed.
Diagnosi del DAG non riuscito
Se eseguivi più attività al momento dell'errore, valuta la possibilità di eseguire una sola attività e diagnosticare la pressione delle risorse durante quel periodo per identificare quali attività causano la pressione delle risorse e quali devi aumentare.
Esamina i log delle attività del secondo DAG,
long_running_create_large_txt_file_print_logs:
Nella console Google Cloud, vai alla pagina Ambienti.
Nell'elenco degli ambienti, fai clic sul nome del tuo ambiente. Viene visualizzata la pagina Dettagli ambiente.
Vai alla scheda Log, quindi vai a Tutti i log > Log di Airflow > Worker > Visualizza in Esplora log.
Filtra i log per tipo: mostra solo i messaggi di tipo Errore.
Nei log vedrai messaggi simili ai seguenti:
Container storage usage of worker reached 155.7% of the limit.
This likely means that the total size of local files generated by your DAGs is
close to the storage limit of worker.
You may need to decrease the storage usage or increase the worker storage limit
in your Cloud Composer environment configuration.
o
Pod storage usage of worker reached 140.2% of the limit.
A worker pod was evicted at 2023-12-01T12:30:05Z with message: Pod ephemeral
local storage usage exceeds the total limit of containers 1023Mi.
This eviction likely means that the total size of dags and plugins folders plus
local files generated by your DAGs exceeds the storage limit of worker.
Please decrease the storage usage or increase the worker storage limit in your
Cloud Composer environment configuration.
Questi messaggi indicano che, con l'avanzamento dell'attività, i log di Airflow hanno iniziato a generare errori di stampa quando la dimensione dei file generati dal DAG ha superato il limite di spazio di archiviazione del worker e ha iniziato il periodo di tolleranza per la terminazione. Durante il periodo di tolleranza per la terminazione, il consumo dello spazio di archiviazione non è tornato al limite, il che ha comportato l'eliminazione dei pod al termine del periodo di tolleranza.
Esamina l'integrità dell'ambiente e il monitoraggio del consumo di risorse:
Nella console Google Cloud, vai alla pagina Ambienti.
Nell'elenco degli ambienti, fai clic sul nome del tuo ambiente. Viene visualizzata la pagina Dettagli ambiente.
Vai alla scheda Monitoring e seleziona Panoramica.
Nel riquadro Panoramica dell'ambiente, individua il grafico Environment Health (DAG Monitoring Airflow Monitoring). Contiene un'area rossa che corrisponde all'ora in cui sono stati avviati gli errori di stampa dei log.
Seleziona Worker e trova il grafico Utilizzo totale disco worker. Osserva che la riga Utilizzo di dischi ha un picco e supera la riga Limite di dischi nel momento in cui l'attività era in esecuzione.
Aumenta il limite di spazio di archiviazione del worker
Alloca spazio di archiviazione aggiuntivo worker Airflow affinché il DAG di esempio abbia successo:
Nella console Google Cloud, vai alla pagina Ambienti.
Nell'elenco degli ambienti, fai clic sul nome del tuo ambiente. Viene visualizzata la pagina Dettagli ambiente.
Vai alla scheda Configurazione dell'ambiente.
Trova la configurazione Risorse > Carichi di lavoro e fai clic su Modifica.
Nella sezione Worker, nel campo Archiviazione, specifica il nuovo limite di archiviazione per i worker di Airflow. In questo tutorial, imposta il valore su 2 GB.
Salva le modifiche e attendi qualche minuto per il riavvio dei worker di Airflow.
Testa il DAG con il nuovo limite di spazio di archiviazione
Attiva di nuovo il DAG long_running_create_large_txt_file_print_logs e attendi 1 ora e 15 minuti fino al termine dell'esecuzione.
Nei log di output dell'esecuzione dei DAG, vedrai
Marking task as SUCCESSe lo stato dell'attività indicherà Success, con una durata di 1 ora e 15 minuti, pari al tempo di attesa impostato nel codice DAG.Consulta la sezione Panoramica dell'ambiente nella scheda Monitoraggio e assicurati che non ci siano aree rosse.
Fai clic sulla sezione Worker e trova il grafico Utilizzo totale del disco dei worker. Noterai che la riga Limite di dischi riflette la modifica del limite di spazio di archiviazione, mentre la riga Utilizzo disco rientra nell'intervallo consentito.
Riepilogo
In questo tutorial hai diagnosticato il motivo di un errore dei DAG e hai identificato il tipo di risorsa che causa pressione eseguendo il debug di due DAG di esempio che non funzionano a causa della mancanza di memoria e spazio di archiviazione del worker. Hai quindi eseguito correttamente i DAG dopo aver allocato più memoria e spazio di archiviazione ai worker. Tuttavia, ti consigliamo di ottimizzare i DAG (flussi di lavoro) per ridurre il consumo di risorse dei worker, in quanto non è possibile aumentare le risorse oltre una determinata soglia.
Esegui la pulizia
Per evitare che al tuo account Google Cloud vengano addebitati costi relativi alle risorse utilizzate in questo tutorial, elimina il progetto che contiene le risorse oppure mantieni il progetto ed elimina le singole risorse.
Elimina il progetto
- Nella console Google Cloud, vai alla pagina Gestisci risorse.
- Nell'elenco dei progetti, seleziona il progetto che vuoi eliminare, quindi fai clic su Elimina.
- Nella finestra di dialogo, digita l'ID del progetto e fai clic su Chiudi per eliminare il progetto.
Elimina singole risorse
Se prevedi di esplorare più tutorial e guide rapide, puoi riutilizzare i progetti per evitare di superare i limiti di quota.
Elimina l'ambiente Cloud Composer. Durante questa procedura elimini anche il bucket dell'ambiente.