Risoluzione dei problemi dei DAG

Cloud Composer 1 | Cloud Composer 2 | Cloud Composer 3

Questa pagina fornisce i passaggi per la risoluzione dei problemi e le informazioni per un flusso di lavoro comune che le applicazioni presentino problemi di prestazioni.

Molti problemi di esecuzione dei DAG sono causati da prestazioni dell'ambiente non ottimali. Puoi ottimizzare il tuo ambiente Cloud Composer 2 seguendo lo strumento di ottimizzazione le prestazioni e i costi dell'ambiente.

Alcuni problemi di esecuzione dei DAG potrebbero essere causati dallo scheduler di Airflow non funziona correttamente o in modo ottimale. Segui le istruzioni per la risoluzione dei problemi relativi all'agente di pianificazione per risolvere questi problemi.

Risoluzione dei problemi relativi al flusso di lavoro

Per iniziare la risoluzione dei problemi:

  1. Controlla i log di Airflow.

    Puoi aumentare il livello di logging di Airflow eseguendo l'override del parametro seguendo l'opzione di configurazione Airflow.

    Flusso d'aria 2

    Sezione Chiave Valore
    logging logging_level Il valore predefinito è INFO. Imposta su DEBUG per ottenere un livello di dettaglio maggiore nei messaggi di log.

    Flusso d'aria 1

    Sezione Chiave Valore
    core logging_level Il valore predefinito è INFO. Imposta su DEBUG per ottenere un livello di dettaglio maggiore nei messaggi di log.

  2. Controlla la dashboard di monitoraggio.

  3. Esamina Cloud Monitoring.

  4. Nella console Google Cloud, controlla se sono presenti errori nelle pagine dei componenti del tuo ambiente.

  5. Nell'interfaccia web di Airflow, controlla le istanze di attività non riuscite nella Visualizzazione grafico del DAG.

    Sezione Chiave Valore
    webserver dag_orientation LR, TB, RL o BT

Debug degli errori dell'operatore

Per eseguire il debug di un errore di un operatore:

  1. Verifica la presenza di errori specifici delle attività.
  2. Controlla i log di Airflow.
  3. Esamina Cloud Monitoring.
  4. Controlla i log specifici dell'operatore.
  5. Correggi gli errori.
  6. Carica il DAG nella cartella dags/.
  7. Nell'interfaccia web di Airflow, cancellare gli stati passati per il DAG.
  8. Riprendi o esegui il DAG.

Risoluzione dei problemi di esecuzione delle attività

Airflow è un sistema distribuito con molte entità come scheduler, esecutore, worker che comunicano tra loro tramite una coda di attività e Airflow un database e inviare indicatori (come SIGTERM). Il seguente diagramma mostra un panoramica delle interconnessioni tra i componenti Airflow.

Interazione tra i componenti Airflow
Figura 1. Interazione tra i componenti Airflow (fai clic per ingrandire)
di Gemini Advanced.

In un sistema distribuito come Airflow, potrebbe esserci una certa connettività di rete o l'infrastruttura sottostante potrebbe riscontrare problemi intermittenti. Ciò può portare a situazioni in cui le attività possono non riuscire ed essere riprogrammate esecuzione o le attività potrebbero non essere state completate correttamente (ad esempio, o attività bloccate in esecuzione). Airflow ha meccanismi per gestire situazioni di questo tipo e riprende automaticamente il normale funzionamento. Persone che seguo spiegano i problemi comuni che si verificano durante l'esecuzione delle attività da parte di Airflow: Compiti con gli zombi, pillole velenose e segnali SIGTERM.

Risoluzione dei problemi relativi alle attività Zombie

Airflow rileva due tipi di mancata corrispondenza tra un'attività e un processo che la esegue:

  • Le attività zombie sono attività che dovrebbero essere in esecuzione, ma non lo sono. Ciò può accadere se il processo dell'attività è stato interrotto o non risponde, se il worker Airflow non ha segnalato in tempo lo stato dell'attività perché è sovraccaricato o se la VM in cui viene eseguita l'attività è stata arrestata. Airflow le trova periodicamente e le esegue o le ritenta, a seconda delle impostazioni.

    Scoprire le attività zombie

    resource.type="cloud_composer_environment"
resource.labels.environment_name="ENVIRONMENT_NAME"
log_id("airflow-scheduler")
textPayload:"Detected zombie job"

  • Le attività non morte sono attività che non dovrebbero essere in esecuzione. Airflow rileva tali attività periodicamente e le termina.

I motivi e le soluzioni più comuni per le attività Zombie sono elencati di seguito.

Il worker Airflow ha esaurito la memoria

Ogni worker Airflow può eseguire fino a [celery]worker_concurrency istanze di attività contemporaneamente. Se un consumo cumulativo di memoria di queste istanze di attività supera il limite di memoria per un worker Airflow, su di esso verrà eseguito un processo casuale è stato interrotto per liberare risorse.

Scoprire gli eventi di esaurimento memoria dei worker di Airflow

resource.type="k8s_node"
resource.labels.cluster_name="GKE_CLUSTER_NAME"
log_id("events")
jsonPayload.message:"Killed process"
jsonPayload.message:("airflow task" OR "celeryd")

Soluzioni:

Il worker Airflow è stato rimosso

L'espulsione dei pod è una parte normale dell'esecuzione dei carichi di lavoro su Kubernetes. GKE rimuove i pod se hanno esaurito lo spazio di archiviazione o li liberano per i carichi di lavoro con una priorità più elevata.

Scopri le eliminazioni dei worker di Airflow

resource.type="k8s_pod"
resource.labels.cluster_name="GKE_CLUSTER_NAME"
resource.labels.pod_name:"airflow-worker"
log_id("events")
jsonPayload.reason="Evicted"

