Cloud Composer 1 | Cloud Composer 2 | Cloud Composer 3
En este instructivo, se proporcionan pasos para depurar un DAG de Airflow con errores en Cloud Composer y diagnosticar problemas relacionados con los recursos de los trabajadores, como falta de memoria del trabajador o espacio de almacenamiento, con la ayuda de los registros y el entorno y supervisión de los datos.
Introducción
Este instructivo se enfoca en problemas relacionados con recursos para demostrar formas de depurar un en el DAG.
La falta de recursos de trabajadores asignados causa fallas en el DAG. Si una tarea de Airflow se queda sin memoria o almacenamiento, es posible que veas una excepción de Airflow, como las siguientes:
WARNING airflow.exceptions.AirflowException: Task received SIGTERM signal
INFO - Marking task as FAILED.
o
Task exited with return code Negsignal.SIGKILL
En esos casos, la recomendación general es aumentar el trabajador de Airflow recursos o reducir la cantidad de tareas por trabajador. Sin embargo, como las excepciones de Airflow pueden ser genéricas, puede ser un desafío identificar el recurso en particular que está causando el problema.
En este instructivo, se explica cómo puedes diagnosticar el motivo de una falla del DAG y Identificar el tipo de recurso que causa problemas depurando dos DAG de ejemplo que fallan debido a la falta de memoria y almacenamiento para los trabajadores.
Objetivos
Ejecuta DAG de ejemplo que fallan por los siguientes motivos:
- Falta de memoria del trabajador
- Falta de almacenamiento de los trabajadores
Diagnostica los motivos de la falla
Aumenta los recursos de trabajadores asignados
Prueba los DAG con los nuevos límites de recursos
Costos
En este instructivo, se usan los siguientes componentes facturables de Google Cloud:
- Cloud Composer (consulta los costos adicionales)
- Cloud Monitoring
Cuando finalices este instructivo, puedes borrar los recursos creados para evitar que se te siga facturando. Para obtener más información, consulta Cómo realizar una limpieza.
Antes de comenzar
En esta sección, se describen las acciones que debes realizar antes de comenzar el instructivo.
Crea y configura un proyecto
Para este instructivo, necesitas Google Cloud proyecto. Configura el proyecto de la siguiente manera:
En la consola de Google Cloud, selecciona o crea un proyecto:
Asegúrate de tener habilitada la facturación para tu proyecto. Obtén información para verificar si la facturación está habilitada en un proyecto.
Asegúrate de que el usuario de tu proyecto de Google Cloud tenga los siguientes roles para crear los recursos necesarios:
- Administrador de objetos de almacenamiento y entorno
(
roles/composer.environmentAndStorageObjectAdmin
) - Administrador de Compute (
roles/compute.admin
) - Editor de Monitoring (
roles/monitoring.editor
)
- Administrador de objetos de almacenamiento y entorno
(
Habilita las API para tu proyecto.
Enable the Cloud Composer API.
Crea tu entorno de Cloud Composer
Crea un entorno de Cloud Composer 2.
Como parte de la creación del entorno,
otorgas la extensión de agente de servicio de la API de Cloud Composer v2
(roles/composer.ServiceAgentV2Ext
) para el agente de servicio de Composer
de servicio predeterminada. Cloud Composer usa esta cuenta para realizar operaciones
en tu proyecto de Google Cloud.
Verifica los límites de recursos de los trabajadores
Verifica los límites de recursos de trabajadores de Airflow en tu entorno:
En la consola de Google Cloud, ve a la página Entornos.
En la lista de entornos, haz clic en el nombre de tu entorno. Se abrirá la página Detalles del entorno.
Ve a la pestaña Configuración del entorno.
Ve a Recursos > Configuración de cargas de trabajo. > Trabajador.
Verifica que los valores sean 0.5 CPU virtuales, 1.875 GB de memoria y 1 GB de almacenamiento. Estos son los límites de recursos de los trabajadores de Airflow con los que trabajarás en los próximos pasos de este instructivo.
Ejemplo: Diagnostica problemas de memoria insuficiente
Sube el siguiente DAG de muestra al entorno
que creaste en los pasos anteriores. En este instructivo, este DAG se llama
create_list_with_many_strings
Este DAG contiene una tarea que ejecuta los siguientes pasos:
- Crea una lista vacía
s
. - Ejecuta un ciclo para agregar la cadena
More
a la lista. - Imprime cuánta memoria consume la lista y espera 1 segundo cada 1 minutos de iteración.
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
)
Activa el DAG de ejemplo
Activa el DAG de muestra, create_list_with_many_strings
:
En la consola de Google Cloud, ve a la página Entornos.
En la columna Servidor web de Airflow, sigue el vínculo de Airflow para tu entorno.
En la interfaz web de Airflow, en la página DAG, en la columna Vínculos de tu DAG, haz clic en el botón Activar DAG.
Haz clic en Activar.
En la página DAG, haz clic en la tarea que activaste y revisa el resultado. para asegurarte de que tu DAG comenzó a ejecutarse.
Mientras se ejecuta la tarea, los registros de salida imprimirán el tamaño de memoria en GB que usa el DAG.
Después de varios minutos, la tarea fallará porque supera el trabajador de Airflow. de 1.875 GB.
Diagnostica el DAG con errores
Si estabas ejecutando varias tareas en el momento de la falla, considera ejecutar realizar una sola tarea y diagnosticar la presión de los recursos durante ese tiempo para identificar qué tareas presionan los recursos y cuáles debes aumentar.
Revisa los registros de tareas de Airflow
Observa que la tarea del DAG create_list_with_many_strings
tiene un estado Failed
.
Revisa los registros de la tarea. Verás la siguiente entrada de registro:
```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.
Revisa las cargas de trabajo
Revisa las cargas de trabajo para verificar que la carga de tu tarea no haga que el nodo en el que se ejecuta el Pod supere el límite de consumo de memoria:
En la consola de Google Cloud, ve a la página Entornos.
En la lista de entornos, haz clic en el nombre de tu entorno. Se abrirá la página Detalles del entorno.
Ve a la pestaña Configuración del entorno.
En Recursos > Clúster de GKE > Cargas de trabajo, haz clic en Ver cargas de trabajo del clúster.
Verifica si algunos de los Pods de la carga de trabajo tienen estados similares a los siguientes:
Error with exit code 137 and 1 more issue. ContainerStatusUnknown with exit code 137 and 1 more issue
Exit code 137
significa que un contenedor o un Pod intenta usar más memoria que la permitida. El proceso finaliza para evitar el uso de memoria.
Revisa el estado del entorno y la supervisión del consumo de recursos
Revisa el estado del entorno y la supervisión del consumo de recursos:
En la consola de Google Cloud, ve a la página Entornos.
En la lista de entornos, haz clic en el nombre de tu entorno. Se abrirá la página Detalles del entorno.
Ve a la pestaña Monitoring y selecciona Overview.
En el panel Descripción general del entorno, busca la Gráfico del estado del entorno (DAG de supervisión de Airflow) Contiene un ícono rojo que corresponde a la hora en que los registros comenzaron a imprimir errores.
Selecciona Trabajadores y busca el gráfico Uso de memoria total de trabajadores. Observa que la línea Memory usage (Uso de memoria) tiene un aumento repentino en el momento en que se estaba ejecutando la tarea.
Aunque la línea de uso de memoria del gráfico no alcanza el límite, cuando diagnostiques los motivos de la falla, debes tener en cuenta solo la memoria asignable, mientras que la línea Límite de memoria del gráfico representa la memoria total disponible (incluida la capacidad reservada por GKE).
En este ejemplo, el límite de memoria del trabajador se establece en 1.875 GB. GKE reserva el 25% de los primeros 4 GiB de memoria. GKE también reserva una cuota umbral de expulsión: 100 MiB de memoria en cada nodo para la expulsión de kubelet
La memoria asignable se calcula de la siguiente manera:
ALLOCATABLE = CAPACITY - RESERVED - EVICTION-THRESHOLD
Si el límite de memoria es de 1.875 GB, la memoria asignable real es la siguiente:
1.75 GiB (1.875GB) - 0.44 (25% GiB reserved) - 0.1 = 1.21 GiB (~1.3 GB).
Cuando adjuntes este límite real al gráfico de uso de memoria, verás que el aumento repentino del uso de memoria de la tarea alcanza el límite real de memoria y puedes concluir que la tarea falló debido a que no había suficiente memoria para los trabajadores.
Aumenta el límite de memoria de trabajador
Asigna memoria adicional para el trabajador para que el DAG de muestra se realice correctamente:
En la consola de Google Cloud, ve a la página Entornos.
En la lista de entornos, haz clic en el nombre de tu entorno. Se abrirá la página Detalles del entorno.
Ve a la pestaña Configuración del entorno.
Busca la configuración de Recursos > Cargas de trabajo y haz clic en Editar.
En la sección Worker, en el campo Memory, especifica la memoria nueva para los trabajadores de Airflow. En este instructivo, usarás 3 GB.
Guarda los cambios y espera unos minutos para que los trabajadores de Airflow se reinicien.
Prueba tu DAG con el nuevo límite de memoria
Vuelve a activar el DAG create_list_with_many_strings
y espera hasta que
termine de ejecutarse.
En los registros de salida de la ejecución de tu DAG, verás
Marking task as SUCCESS
, y el estado de la tarea indicará Success.Revisa la sección Descripción general del entorno en la pestaña Supervisión y asegúrate de que no haya áreas rojas.
Haz clic en la sección Trabajadores y busca el gráfico Uso total de la memoria de trabajadores. Verás que la línea Límite de memoria refleja el cambio en límite de memoria, y la línea Memory usage está muy por debajo del valor real de memoria asignable.
Ejemplo: Diagnostica problemas de falta de almacenamiento
En este paso, subirás dos DAG que crean archivos grandes. El primer DAG crea un archivo grande. El segundo DAG crea un archivo grande e imita un una operación de larga duración.
El tamaño del archivo en ambos DAG supera el almacenamiento predeterminado de trabajadores de Airflow límite de 1 GB, pero el segundo DAG tiene una tarea de espera adicional para extender su de manera artificial.
Investigarás las diferencias en el comportamiento de ambos DAG en la próxima pasos.
Sube un DAG que cree un archivo grande
Sube el siguiente DAG de muestra al entorno
que creaste en los pasos anteriores. En este instructivo, este DAG se llama
create_large_txt_file_print_logs
Este DAG contiene una tarea que ejecuta los siguientes pasos:
- Escribe un archivo
localfile.txt
de 1.5 GB en el almacenamiento del trabajador de Airflow. - Imprime el tamaño del archivo creado mediante el módulo
os
de Python. - Imprime la duración de la ejecución de DAG cada 1 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)
Sube un DAG que cree un archivo grande en una operación de larga duración
Para imitar un DAG de larga duración y analizar el impacto de la duración de la tarea
en el estado final, sube el segundo DAG de muestra a tu
entorno. En este instructivo, este DAG se llama
long_running_create_large_txt_file_print_logs
Este DAG contiene una tarea que ejecuta los siguientes pasos:
- Escribe un archivo
localfile.txt
de 1.5 GB en el almacenamiento del trabajador de Airflow. - Imprime el tamaño del archivo creado con el módulo
os
de Python. - Espera 1 hora 15 minutos para imitar el tiempo necesario para las operaciones con el archivo, por ejemplo, cuando lo lea.
- Imprime la duración de la ejecución del DAG cada 1 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)
Activa DAG de muestra
Activa el primer DAG, create_large_txt_file_print_logs
:
En la consola de Google Cloud, ve a la página Entornos.
En la columna Servidor web de Airflow, sigue el vínculo de Airflow para tu entorno.
En la interfaz web de Airflow, en la página DAG, en la columna Vínculos de tu DAG, haz clic en el botón Activar DAG.
Haz clic en Activar.
En la página DAGs, haz clic en la tarea que activaste y revisa los registros de salida para asegurarte de que tu DAG comenzó a ejecutarse.
Espera hasta que aparezca la tarea que creaste con el botón Se completó
create_large_txt_file_print_logs
DAG. Esto puede tardar varias minutos.En la página DAGs, haz clic en la ejecución del DAG. Verás que tu tarea tiene un estado
Success
, a pesar de que se superó el límite de almacenamiento.
Revisa los registros de Airflow de la tarea:
En la consola de Google Cloud, ve a la página Entornos.
En la lista de entornos, haz clic en el nombre de tu entorno. Se abrirá la página Detalles del entorno.
Ve a la pestaña Registros y, luego, a Todos los registros > Registros de Airflow > Workers > Ver en el Explorador de registros.
Filtra los registros por tipo: muestra solo los mensajes de Error.
En los registros, verás mensajes similares a los siguientes:
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.
Estos registros indican que el pod inició el proceso de "apagado en caliente" porque el almacenamiento utilizado superó el límite y se eliminó en 1 hora. Sin embargo, la ejecución de DAG no falló porque se completó dentro del período de gracia de finalización de Kubernetes, que se explica más adelante en este instructivo.
Para ilustrar el concepto del período de gracia de rescisión, revisa el resultado
del segundo DAG de muestra, long_running_create_large_txt_file_print_logs
.
Activa el segundo DAG, long_running_create_large_txt_file_print_logs
:
En la consola de Google Cloud, ve a la página Entornos.
En la columna Servidor web de Airflow, sigue el vínculo de Airflow para tu entorno.
En la interfaz web de Airflow, en la página DAG, en la columna Vínculos de tu DAG, haz clic en el botón Activar DAG.
Haz clic en Activar.
En la página DAG, haz clic en la tarea que activaste y revisa el resultado. para asegurarte de que tu DAG comenzó a ejecutarse.
Espera hasta que falle la ejecución del DAG de
long_running_create_large_txt_file_print_logs
. El proceso tardará aproximadamente una hora.
Revisa los resultados de la ejecución del DAG:
En la página DAG, haz clic en el
long_running_create_large_txt_file_print_logs
ejecución de DAG. Verás lo siguiente: que la tarea tenga un estadoFailed
y que la duración de la ejecución se haya exactamente 1 hora y 5 minutos, que es menos que el período de espera de la tarea de 1 hora y 15 minutos.Revisa los registros de la tarea. Una vez que el DAG haya creado el archivo
localfile.txt
en el contenedor del trabajador de Airflow, se imprimirá el registro que inició el DAG en espera, y la duración de la ejecución se imprime en los registros de tareas cada 1 minuto. En este ejemplo, el DAG imprime el registrolocalfile.txt size:
y el tamaño del archivolocalfile.txt
serán de 1.5 GB.
Cuando el archivo escrito en el contenedor del trabajador de Airflow supera el almacenamiento límite, se supone que la ejecución del DAG fallará. Sin embargo, la tarea no falla de inmediato y sigue ejecutándose hasta que su duración alcanza 1 hora y 5 minutos. Esto sucede porque Kubernetes no finaliza la tarea de inmediato y sigue ejecutándose para permitir 1 hora de tiempo de recuperación, conocido como “período de gracia de finalización”. Una vez que un nodo se queda sin recursos, Kubernetes no finaliza la Pod de inmediato para controlar la finalización con facilidad, de modo que haya una impacto en el usuario final.
El período de gracia de finalización ayuda a los usuarios a recuperar archivos después de que fallan las tareas. Sin embargo, puede generar confusión en el diagnóstico de los DAG. Cuando se supera el límite de almacenamiento del trabajador de Airflow, el estado de la tarea final depende de la duración de la ejecución del DAG:
Si la ejecución de DAG supera el límite de almacenamiento del trabajador, pero se completa en menos de 1 hora, la tarea se completa con un estado
Success
porque se completó dentro del período de gracia de la finalización. Sin embargo, Kubernetes finaliza el Pod y el archivo escrito se borra del contenedor de inmediato.Si el DAG supera el límite de almacenamiento de trabajadores y se ejecuta durante más de 1 hora, el DAG se sigue ejecutando durante 1 hora y puede superar el límite de almacenamiento miles de por ciento antes de que Kubernetes elimine las marcas del Pod y Airflow la tarea como
Failed
.
Diagnostica el DAG que falló
Si estabas ejecutando varias tareas en el momento de la falla, considera ejecutar solo una tarea y diagnosticar la presión de recursos durante ese tiempo para identificar qué tareas causan presión de recursos y qué recursos debes aumentar.
Revisa los registros de tareas del segundo DAG, long_running_create_large_txt_file_print_logs
:
En la consola de Google Cloud, ve a la página Entornos.
En la lista de entornos, haz clic en el nombre de tu entorno. Se abrirá la página Detalles del entorno.
Ve a la pestaña Registros y, luego, a Todos los registros > Registros de Airflow > Workers > Ver en el Explorador de registros.
Filtra los registros por tipo: muestra solo los mensajes de Error.
En los registros, verás mensajes similares a los siguientes:
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.
Estos mensajes indican que, a medida que avanzaba la tarea, los registros de Airflow comenzaron a imprimir errores cuando el tamaño de los archivos que generaba tu DAG superó el límite de almacenamiento del trabajador y comenzó el período de gracia de rescisión. Durante el el período de gracia de terminación, el consumo de almacenamiento no volvió al límite lo que generó la expulsión del Pod después de que finalizara el período de gracia.
Revisa el estado del entorno y la supervisión del consumo de recursos:
En la consola de Google Cloud, ve a la página Entornos.
En la lista de entornos, haz clic en el nombre de tu entorno. Se abrirá la página Detalles del entorno.
Ve a la pestaña Monitoring y selecciona Overview.
En el panel Descripción general del entorno, busca la Gráfico del estado del entorno (DAG de supervisión de Airflow) Contiene un ícono rojo que corresponde a la hora en que los registros comenzaron a imprimir errores.
Selecciona Trabajadores y busca el gráfico Uso total del disco de trabajadores. Observa que la línea Uso de disco tiene un aumento repentino y supera la línea Límite de disco en el momento en que se ejecutaba la tarea.
Aumenta el límite de almacenamiento de los trabajadores
Asigna almacenamiento adicional de trabajadores de Airflow para que el DAG de muestra haga lo siguiente: tener éxito:
En la consola de Google Cloud, ve a la página Entornos.
En la lista de entornos, haz clic en el nombre de tu entorno. Se abrirá la página Detalles del entorno.
Ve a la pestaña Configuración del entorno.
Busca la configuración Recursos > Cargas de trabajo y selecciona Haz clic en Editar.
En la sección Trabajador, en el campo Almacenamiento, especifica el almacenamiento nuevo. para los trabajadores de Airflow. En este instructivo, establécelo en 2 GB.
Guarda los cambios y espera varios minutos para que tus trabajadores de Airflow reiniciar.
Prueba tu DAG con el nuevo límite de almacenamiento
Vuelve a activar el DAG long_running_create_large_txt_file_print_logs
y
esperar 1 hora y 15 minutos hasta que termine de ejecutarse.
En los registros de salida de tu ejecución del DAG, verás
Marking task as SUCCESS
, y el estado de la tarea indicará Success, con una duración de 1 hora y 15 minutos, que equivale al tiempo de espera establecido en el código del DAG.Revisa la sección Descripción general del entorno en la pestaña Supervisión y asegúrate de que no haya áreas rojas.
Haz clic en la sección Trabajadores y busca el gráfico Uso total del disco de trabajadores. Verás que la línea Límite de discos refleja el cambio en la de almacenamiento y la línea Uso del disco dentro del rango permitido.
Resumen
En este instructivo, diagnosticaste el motivo de una falla de DAG y, a través de la depuración de dos DAG de ejemplo que fallan debido a una falta de memoria y almacenamiento de trabajadores, identificaste el tipo de recurso que causa presión. Luego, ejecutaste los DAG correctamente después de asignar más memoria y almacenamiento a tus trabajadores. Sin embargo, es recomendado para Optimiza tus DAG (flujos de trabajo) reducir el consumo de recursos de los trabajadores, ya que no es lo que es posible aumentar los recursos más allá de cierto umbral.
Limpia
Para evitar que se apliquen cargos a tu cuenta de Google Cloud por los recursos usadas en este instructivo, borra el proyecto que contiene los recursos o conservarlo y borrar los recursos individuales.
Borra el proyecto
- 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.
Borra los recursos individuales
Si planeas explorar varios instructivos y guías de inicio rápido, la reutilización de proyectos puede ayudarte a evitar exceder los límites de las cuotas del proyecto.
Borra el entorno de Cloud Composer. También borrar el bucket del entorno durante el procedimiento.