Cloud Composer 1 | Cloud Composer 2 | Cloud Composer 3
Questo tutorial fornisce i passaggi per eseguire il debug di un DAG Airflow non riuscito in Cloud Composer e diagnosticare i problemi relativi alle risorse dei worker, ad esempio mancanza di memoria o di spazio di archiviazione per i worker, con l'aiuto dei log e dell'ambiente il monitoraggio.
Introduzione
Questo tutorial si concentra sui problemi relativi alle risorse per dimostrare i modi per eseguire il debug di un DAG.
La mancanza di risorse worker allocate causa errori del DAG. Se un'attività Airflow esaurisce la memoria o lo spazio di archiviazione, potresti vedere un'eccezione Airflow, quali:
WARNING airflow.exceptions.AirflowException: Task received SIGTERM signal
INFO - Marking task as FAILED.
o
Task exited with return code Negsignal.SIGKILL
In questi casi, il consiglio generale è di aumentare le risorse dei worker Airflow o di ridurre il numero di attività per worker. Tuttavia, poiché Airflow le eccezioni possono essere generiche, la risorsa specifica che causa il problema.
Questo tutorial spiega come diagnosticare il motivo di un errore del DAG e Identifica il tipo di risorsa che causa problemi eseguendo il debug di due DAG di esempio che non riescono a causa della mancanza di memoria e spazio di archiviazione dei worker.
Obiettivi
Esegui DAG di esempio che non vanno a buon fine per i seguenti motivi:
- Mancanza di memoria del worker
- Spazio di archiviazione insufficiente per i worker
Diagnostica i motivi dell'errore
Aumentare le risorse worker allocate
Testa i DAG con i nuovi limiti di risorse
Costi
Questo tutorial utilizza i seguenti componenti fatturabili di Google Cloud:
Al termine di questo tutorial, puoi evitare la fatturazione continua eliminando il le risorse che hai creato. Per maggiori dettagli, consulta la sezione Pulizia.
Prima di iniziare
Questa sezione descrive le azioni necessarie prima di iniziare il tutorial.
Creare e configurare un progetto
Per questo tutorial, hai bisogno di 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 per un progetto.
Assicurati che l'utente del progetto Google Cloud disponga dei seguenti ruoli per creare le risorse necessarie:
- Amministratore oggetti di ambiente e archiviazione
(
roles/composer.environmentAndStorageObjectAdmin
) - Amministratore Compute (
roles/compute.admin
) - Editor di Monitoring (
roles/monitoring.editor
)
- Amministratore oggetti di ambiente e archiviazione
(
Abilita le API per il tuo progetto
Enable the Cloud Composer API.
Crea l'ambiente Cloud Composer
Crea un ambiente Cloud Composer 2.
Nell'ambito della creazione dell'ambiente,
devi concedere l'estensione agente di servizio API Cloud Composer v2
(roles/composer.ServiceAgentV2Ext
) all'agente di servizio Composer
. Cloud Composer utilizza questo account per eseguire operazioni
nel tuo progetto Google Cloud.
Controllare i limiti delle risorse worker
Controlla i limiti delle risorse dei worker Airflow nel tuo ambiente:
Nella console Google Cloud, vai alla pagina Ambienti.
Nell'elenco degli ambienti, fai clic sul nome del tuo ambiente. Si apre la pagina Dettagli ambiente.
Vai alla scheda Configurazione dell'ambiente.
Vai a Risorse > Configurazione 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 nel prossimo passaggi di questo tutorial.
Esempio: diagnosticare i problemi di esaurimento della memoria
Carica il seguente DAG di esempio 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 accodare la stringa
More
all'elenco. - Stampa la quantità di memoria utilizzata dall'elenco e attende 1 secondo in ogni evoluzione 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 Server web Airflow, segui il link Airflow per il tuo completamente gestito di Google Cloud.
Nell'interfaccia web di Airflow, nella pagina DAG, nella colonna Collegamenti per il DAG, fai clic sul pulsante Attiva DAG.
Fai clic su Attiva.
Nella pagina DAG, fai clic sull'attività attivata ed esamina l'output per assicurarti che il DAG sia stato avviato.
Mentre l'attività è in esecuzione, i log di output stamperanno la dimensione della memoria in GB usato dal DAG.
Dopo diversi minuti, l'attività non riuscirà perché supera il worker di Airflow di 1,875 GB.
Diagnostica il DAG non riuscito
Se al momento dell'errore erano in esecuzione più attività, ti consigliamo di eseguire solo un'attività e di diagnosticare la pressione sulle risorse durante questo periodo per identificare le attività che causano la pressione sulle risorse e le risorse che devi aumentare.
Esamina i log delle attività Airflow
Osserva che l'attività del DAG create_list_with_many_strings
ha un
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 i carichi di lavoro
Rivedi i carichi di lavoro per verificare che il carico dell'attività non causi il nodo in cui il pod viene eseguito per superare il limite di consumo di memoria:
Nella console Google Cloud, vai alla pagina Ambienti.
Nell'elenco degli ambienti, fai clic sul nome dell'ambiente. Viene visualizzata la pagina Dettagli dell'ambiente.
Vai alla scheda Configurazione dell'ambiente.
In Risorse > Cluster GKE > Carichi di lavoro, fai clic su Visualizza carichi di lavoro del cluster.
Controlla se alcuni dei pod del carico di lavoro hanno stati simili ai seguenti:
Error with exit code 137 and 1 more issue. ContainerStatusUnknown with exit code 137 and 1 more issue
Exit code 137
indica che un container o un pod sta tentando di utilizzare più memoria di quella consentita. Il processo viene terminato per evitare l'utilizzo della memoria.
Rivedi il monitoraggio dell'integrità dell'ambiente e del consumo di risorse
Esamina il monitoraggio dell'integrità dell'ambiente e del consumo di risorse:
Nella console Google Cloud, vai alla pagina Ambienti.
Nell'elenco degli ambienti, fai clic sul nome dell'ambiente. Si apre la pagina Dettagli ambiente.
Vai alla scheda Monitoraggio e seleziona Panoramica.
Nel riquadro Panoramica dell'ambiente, individua la Grafico Environment Health (Airflow Monitoring DAG). Contiene un'immagine corrispondente al momento in cui i log hanno iniziato a stampare gli errori.
Seleziona Worker e individua il grafico Utilizzo totale memoria worker. Nota che la riga Utilizzo della memoria presenta un picco al momento dell'esecuzione della task.
Anche se la linea di utilizzo della memoria sul grafico non raggiunge il limite, quando diagnostichi i motivi dell'errore, devi prendere in considerazione solo la memoria allocabile, mentre la linea Limite memoria sul grafico rappresenta la memoria totale disponibile (inclusa la capacità riservata 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 prenota anche un'ulteriore soglia di espulsione: 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 della memoria utilizzata, vedrà che il picco di utilizzo della memoria dell'attività raggiunge la memoria effettiva puoi concludere che l'attività non è riuscita a causa di un worker insufficiente la memoria.
Aumentare il limite di memoria del worker
Alloca memoria aggiuntiva per i worker affinché il DAG di esempio vada a buon fine:
Nella console Google Cloud, vai alla pagina Ambienti.
Nell'elenco degli ambienti, fai clic sul nome dell'ambiente. Viene visualizzata la pagina Dettagli dell'ambiente.
Vai alla scheda Configurazione dell'ambiente.
Trova la configurazione Risorse > Carichi di lavoro e fai clic su Modifica.
Nella sezione Worker, nel campo Memoria, specifica il nuovo limite di memoria per i worker Airflow. In questo tutorial vengono utilizzati 3 GB.
Salva le modifiche e attendi diversi minuti per i worker di Airflow riavvia.
Testa il DAG con il nuovo limite di memoria
Attiva di nuovo il DAG create_list_with_many_strings
e attendi che venga
termina l'esecuzione.
Nei log di output dell'esecuzione del DAG vedrai
Marking task as SUCCESS
e lo stato dell'attività sarà Success.Esamina la sezione Panoramica dell'ambiente nella scheda Monitoring e assicurati che non ci siano aree rosse.
Fai clic sulla sezione Worker e trova Utilizzo totale della memoria dei worker grafico. Vedrai che la riga Limite di memoria riflette la modifica del limite di memoria e la riga Utilizzo della memoria è molto al di sotto del limite di memoria allocato effettivo.
Esempio: diagnostica dei problemi di spazio di archiviazione esaurito
In questo passaggio carichi 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 simula un'operazione di lunga durata.
Le dimensioni del file in entrambi i DAG superano lo spazio di archiviazione worker predefinito di Airflow di 1 GB, ma il secondo DAG ha un'attività di attesa aggiuntiva per estendere per la durata, in modo artificiale.
Analizzerai le differenze nel comportamento di entrambi i DAG nel prossimo passaggi.
Carica un DAG che crea un file di grandi dimensioni
Carica il seguente DAG di esempio 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.txt
da 1,5 GB nello spazio di archiviazione del worker di Airflow. - Stampa la dimensione del file creato utilizzando il modulo
os
Python. - Stampa la durata dell'esecuzione del 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 simulare un DAG di lunga durata e esaminare l'impatto della durata dell'attività sullo stato finale, 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.txt
da 1,5 GB nello spazio di archiviazione del worker di Airflow. - Stampa la dimensione del file creato utilizzando il modulo
os
Python. - Attende 1 ora e 15 minuti per simulare il tempo necessario per le operazioni con il file, ad esempio la lettura.
- Stampa la durata dell'esecuzione del 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)
Attivare DAG di esempio
Attiva il primo DAG, create_large_txt_file_print_logs
:
Nella console Google Cloud, vai alla pagina Ambienti.
Nella colonna Server web Airflow, segui il link Airflow per il tuo completamente gestito di Google Cloud.
Nell'interfaccia web di Airflow, nella pagina DAG, nella colonna Collegamenti per il DAG, fai clic sul pulsante Attiva DAG.
Fai clic su Attiva.
Nella pagina DAG, fai clic sull'attività che hai attivato ed esamina i log di output per assicurarti che il DAG abbia iniziato a funzionare.
Attendi il completamento dell'attività creata con il DAG
create_large_txt_file_print_logs
. L'operazione potrebbe richiedere diversi minuti.Nella pagina DAG, fai clic sull'esecuzione del DAG. Vedrai la tua attività ha uno stato
Success
, anche se il limite di spazio di archiviazione è stato superato.
Esamina i log di Airflow della task:
Nella console Google Cloud, vai alla pagina Ambienti.
Nell'elenco degli ambienti, fai clic sul nome del tuo ambiente. Si apre la pagina Dettagli ambiente.
Vai alla scheda Log, poi a Tutti i log > Log Airflow > Worker > Visualizza in Esplora log.
Filtra i log per tipo: mostra solo i messaggi di errore.
Nei log verranno visualizzati 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 caldo" perché lo spazio di archiviazione utilizzato ha superato il limite ed è stato eliminato in un'ora. Tuttavia, l'esecuzione del DAG non è andata a buon fine perché è stata completata entro il periodo di tolleranza per la chiusura di Kubernetes, che viene 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 Server web Airflow, segui il link Airflow per il tuo ambiente.
Nell'interfaccia web di Airflow, nella pagina DAG, nella colonna Collegamenti per il DAG, fai clic sul pulsante Attiva DAG.
Fai clic su Attiva.
Nella pagina DAG, fai clic sull'attività che hai attivato ed esamina l'output per assicurarti che il DAG sia stato avviato.
Attendi l'esecuzione del DAG
long_running_create_large_txt_file_print_logs
non riesce. L'operazione richiederà circa un'ora.
Esamina i risultati dell'esecuzione del DAG:
Nella pagina DAG, fai clic sul
long_running_create_large_txt_file_print_logs
esecuzione di DAG. Vedrai che l'attività abbia lo statoFailed
e che la durata dell'esecuzione sia stata esattamente 1 ora e 5 minuti, ossia meno del 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.txt
dal container del worker Airflow, il log stampa che il DAG ha avviato in attesa e la durata dell'esecuzione viene stampata nei log delle attività ogni minuto. In questo esempio, il DAG stampa il loglocalfile.txt size:
e le dimensioni del filelocalfile.txt
sarà di 1,5 GB.
Quando il file scritto nel contenitore del worker Airflow supera il limite di spazio di archiviazione, l'esecuzione del DAG dovrebbe non riuscire. Tuttavia, l'attività non termina immediatamente e continua a funzionare fino a quando la sua durata non raggiunge 1 ora e 5 minuti. Questo accade perché Kubernetes non termina immediatamente l'attività e mantiene in esecuzione per consentire un'ora di tempo per il ripristino, noto come "grazia di punto". Quando un nodo esaurisce le risorse, Kubernetes non termina immediatamente il pod per gestire la terminazione in modo corretto, in modo da avere un impatto minimo sull'utente finale.
Il periodo di tolleranza per l'interruzione consente agli utenti di recuperare i file dopo gli errori delle attività, ma può creare confusione durante la diagnostica dei DAG. Quando viene superato il limite di archiviazione del worker Airflow, lo stato dell'attività finale dipende dalla durata dell'esecuzione del 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
Success
perché è stata completata entro il periodo di tolleranza per la terminazione. Tuttavia, Kubernetes termina il pod e il file scritto viene immediatamente eliminato dal container.Se il DAG supera il limite di spazio di archiviazione del worker ed è in esecuzione da più di un'ora, rimane in esecuzione per un'ora e può superare il limite di spazio di archiviazione per migliaia di percento prima che Kubernetes elimini il pod e Airflow contrassegni l'attività come
Failed
.
Diagnostica il DAG non riuscito
Se al momento dell'errore era in esecuzione più attività, valuta la possibilità di eseguire una sola attività e diagnosticare la pressione delle risorse durante quel periodo per identificare quali attività causano pressione sulle risorse e quali devono 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. Si apre la pagina Dettagli ambiente.
Vai alla scheda Log, poi a Tutti i log > Log Airflow > Worker > Visualizza in Esplora log.
Filtra i log per tipo: mostra solo i messaggi di errore.
Nei log verranno visualizzati 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 stampare errori quando le dimensioni dei file generati dal DAG hanno superato il limite di spazio di archiviazione del worker ed è iniziato il periodo di tolleranza per la chiusura. Durante periodo di tolleranza in caso di chiusura, il consumo dello spazio di archiviazione non è tornato al limite che ha portato all'eliminazione dei pod al termine del periodo di tolleranza per la chiusura.
Rivedi il monitoraggio dell'integrità dell'ambiente e del consumo di risorse:
Nella console Google Cloud, vai alla pagina Ambienti.
Nell'elenco degli ambienti, fai clic sul nome dell'ambiente. Si apre la pagina Dettagli ambiente.
Vai alla scheda Monitoraggio e seleziona Panoramica.
Nel riquadro Panoramica dell'ambiente, individua la Grafico Environment Health (Airflow Monitoring DAG). Contiene un'immagine corrispondente al momento in cui i log hanno iniziato a stampare gli errori.
Seleziona Worker e trova il grafico Utilizzo totale del disco dei worker. Nota che la riga Utilizzo disco presenta un picco e supera la riga Limite disco al momento dell'esecuzione dell'attività.
Aumentare il limite di spazio di archiviazione dell'operatore
Alloca spazio di archiviazione aggiuntivo per i worker Airflow affinché il DAG di esempio vada a buon fine:
Nella console Google Cloud, vai alla pagina Ambienti.
Nell'elenco degli ambienti, fai clic sul nome dell'ambiente. Viene visualizzata la pagina Dettagli dell'ambiente.
Vai alla scheda Configurazione dell'ambiente.
Individua la configurazione Risorse > Carichi di lavoro e fai clic su Modifica.
Nella sezione Worker, specifica il nuovo spazio di archiviazione nel campo Storage per i worker di Airflow. In questo tutorial, imposta il valore su 2 GB.
Salva le modifiche e attendi diversi minuti per i worker di Airflow riavvia.
Testa il tuo 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 del DAG, vedrai
Marking task as SUCCESS
e lo stato dell'attività indicherà Successo, con una durata di 1 ora e 15 minuti, pari al tempo di attesa impostato nel codice DAG.Esamina la sezione Panoramica dell'ambiente nella scheda Monitoring e assicurati che non ci siano aree rosse.
Fai clic sulla sezione Worker e trova il grafico Utilizzo totale disco worker. Vedrai che la riga Limite del disco riflette la modifica nella limite di spazio di archiviazione e la riga Utilizzo del disco rientra nell'intervallo consentito.
Riepilogo
In questo tutorial hai diagnosticato il motivo di un errore del DAG e identificato il tipo di risorsa che causa la pressione eseguendo il debug di due DAG di esempio che non vanno a buon fine a causa della mancanza di memoria e spazio di archiviazione dei worker. Dopo aver allocato più memoria e spazio di archiviazione ai tuoi worker, hai eseguito correttamente i DAG. Tuttavia, è consigliabile ottimizzare i DAG (workflow) 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 usati in questo tutorial, elimina il progetto che contiene le risorse o mantenere il progetto ed eliminare le singole risorse.
Elimina il progetto
- In the Google Cloud console, go to the Manage resources page.
- In the project list, select the project that you want to delete, and then click Delete.
- In the dialog, type the project ID, and then click Shut down to delete the project.
Elimina singole risorse
Se prevedi di esplorare più tutorial e guide rapide, puoi riutilizzare i progetti può aiutarti a evitare di superare i limiti di quota.
Elimina l'ambiente Cloud Composer. Inoltre, durante questa procedura viene eliminato il bucket dell'ambiente.