Soluzioni:

Il worker Airflow è stato terminato

I worker di Airflow potrebbero essere rimossi esternamente. Se le attività in esecuzione non vengono completate durante un periodo di interruzione, verranno interrotte e potrebbero essere rilevate come zombie.

Scopri le terminazioni dei pod dei worker di Airflow

resource.type="k8s_cluster"
resource.labels.cluster_name="GKE_CLUSTER_NAME"
protoPayload.methodName:"pods.delete"
protoPayload.response.metadata.name:"airflow-worker"

Possibili scenari e soluzioni:

  • I worker Airflow vengono riavviati durante le modifiche dell'ambiente, ad esempio upgrade o installazione di pacchetti:

    Scopri le modifiche dell'ambiente Composer

    resource.type="cloud_composer_environment"
resource.labels.environment_name="ENVIRONMENT_NAME"
log_id("cloudaudit.googleapis.com%2Factivity")

    Puoi eseguire queste operazioni quando non sono in esecuzione attività critiche o attivare i tentativi di esecuzione di nuovo delle attività.

  • Vari componenti potrebbero essere temporaneamente non disponibili durante la manutenzione operazioni:

    Scopri le operazioni di manutenzione di GKE

    resource.type="gke_nodepool"
resource.labels.cluster_name="GKE_CLUSTER_NAME"
protoPayload.metadata.operationType="UPGRADE_NODES"

    Puoi specificare periodi di manutenzione per ridurre al minimo si sovrappone all'esecuzione delle attività critiche.

  • Nelle versioni di Cloud Composer 2 precedenti alla 2.4.5, viene eseguita una terminazione di Airflow il worker potrebbe ignorare il segnale SIGTERM e continuare a eseguire attività:

    Scopri il scale down tramite la scalabilità automatica di Composer

    resource.type="cloud_composer_environment"
resource.labels.environment_name="ENVIRONMENT_NAME"
log_id("airflow-worker-set")
textPayload:"Workers deleted"

    Puoi eseguire l'upgrade a una versione successiva di Cloud Composer in cui è stato risolto.

Il worker di Airflow era sottoposto a un carico elevato

La quantità di risorse di CPU e memoria disponibili per un worker Airflow è limitata dalla configurazione dell'ambiente. Se un utilizzo si avvicina ai limiti, causerebbe una contesa delle risorse e ritardi non necessari durante l'esecuzione della task. In situazioni estreme, quando le risorse scarseggiano per periodi di tempo più lunghi, potrebbero verificarsi attività zombie.

Soluzioni:

Il database Airflow era sotto un carico elevato

Un database viene utilizzato da vari componenti di Airflow per comunicare tra loro e, in particolare, per memorizzare gli heartbeat delle istanze di attività. La scarsità di risorse nel database comporterà tempi di query più lunghi e potrebbe influire sull'esecuzione di un'attività.

Soluzioni:

Il database Airflow era temporaneamente non disponibile

Un worker Airflow potrebbe impiegare del tempo per rilevare e gestire in modo corretto gli errori intermittenti, ad esempio i problemi di connettività temporanei. Potrebbe superare la soglia di rilevamento degli zombie predefinita.

Scoprire i timeout heartbeat di Airflow

resource.type="cloud_composer_environment"
resource.labels.environment_name="ENVIRONMENT_NAME"
log_id("airflow-worker")
textPayload:"Heartbeat time limit exceeded"

Soluzioni:

  • Aumenta il timeout per le attività zombie e sostituisci il valore dell'opzione di configurazione di Airflow [scheduler]scheduler_zombie_task_threshold:

    Sezione Chiave Valore Note
    scheduler scheduler_zombie_task_threshold Nuovo timeout (in secondi) Il valore predefinito il valore è 300

Risoluzione dei problemi relativi alla pillola avvelenata

La pillola velenosa è un meccanismo utilizzato da Airflow per arrestare le attività Airflow.

Airflow utilizza la pillola avvelenata in queste situazioni:

  • Quando un programmatore termina un'attività che non è stata completata in tempo.
  • Quando un'attività scade o viene eseguita per troppo tempo.

Quando Airflow utilizza la pillola avvelenata, puoi vedere le seguenti voci di log nei log di un worker Airflow che ha eseguito l'attività:

  INFO - Subtask ... WARNING - State of this instance has been externally set
  to success. Taking the poison pill.
  INFO - Subtask ... INFO - Sending Signals.SIGTERM to GPID <X>
  INFO - Subtask ... ERROR - Received SIGTERM. Terminating subprocesses.

Possibili soluzioni:

  • Verifica che nel codice dell'attività non siano presenti errori che potrebbero causare un'esecuzione troppo lunga.
  • (Cloud Composer 2) Aumentare la CPU e la memoria di Airflow in modo che le attività vengano eseguite più velocemente.

  • Aumenta il valore dell'opzione [celery_broker_transport_options]visibility-timeout Configurazione Airflow.

    Di conseguenza, lo scheduler attende più tempo per il completamento di un'attività, prima di considerare l'attività come un compito zombie. Questa opzione è particolarmente utile per le attività che richiedono molto tempo e durano molte ore. Se il valore è troppo basso (ad esempio 3 ore), lo scheduler considera "bloccate" le attività in esecuzione per 5 o 6 ore (attività zombie).

  • Aumenta il valore del flusso d'aria [core]killed_task_cleanup_time di configurazione.

    Un valore più lungo offre più tempo ai worker Airflow per completare le attività in modo corretto. Se il valore è troppo basso, le attività Airflow potrebbero essere interrotte all'improvviso, senza abbastanza tempo per finire il lavoro con grazia.

