Soluciona problemas del procesador de DAG

Managed Airflow (3ª gen.) | Managed Airflow (2ª gen.) | Managed Airflow (1ª gen. heredada)

En esta página, solo se abordan los problemas relacionados con el procesamiento de archivos DAG. Para obtener información sobre los problemas de programación de tareas, consulta Solución de problemas del programador de Airflow.

Flujo de trabajo para la solución de problemas

Inspecciona los registros del procesador de DAG

Si tienes DAG complejos, es posible que los procesadores de DAG de Airflow no analicen todos tus DAG. Esto puede generar muchos problemas que tienen los siguientes síntomas.

Síntomas:

• Si un procesador de DAG tiene problemas cuando analiza tus DAG, es posible que se produzca una combinación de los problemas enumerados. Si los DAG se generan de forma dinámica, estos problemas pueden ser más potentes en comparación con los DAG estáticos.

  • Los DAG no son visibles en la IU de Airflow ni en la IU de DAG.
  • Los DAG no están programados para su ejecución.
  • Hay errores en los registros del procesador de DAG, por ejemplo:

  dag-processor-manager [2023-04-21 21:10:44,510] {manager.py:1144} ERROR -
  Processor for /home/airflow/gcs/dags/dag-example.py with PID 68311 started
  at 2023-04-21T21:09:53.772793+00:00 has timed out, killing it.
  

  o

  dag-processor-manager [2023-04-26 06:18:34,860] {manager.py:948} ERROR -
  Processor for /home/airflow/gcs/dags/dag-example.py exited with return
  code 1.
  

• Se cancelan las tareas de Airflow programadas para su ejecución, y las ejecuciones de DAG para los DAG que no se pudieron analizar pueden marcarse como failed. Por ejemplo:

  airflow-scheduler Failed to get task '<TaskInstance: dag-example.task1--1
  manual__2023-04-17T10:02:03.137439+00:00 [removed]>' for dag
  'dag-example'. Marking it as removed.
  

Solución:

• Aumenta los parámetros relacionados con el análisis de DAG:

  • Aumenta [core]dagbag_import_timeout a al menos 120 segundos (o más, si es necesario).

  • Increase [core]dag_file_processor_timeout a al menos 180 segundos (o más, si es necesario). Este valor debe ser superior a [core]dagbag_import_timeout.

• Corrige o quita los DAG que causan problemas a los procesadores de DAG.

Inspecciona los tiempos de análisis del DAG

Para verificar si el problema ocurre durante el análisis del DAG, sigue estos pasos.

gcloud composer environments run ENVIRONMENT_NAME \
    --location LOCATION \
    dags report

file                  | duration       | dag_num | task_num | dags
======================+================+=========+==========+===================
/manydagsbig.py       | 0:00:00.038334 | 2       | 10       | serial-0,serial-0
/airflow_monitoring.py| 0:00:00.001620 | 1       | 1        | airflow_monitoring

Soluciona problemas durante el análisis de DAG

En las siguientes secciones, se describen los síntomas y las posibles soluciones para algunos problemas comunes durante el análisis de DAG.

Haz que el procesador de DAG ignore archivos innecesarios

Para que el procesador de DAG de Airflow ignore archivos innecesarios, haz lo siguiente:

1. Crea un archivo .airflowignore.
2. En este archivo, se muestra una lista de los archivos y las carpetas que se deben ignorar.
3. Sube este archivo a la carpeta /dags en el bucket de tu entorno.

Para obtener más información sobre el formato de archivo .airflowignore, consulta la documentación de Airflow.

Airflow procesa los DAG pausados

Puedes pausar los DAG para detener su ejecución. Esto ahorra recursos de los trabajadores de Airflow.

Los procesadores de DAG de Airflow continúan analizando los DAG pausados. Si deseas mejorar el rendimiento de los procesadores de DAG, usa .airflowignore o borra los DAG pausados de la carpeta de DAG.

Problemas comunes

En las siguientes secciones, se describen los síntomas y las posibles soluciones para algunos problemas comunes de análisis.

Tiempo de espera de importación de carga de DAG

Síntoma:

• En la interfaz web de Airflow, en la parte superior de la página de la lista de DAG, aparece un cuadro de alerta rojo que muestra Broken DAG: [/path/to/dagfile] Timeout.

• En Cloud Monitoring, los registros de airflow-scheduler contienen entradas similares a las siguientes:

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

Solución:

Anula la dag_file_processor_timeout configuración de Airflow y permite más tiempo para el análisis del DAG.

Un DAG no es visible en la IU de Airflow ni en la IU de DAG, y el programador no lo programa

El procesador de DAG analiza cada DAG antes de que el programador pueda programarlo y antes de que un DAG se vuelva visible en la IU de Airflow o la IU de DAG.

Las siguientes opciones de configuración de Airflow definen los tiempos de espera para analizar los DAG:

• [core]dagbag_import_timeout define la cantidad de tiempo que tiene el procesador de DAG para analizar un solo DAG.

• [core]dag_file_processor_timeout define la cantidad total de tiempo que el procesador de DAG puede dedicar a analizar todos los DAG.

Si un DAG no es visible en la IU de Airflow ni en la IU de DAG, haz lo siguiente:

• Verifica los registros del procesador de DAG para ver si el procesador de DAG puede procesar correctamente tu DAG. En caso de problemas, es posible que veas las siguientes entradas de registro en los registros del procesador de DAG o del programador:

  [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.
  

• Verifica los registros del programador para ver si el programador funciona correctamente. En caso de problemas, es posible que veas las siguientes entradas de registro en los registros del programador:

  DagFileProcessorManager (PID=732) last sent a heartbeat 240.09 seconds ago! Restarting it
  Process timed out, PID: 68496
  

Soluciones:

• Corrige todos los errores de análisis de DAG. El procesador de DAG analiza varios DAG y, en casos excepcionales, los errores de análisis de un DAG pueden afectar negativamente el análisis de otros DAG.

• Si el análisis de tu DAG tarda más que la cantidad de segundos definida en [core]dagbag_import_timeout, aumenta este tiempo de espera.

• Si el análisis de todos tus DAG tarda más que la cantidad de segundos definida en [core]dag_file_processor_timeout, aumenta este tiempo de espera.

• Si tu DAG tarda mucho en analizarse, también puede significar que no se implementa de manera óptima. Por ejemplo, si lee muchas variables de entorno o realiza llamadas a servicios externos o a la base de datos de Airflow. En la medida de lo posible, evita realizar estas operaciones en las secciones globales de los DAG.

• Aumenta los recursos de CPU y memoria para el procesador de DAG para que pueda funcionar más rápido.

• Disminuye la frecuencia del análisis de DAG.

• Reduce la carga en la base de datos de Airflow.

¿Qué sigue?

• Soluciona problemas de sincronización de archivos
• Solución de problemas del programador de Airflow
• Soluciona problemas de los DAG