Risoluzione dei problemi relativi agli indicatori SIGTERM

Gli indicatori SIGTERM vengono utilizzati da Linux, Kubernetes, Airflow Scheduler e Celery per terminare i processi responsabili dell'esecuzione di worker o attività Airflow.

Esistono diversi motivi per cui gli indicatori SIGTERM vengono inviati in un ambiente:

  • Un'attività è diventata un'attività Zombie e deve essere interrotta.

  • Il programmatore ha rilevato un duplicato di un'attività e invia all'attività i segnali Poison Pill e SIGTERM per interromperla.

  • In Scalabilità automatica orizzontale dei pod, GKE Il piano di controllo invia indicatori SIGTERM per rimuovere i pod che non sono più necessaria.

  • Lo scheduler può inviare segnali SIGTERM al processo DagFileProcessorManager. Questi indicatori SIGTERM vengono utilizzati dal programmatore per gestire il ciclo di vita del processo DagFileProcessorManager e possono essere ignorati in tutta sicurezza.

    Esempio:

    Launched DagFileProcessorManager with pid: 353002
Sending Signals.SIGTERM to group 353002. PIDs of all processes in the group: []
Sending the signal Signals.SIGTERM to group 353002
Sending the signal Signals.SIGTERM to process 353002 as process group is missing.

  • Condizione di gara tra il callback heartbeat e i callback di uscita in local_task_job, che monitora l'esecuzione dell'attività. Se il battito cardiaco rileva che un'attività è stata contrassegnata come completata, non può distinguere se l'attività stessa è riuscita o che Airflow è stato detto di prendere in considerazione l'attività riuscito. In ogni caso, un esecutore dell'attività verrà terminato senza attendere per uscire.

    Questi segnali SIGTERM possono essere ignorati in tutta sicurezza. L'attività è già in e l'esecuzione del DAG nel suo insieme non sarà interessati.

    La voce di log Received SIGTERM. è l'unica differenza tra l'uscita normale e l'interruzione dell'attività nello stato di esito positivo.

    Condizione di gara tra i callback heartbeat ed exit
    Figura 2. Condizione di gara tra i callback heartbeat ed exit (fai clic per ingrandire)

  • Un componente Airflow utilizza più risorse (CPU, memoria) di quanto consentito dal nodo del cluster.

  • Il servizio GKE esegue operazioni di manutenzione e invia segnali SIGTERM ai pod in esecuzione su un nodo di cui sta per essere eseguito l'upgrade. Quando un'istanza di attività viene terminata con SIGTERM, puoi visualizzare il seguente log voci nei log di un worker Airflow che ha eseguito l'attività:

{local_task_job.py:211} WARNING - State of this instance has been externally
set to queued. Terminating instance. {taskinstance.py:1411} ERROR - Received
SIGTERM. Terminating subprocesses. {taskinstance.py:1703} ERROR - Task failed
with exception

Possibili soluzioni:

Questo problema si verifica quando una VM che esegue l'attività esaurisce la memoria. Questo non è alle configurazioni Airflow ma alla quantità di memoria disponibile VM.

L'aumento della memoria dipende dalla versione di Cloud Composer che utilizzi. Ad esempio:

  • In Cloud Composer 2, puoi assegnare più risorse di CPU e memoria ad Airflow worker.

  • Nel caso di Cloud Composer 1, puoi ricreare l'ambiente utilizzando una e un tipo di macchina con più prestazioni.

  • In entrambe le versioni di Cloud Composer, puoi abbassare il valore dell'opzione di configurazione di Airflow [celery]worker_concurrency concurrency. Questa opzione determina quante attività vengono eseguite contemporaneamente da un determinato worker Airflow.

Per saperne di più sull'ottimizzazione del tuo ambiente Cloud Composer 2, vedi Ottimizza le prestazioni e i costi dell'ambiente

Query di Cloud Logging per scoprire i motivi dei riavvii o delle eliminazioni dei pod

Gli ambienti di Cloud Composer utilizzano i cluster GKE come infrastruttura di calcolo livello di sicurezza. In questa sezione potrai trovare query utili che ti aiuteranno a trova motivi per i riavvii o le eliminazioni del worker o dello scheduler Airflow.

Le query presentate di seguito potrebbero essere ottimizzate nel seguente modo:

  • puoi specificare la sequenza temporale che ti interessa in Cloud Logging; ad esempio le ultime 6 ore, i 3 giorni o puoi definire un intervallo di tempo personalizzato

  • devi specificare Cloud Composer CLUSTER_NAME

  • puoi anche limitare la ricerca a un pod specifico aggiungendo POD_NAME

Scopri i container riavviati

    resource.type="k8s_node"
    log_id("kubelet")
    jsonPayload.MESSAGE:"will be restarted"
    resource.labels.cluster_name="CLUSTER_NAME"

Query alternativa per limitare i risultati a un pod specifico:

    resource.type="k8s_node"
    log_id("kubelet")
    jsonPayload.MESSAGE:"will be restarted"
    resource.labels.cluster_name="CLUSTER_NAME"
    "POD_NAME"

Scoprire l'arresto dei container a seguito di un evento di esaurimento della memoria

    resource.type="k8s_node"
    log_id("events")
    (jsonPayload.reason:("OOMKilling" OR "SystemOOM")
      OR jsonPayload.message:("OOM encountered" OR "out of memory"))
    severity=WARNING
    resource.labels.cluster_name="CLUSTER_NAME"

Query alternativa per limitare i risultati a un pod specifico:

    resource.type="k8s_node"
    log_id("events")
    (jsonPayload.reason:("OOMKilling" OR "SystemOOM")
      OR jsonPayload.message:("OOM encountered" OR "out of memory"))
    severity=WARNING
    resource.labels.cluster_name="CLUSTER_NAME"
    "POD_NAME"

Scoprire i container che hanno interrotto l'esecuzione

    resource.type="k8s_node"
    log_id("kubelet")
    jsonPayload.MESSAGE:"ContainerDied"
    severity=DEFAULT
    resource.labels.cluster_name="CLUSTER_NAME"

Query alternativa per limitare i risultati a un pod specifico:

    resource.type="k8s_node"
    log_id("kubelet")
    jsonPayload.MESSAGE:"ContainerDied"
    severity=DEFAULT
    resource.labels.cluster_name="CLUSTER_NAME"
    "POD_NAME"

Impatto delle operazioni di aggiornamento o upgrade sulle esecuzioni delle attività Airflow

Le operazioni di aggiornamento o upgrade interrompono le attività Airflow attualmente in esecuzione, a meno che un'attività non venga eseguita in modalità posticipabile.

Ti consigliamo di eseguire queste operazioni quando prevedi un impatto minimo sulle esecuzioni delle attività di Airflow e di configurare meccanismi di ripetizione appropriati nelle attività e nei DAG.

Risoluzione dei problemi relativi alle attività KubernetesExecutor

CeleryKubernetesExecutor è un tipo di esecutore in Cloud Composer 3 che possono usare CeleryExecutor e KubernetesExecutor contemporaneamente nel tempo.

Per ulteriori informazioni sulla risoluzione dei problemi relativi alle attività eseguite con KubernetesExecutor, consulta la pagina Utilizzare CeleryKubernetesExecutor.

Problemi comuni

Le sezioni seguenti descrivono i sintomi e le potenziali correzioni di alcuni problemi comuni relativi ai DAG.

L'attività Airflow è stata interrotta da Negsignal.SIGKILL

A volte la tua attività potrebbe utilizzare più memoria di quella allocata al worker Airflow. In questo caso, potrebbe essere interrotta da Negsignal.SIGKILL. Il sistema invia questo segnale per evitare un ulteriore consumo di memoria che potrebbe influire sull'esecuzione di altre attività Airflow. Nel log del worker di Airflow potresti vedere la seguente voce di log:

{local_task_job.py:102} INFO - Task exited with return code Negsignal.SIGKILL

Negsignal.SIGKILL potrebbe essere visualizzato anche come codice -9.

Possibili soluzioni:

  • Riduci il worker_concurrency dei worker Airflow.

  • Nel caso di Cloud Composer 2, aumenta la memoria dei worker Airflow.

  • Nel caso di Cloud Composer 1, esegui l'upgrade a un tipo di macchina più grande utilizzato nel cluster Cloud Composer.

  • Ottimizza le tue attività per utilizzare meno memoria.

  • Gestisci le attività ad alta intensità di risorse in Cloud Composer utilizzando KubernetesPodOperator o GKEStartPodOperator per l'isolamento delle attività e l'allocazione personalizzata delle risorse.

L'attività non riesce senza l'emissione di log a causa di errori di analisi dei DAG

A volte potrebbero verificarsi lievi errori DAG che portano a una situazione in cui uno scheduler Airflow e un processore DAG sono in grado di pianificare l'esecuzione delle attività e analizzare un file DAG (rispettivamente), ma il worker Airflow non riesce a eseguire le attività da un DAG di questo tipo, in quanto ci sono errori di programmazione nel file DAG Python. Ciò potrebbe portare a una situazione in cui un'attività Airflow è contrassegnata come Failed e non è presente alcun log della relativa esecuzione.

Soluzioni:

  • Verifica nei log dei worker di Airflow che non siano presenti errori generati da Worker Airflow correlato a errori di analisi dei DAG o dei DAG mancanti.

  • Aumenta i parametri relativi all'analisi dei DAG:

  • Consulta anche Ispezione dei log del processore DAG.

L'attività non riesce senza emettere log a causa della pressione delle risorse

Sintomo: durante l'esecuzione di un'attività, il processo secondario del worker Airflow è responsabile per l'esecuzione delle attività Airflow viene interrotta bruscamente. L'errore visibile nel log del worker di Airflow potrebbe essere simile a quello seguente:

...
File "/opt/python3.8/lib/python3.8/site-packages/celery/app/trace.py", line 412, in trace_task    R = retval = fun(*args, **kwargs)  File "/opt/python3.8/lib/python3.8/site-packages/celery/app/trace.py", line 704, in __protected_call__    return self.run(*args, **kwargs)  File "/opt/python3.8/lib/python3.8/site-packages/airflow/executors/celery_executor.py", line 88, in execute_command    _execute_in_fork(command_to_exec)  File "/opt/python3.8/lib/python3.8/site-packages/airflow/executors/celery_executor.py", line 99, in _execute_in_fork
raise AirflowException('Celery command failed on host: ' + get_hostname())airflow.exceptions.AirflowException: Celery command failed on host: airflow-worker-9qg9x
...

Soluzione:

L'attività non riesce senza emettere log a causa dell'eliminazione del pod

I pod di Google Kubernetes Engine sono soggetti al ciclo di vita dei pod Kubernetes e all'espulsione dei pod. I picchi di attività e la pianificazione in co-scheduling dei worker sono le due cause più comuni dell'espulsione dei pod in Cloud Composer.

L'eliminazione dei pod può verificarsi quando un determinato pod fa un uso eccessivo delle risorse di un nodo, rispetto alle aspettative configurate per il consumo di risorse per il nodo. Per Ad esempio, l'eliminazione potrebbe avvenire quando in un pod vengono eseguite diverse attività che usano molta memoria, e il relativo carico combinato fa sì che il nodo in cui viene eseguito questo pod superi limite per il consumo di memoria.

Se un pod di worker di Airflow viene espulso, tutte le istanze di attività in esecuzione su quel pod vengono interrotte e successivamente contrassegnate come non riuscite da Airflow.

I log vengono inseriti nel buffer. Se un pod di lavoro viene espulso prima dello svuotamento del buffer, i log non vengono emessi. L'errore di attività senza log indica che i worker di Airflow vengono riavviati a causa di un esaurimento della memoria (OOM). Alcuni log potrebbero essere presenti in Cloud Logging anche se i log di Airflow non sono stati emessi.

Per visualizzare i log:

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

    Vai ad Ambienti

  2. Nell'elenco degli ambienti, fai clic sul nome dell'ambiente. Viene visualizzata la pagina Dettagli dell'ambiente.

  3. Vai alla scheda Log.

  4. Visualizza i log dei singoli worker in Tutti i log -> Log Airflow -> Worker -> (singolo worker).

L'esecuzione del DAG è limitata dalla memoria. Ogni esecuzione di attività inizia con due procedimenti Airflow: esecuzione e monitoraggio delle attività. Ogni nodo può gestire fino a 6 attività contemporaneamente (circa 12 processi caricati con i moduli Airflow). Può essere utilizzata più memoria, a seconda della natura del DAG.

Sintomo:

  1. Nella console Google Cloud, vai alla pagina Carichi di lavoro.

    Vai a Carichi di lavoro

  2. Se sono presenti airflow-worker pod che mostrano Evicted, fai clic su ogni pod rimosso pod e cerca il token The node was low on resource: memory nella parte superiore della finestra.

Correzione:

  • In Cloud Composer 1, crea un nuovo ambiente Cloud Composer con un tipo di macchina più grande rispetto al tipo di macchina corrente.
  • In Cloud Composer 2, aumenta i limiti di memoria per i worker di Airflow.
  • Controlla i log dei pod airflow-worker per individuare possibili cause dell'eliminazione. Per ulteriori informazioni sul recupero dei log dai singoli pod, consulta Risolvere i problemi relativi ai carichi di lavoro di cui è stato eseguito il deployment.
  • Assicurati che le attività nel DAG siano idempotenti e che sia possibile recuperare.

  • Evita di scaricare file non necessari nel file system locale dei worker di Airflow.

    I worker di Airflow hanno una capacità limitata del file system locale. Ad esempio, in Cloud Composer 2 un worker può avere da 1 GB a 10 GB di spazio di archiviazione. Quando lo spazio di archiviazione finisce, il pod worker di Airflow viene espulso dal piano di controllo GKE. Questa operazione non riesce a tutte le attività rimosse worker era in esecuzione.

    Esempi di operazioni problematiche:

    • Download di file o oggetti e archiviazione locale in un flusso Airflow worker. Archivia invece questi oggetti direttamente in un servizio adatto, ad esempio un bucket Cloud Storage.
    • Accesso a oggetti di grandi dimensioni nella cartella /data da un worker Airflow. Il worker di Airflow scarica l'oggetto nel proprio file system locale. Implementa invece i DAG in modo che i file di grandi dimensioni vengano elaborati all'esterno del pod di lavoro Airflow.

Timeout importazione caricamento DAG

Sintomo:

  • Nell'interfaccia web di Airflow, nella parte superiore della pagina dell'elenco dei DAG, viene visualizzato un allarme rosso mostra Broken DAG: [/path/to/dagfile] Timeout.

  • In Cloud Monitoring: i log airflow-scheduler contengono voci simili a:

    • ERROR - Process timed out
    • ERROR - Failed to import: /path/to/dagfile
    • AirflowTaskTimeout: Timeout

Correzione:

Esegui l'override dell'opzione di configurazione di Airflow dag_file_processor_timeout e concedi più tempo per l'analisi del DAG:

Sezione Chiave Valore
core dag_file_processor_timeout Nuovo valore di timeout

L'esecuzione del DAG non termina entro il tempo previsto

Sintomo:

A volte un'esecuzione del DAG non termina perché le attività di Airflow si bloccano e l'esecuzione del DAG dura più a lungo del previsto. In condizioni normali, le attività Airflow non rimangono indefinitamente nello stato in coda o in esecuzione, perché Airflow dispone di procedure di timeout e pulizia che aiutano a evitare questa situazione.

Correzione:

  • Utilizza la dagrun_timeout per i DAG. Ad esempio: dagrun_timeout=timedelta(minutes=120). Di conseguenza, ogni esecuzione del DAG deve essere completata entro il timeout dell'esecuzione del DAG e le attività non completate devono essere contrassegnate come Failed o Upstream Failed. Per ulteriori informazioni sugli stati delle attività di Airflow, consulta la documentazione di Apache Airflow.

  • Utilizza la timeout dell'esecuzione di attività per definire un timeout predefinito per le attività eseguite in base a Apache Operatori Airflow.

Esecuzioni di DAG non eseguite

Sintomo:

Quando una data di pianificazione per un DAG è impostata in modo dinamico, questo può portare a vari e inaspettati. Ad esempio:

  • L'esecuzione di un DAG è sempre nel futuro e il DAG non viene mai eseguito.

  • Le esecuzioni DAG passate sono contrassegnate come eseguite e riuscite nonostante non lo siano.

Ulteriori informazioni sono disponibili nella documentazione di Apache Airflow.

Correzione:

  • Segui i consigli riportati nella documentazione di Apache Airflow.

  • Imposta start_date statico per i DAG. In alternativa, puoi utilizzare catchup=False per disabilitare l'esecuzione del DAG per date passate.

  • Evita di utilizzare datetime.now() o days_ago(<number of days>), a meno che non sappia gli effetti collaterali di questo approccio.

Aumento del traffico di rete da e verso il database Airflow

La quantità di traffico della rete tra i cluster GKE del tuo ambiente cluster e il database Airflow dipende dal numero di DAG, dal numero le attività nei DAG e il modo in cui i DAG accedono ai dati nel database Airflow. I seguenti fattori possono influire sull'utilizzo della rete:

  • Query al database Airflow. Se i tuoi DAG eseguono molte query, generano grandi quantità di traffico. Esempi: controllo dello stato delle attività prima di procedere con altre attività, query sulla tabella XCom, dumping dei contenuti del database Airflow.

  • Numero elevato di attività. Più attività sono da pianificare, più viene generato traffico di rete. Questo vale sia per il numero totale di attività nei DAG sia per la frequenza di pianificazione. Quando lo scheduler Airflow pianifica le esecuzioni dei DAG, esegue query sul database Airflow e genera traffico.

  • L'interfaccia web di Airflow genera traffico di rete perché esegue query il database Airflow. Usare in modo intensivo pagine con grafici, attività e possono generare grandi volumi di traffico di rete.

Il DAG fa arrestare in modo anomalo il server web Airflow o ne causa il ritorno di un errore 502 gateway timeout

Gli errori del server web possono verificarsi per diversi motivi. Controlla i log di airflow-webserver in Cloud Logging per determinare la causa dell'errore 502 gateway timeout.

Calcolo pesante

Questa sezione riguarda solo Cloud Composer 1.

Evita di eseguire calcoli pesanti al momento dell'analisi del DAG.

A differenza dei nodi worker e di pianificazione, i cui tipi di macchine possono essere personalizzati per avere una maggiore capacità di CPU e memoria, il server web utilizza un tipo di macchina fisso, che può causare errori di analisi del DAG se il calcolo in fase di analisi è troppo pesante.

Tieni presente che il server web ha 2 vCPU e 2 GB di memoria. Il valore predefinito per core-dagbag_import_timeout è 30 secondi. Questo valore di timeout definisce il limite massimo di tempo impiegato da Airflow per caricare un modulo Python nella cartella dags/.

Autorizzazioni errate

Questa sezione si applica solo a Cloud Composer 1.

Il server web non viene eseguito con lo stesso account di servizio dei worker e dello schedulatore. Di conseguenza, i worker e lo scheduler potrebbero essere in grado di accedere e le risorse gestite dall'utente a cui il server web non può accedere.

Ti consigliamo di evitare di accedere a risorse non pubbliche durante Analisi dei DAG. A volte, questo è inevitabile e dovrai concedere le autorizzazioni all'account di servizio del server web. Il nome dell'account di servizio viene ricavato dal dominio del server web. Ad esempio, se il dominio è example-tp.appspot.com, l'account di servizio è example-tp@appspot.gserviceaccount.com.

Errori DAG

Questa sezione si applica solo a Cloud Composer 1.

Il server web viene eseguito su App Engine ed è separato da al cluster GKE dell'ambiente. Il server web analizza il DAG file di definizione e un 502 gateway timeout può verificarsi in caso di errori nel DAG. Airflow funziona normalmente senza un server web funzionante se il DAG problematico non interrompe i processi in esecuzione in GKE. In questo caso, puoi utilizzare gcloud composer environments run per recuperare dettagli dal tuo ambiente e come soluzione alternativa se il server web non disponibile.

In altri casi, puoi eseguire l'analisi dei DAG in GKE e cercare DAG che generano eccezioni irreversibili di Python o timeout (il valore predefinito è 30 secondi). Per risolvere il problema, connettiti a una shell remota in un contenitore di worker Airflow e controlla la presenza di errori di sintassi. Per ulteriori informazioni, consulta Test dei DAG.

Gestione di un numero elevato di DAG e plug-in nelle cartelle DAG e plugin

I contenuti di /dags e /plugins cartelle sono sincronizzati da del bucket del tuo ambiente ai file system locali dei worker di Airflow e scheduler.

Maggiore è il numero di dati archiviati in queste cartelle, maggiore sarà il tempo necessario per eseguire la sincronizzazione dei dati. Per risolvere queste situazioni:

  • Limita il numero di file nelle cartelle /dags e /plugins. Memorizza solo il numero minimo di file richiesti.

  • Se possibile, aumenta lo spazio su disco disponibile per gli scheduler e i worker di Airflow.

  • Se possibile, aumenta la CPU e la memoria degli scheduler e dei worker di Airflow, affinché l'operazione di sincronizzazione venga eseguita più velocemente.

  • Nel caso di un numero molto elevato di DAG, dividi i DAG in batch, comprimi in archivi zip e implementali nella cartella /dags. Questo approccio velocizza la procedura di sincronizzazione dei DAG. Componenti Airflow decomprimi gli archivi ZIP prima di elaborare i DAG.

  • La generazione di DAG in modo programmatico potrebbe anche essere un metodo per limitare il numero di file DAG memorizzati nella cartella /dags. Consulta la sezione sui DAG di pubblicità programmatica da evitare problemi con la pianificazione e l'esecuzione dei DAG generati in modo programmatico.

Non pianificare contemporaneamente DAG generati tramite programmazione

La generazione di oggetti DAG tramite programmazione da un file DAG è un metodo efficiente per creare molti DAG simili con piccole differenze.

È importante non pianificare l'esecuzione di tutti questi DAG contemporaneamente. Là è un'alta probabilità che i worker Airflow non abbiano CPU e memoria sufficienti per eseguire tutte le attività pianificate contemporaneamente.

Per evitare problemi con la pianificazione dei DAG programmatici:

  • Aumenta la contemporaneità dei worker e fai lo scale up del tuo ambiente, in modo che possa di eseguire più attività contemporaneamente.
  • Genera i DAG in modo da distribuire le pianificazioni in modo uniforme nel tempo, in modo da evitare di pianificare centinaia di attività contemporaneamente, in modo che i worker di Airflow il tempo necessario per eseguire tutte le attività pianificate.

Errore 504 durante l'accesso al server web Airflow

Consulta la sezione Errore 504 durante l'accesso all'interfaccia utente di Airflow.

Viene lanciata un'eccezione Lost connection to Postgres server during query durante l'esecuzione della task o subito dopo

Le eccezioni Lost connection to Postgres server during query si verificano spesso quando si verificano le seguenti condizioni:

  • Il DAG utilizza PythonOperator o un operatore personalizzato.
  • Il DAG esegue query sul database Airflow.

Se da una funzione richiamabile vengono eseguite più query, le tracce potrebbero puntare erroneamente alla linea self.refresh_from_db(lock_for_update=True) nella codice Airflow; si tratta della prima query del database dopo l'esecuzione dell'attività. La la causa effettiva dell'eccezione si verifica prima di questo, quando una sessione SQLAlchemy non è chiuso correttamente.

Le sessioni SQLAlchemy hanno come ambito un thread e create in una funzione richiamabile una sessione può essere continuata in un secondo momento all'interno del codice Airflow. Se si verificano ritardi significativi tra le query all'interno di una sessione, la connessione potrebbe essere già stata chiusa dal server Postgres. Il timeout della connessione in Gli ambienti Cloud Composer sono impostati su circa 10 minuti.

Risoluzione:

  • Utilizza il decoratore airflow.utils.db.provide_session. Questo decorator fornisce una sessione valida per il database Airflow in session e chiude correttamente la sessione al termine della funzione.
  • Non utilizzare una singola funzione a lunga esecuzione. Sposta invece tutto il database per separare funzioni, in modo che ci siano più funzioni con il decorator di airflow.utils.db.provide_session. In questo caso, le sessioni vengono chiusi automaticamente dopo aver recuperato i risultati della query.

Controllo del tempo di esecuzione di DAG, attività ed esecuzioni parallele dello stesso DAG

Se vuoi controllare la durata di un'esecuzione singola di un determinato DAG, puoi utilizzare il parametro DAG dagrun_timeout. Ad esempio, se prevedi che un'esecuzione singola di un DAG (indipendentemente dal fatto che l'esecuzione termini con esito positivo o negativo) non debba durare più di un'ora, imposta questo parametro su 3600 secondi.

Puoi anche controllare la durata di una singola attività Airflow. Per farlo, puoi utilizzare execution_timeout.

Se vuoi controllare il numero di esecuzioni DAG attive per un determinato DAG, puoi utilizzare l'[core]max-active-runs-per-dag opzione di configurazione di Airflow.

Se vuoi che venga eseguita una sola istanza di un DAG in un determinato momento, imposta il parametro max-active-runs-per-dag su 1.

Problemi che interessano la sincronizzazione di DAG e plug-in con scheduler, worker e server web

Cloud Composer sincronizza i contenuti di /dags e /plugins cartelle a scheduler e worker. Alcuni oggetti nelle cartelle /dags e /plugins la sincronizzazione potrebbe non funzionare correttamente o almeno rallentarla.

  • La cartella /dags viene sincronizzata con gli scheduler e i worker. Questa cartella non viene sincronizzata con i server web in Cloud Composer 2 o se attivi DAG Serialization in Cloud Composer 1.

  • La cartella /plugins viene sincronizzata con gli scheduler, i worker e i server web.

Potresti riscontrare i seguenti problemi:

  • Hai caricato file compressi con gzip che utilizzano la transcodifica di compressione nelle cartelle /dags e /plugins. Di solito accade se utilizzi il flag --gzip-local-all in una gcloud storage cp per caricare i dati nel bucket.

    Soluzione: elimina l'oggetto che ha utilizzato la transcodifica con compressione e poi caricalo di nuovo. al bucket.

  • Uno degli oggetti è denominato ".": questo tipo di oggetto non viene sincronizzato con gli pianificatori e i worker e potrebbe interrompere completamente la sincronizzazione.

    Soluzione: rinomina l'oggetto problematico.

  • Una cartella e un file Python DAG hanno gli stessi nomi, ad esempio a.py. In questo caso, il file DAG non è sincronizzato correttamente con i componenti di Airflow.

    Soluzione: rimuovi la cartella che ha lo stesso nome di un file Python DAG.

  • Uno degli oggetti nelle cartelle /dags o /plugins contiene un simbolo / alla fine del nome dell'oggetto. Questi oggetti possono ingannare il processo di sincronizzazione perché il simbolo / indica che un oggetto è una cartella, non un file.

    Soluzione: rimuovi il simbolo / dal nome dell'oggetto problematico.

  • Non archiviare i file non necessari nelle cartelle /dags e /plugins.

    A volte i DAG e i plug-in che implementi sono accompagnati da file aggiuntivi, ad esempio file che memorizzano i test per questi componenti. Questi file vengono sincronizzati con i worker e gli scheduler e influiscono sul tempo necessario per copiarli in questi elementi.

    Soluzione: non archiviare file aggiuntivi e non necessari in /dags e /plugins cartelle.

Done [Errno 21] Is a directory: '/home/airflow/gcs/dags/...' errore generato da scheduler e worker

Questo problema si verifica perché gli oggetti possono avere uno spazio dei nomi sovrapposto in Cloud Storage, mentre allo stesso tempo gli schedulatori e i worker utilizzano file system tradizionali. Ad esempio, è possibile per aggiungere una cartella e un oggetto con lo stesso nome al di sincronizzare la directory di una VM con un bucket. Quando il bucket viene sincronizzato con gli scheduler e i worker dell'ambiente, generato questo errore, il che può portare a errori nelle attività.

Per risolvere il problema, assicurati che non ci siano spazi dei nomi in sovrapposizione nel bucket dell'ambiente. Ad esempio, se sia /dags/misc (un file) sia /dags/misc/example_file.txt (un altro file) si trovano in un bucket, viene generato un errore dal programmatore.

Interruzioni temporanee durante la connessione a Airflow Metadata DB

Cloud Composer viene eseguito su un'infrastruttura cloud distribuita. Significa che di tanto in tanto potrebbero comparire alcuni problemi temporanei interrompere l'esecuzione delle tue attività Airflow.

In questi casi, nei log dei worker di Airflow potresti visualizzare i seguenti messaggi di errore:

"Can't connect to Postgres server on 'airflow-sqlproxy-service.default.svc.cluster.local' (111)"

o

"Can't connect to Postgres server on 'airflow-sqlproxy-service.default.svc.cluster.local' (104)"

Questi problemi intermittenti possono essere causati anche da operazioni di manutenzione eseguite per gli ambienti Cloud Composer.

Di solito questi errori sono intermittenti e se le tue attività Airflow sono idempotenti e ci sono nuovi tentativi configurati, non dovresti essere immune. Puoi anche definire periodi di manutenzione.

Un altro motivo di questi errori potrebbe essere la mancanza di risorse nel cluster del tuo ambiente. In questi casi, puoi eseguire lo scaling o ottimizzare il tuo ambiente come descritto nelle istruzioni per lo scaling degli ambienti o l'ottimizzazione dell'ambiente.

Un'esecuzione di DAG è contrassegnata come riuscita, ma non contiene attività eseguite

Se un'esecuzione di DAG execution_date è precedente a start_date del DAG, potresti visualizzare esecuzioni di DAG che non hanno esecuzioni di attività, ma sono comunque contrassegnate come riuscite.

Un&#39;esecuzione di DAG riuscita senza attività eseguite
Figura 3. Un'esecuzione di DAG riuscita senza attività eseguite (fai clic per ingrandire)
di Gemini Advanced.

Causa

Questa situazione potrebbe verificarsi in uno dei seguenti casi:

  • La mancata corrispondenza è causata dalla differenza di fuso orario tra execution_date e start_date del DAG. Ad esempio, può accadere quando utilizzando pendulum.parse(...) per impostare start_date.

  • Il start_date del DAG è impostato su un valore dinamico, ad esempio airflow.utils.dates.days_ago(1)

Soluzione:

  • Assicurati che execution_date e start_date utilizzino lo stesso fuso orario.

  • Specifica un start_date statico e combinalo con catchup=False per evitare di eseguire DAG con date di inizio passate.

Un DAG non è visibile nella UI di Airflow o DAG e lo scheduler non lo pianifica

Il processore DAG analizza ogni DAG prima che possa essere programmato dallo scheduler e prima che un DAG diventi visibile in la UI di Airflow o la UI DAG.

Le seguenti opzioni di configurazione di Airflow definiscono i timeout per l'analisi dei DAG:

Se un DAG non è visibile nell'interfaccia utente di Airflow o nell'interfaccia utente del DAG:

  • Controlla i log del processore DAG per verificare se è in grado di elaborare correttamente il DAG. In caso di problemi, potresti vedere le seguenti voci di log nei log del processore DAG o dello scheduler:
[2020-12-03 03:06:45,672] {dag_processing.py:1334} ERROR - Processor for
/usr/local/airflow/dags/example_dag.py with PID 21903 started at
2020-12-03T03:05:55.442709+00:00 has timed out, killing it.
  • Controlla i log dello scheduler per verificare che funzioni correttamente. In caso di problemi, nei log dell'organizzatore potresti visualizzare le seguenti voci di log:
DagFileProcessorManager (PID=732) last sent a heartbeat 240.09 seconds ago! Restarting it
Process timed out, PID: 68496

Soluzioni:

  • Correggi tutti gli errori di analisi dei DAG. Il processore DAG analizza più DAG e in rari casi di analisi degli errori di un DAG possono influire negativamente sull'analisi dei e altri DAG.

  • Se l'analisi del DAG richiede più secondi rispetto a quelli definiti in [core]dagrun_import_timeout, aumenta questo timeout.

  • Se l'analisi di tutti i DAG richiede più secondi di quelli definiti in [core]dag_file_processor_timeout, aumenta questo timeout.

  • Se l'analisi del DAG richiede molto tempo, può anche significare che non è implementato in modo ottimale. Ad esempio, se legge molte variabili di ambiente o esegue chiamate a servizi esterni o al database Airflow. Nella misura del possibile, evita di eseguire queste operazioni nelle sezioni globali dei DAG.

  • Aumenta le risorse di CPU e memoria per lo scheduler in modo che possa funzionare più velocemente.

  • Regola il numero di scheduler.

  • Aumenta il numero di processi del processore DAG in modo da eseguire l'analisi in modo più rapido. Puoi farlo aumentando il valore di [scheduler]parsing_process.

  • Riduci la frequenza di analisi dei DAG.

  • Riduci il carico sul database Airflow.

Sintomi del carico elevato del database Airflow

Per ulteriori informazioni, vedi Sintomi del database Airflow sotto pressione di carico.

Passaggi successivi