Cómo depurar problemas de programación de tareas

Cloud Composer 3 | Cloud Composer 2 | Cloud Composer 1

Nota: Esta página aún no se revisó para Cloud Composer 3 y muestra contenido de Cloud Composer 2.

En este instructivo, se te guiará para diagnosticar y solucionar problemas de programación y análisis de tareas que generan fallas en el programador, errores de análisis y latencia, y fallas en las tareas.

Introducción

El programador de Airflow se ve afectado principalmente por dos factores: la programación de tareas y el análisis de DAG. Los problemas en uno de esos factores pueden tener un impacto negativo en el estado y el rendimiento del entorno.

A veces, se programan demasiadas tareas de forma simultánea. En esta situación, la fila se llena y las tareas permanecen en el estado “programado” o se vuelven a programar después de estar en fila, lo que puede causar fallas en las tareas y latencia de rendimiento.

Otro problema común es la latencia de análisis y los errores causados por la complejidad de un código DAG. Por ejemplo, un código de DAG que contiene variables de Airflow en el nivel superior del código puede generar demoras en el análisis, sobrecarga de la base de datos, fallas de programación y tiempos de espera de DAG.

En este instructivo, diagnosticarás los DAG de ejemplo y aprenderás a solucionar problemas de programación y análisis, mejorar la programación de DAG y optimizar el código de DAG y las configuraciones de entorno para mejorar el rendimiento.

Objetivos

En esta sección, se enumeran los objetivos de los ejemplos de este instructivo.

Ejemplo: Funcionamiento incorrecto del programador y latencia causada por una alta simultaneidad de tareas

• Sube el DAG de muestra que se ejecuta varias veces de forma simultánea y diagnostica los problemas de latencia y mal funcionamiento del programador con Cloud Monitoring.

• Consolida las tareas y evalúa el impacto en el rendimiento para optimizar el código de tu DAG.

• Distribuye las tareas de manera más uniforme en el tiempo y evalúa el impacto en el rendimiento.

• Optimiza las configuraciones de Airflow y del entorno, y evalúa el impacto.

Ejemplo: errores de análisis de DAG y latencia causados por un código complejo

• Sube el DAG de muestra con las variables de Airflow y diagnostica los problemas de análisis con Cloud Monitoring.

• Para optimizar el código DAG, evita las variables de Airflow en el nivel superior del código y evalúa el impacto en el tiempo de análisis.

• Optimiza las configuraciones de Airflow y del entorno, y evalúa el impacto en el tiempo de análisis.

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 un Google Cloud proyecto. Configura el proyecto de la siguiente manera:

1. En la consola de Google Cloud, selecciona o crea un proyecto:

   Ir al Selector de proyectos

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

3. Asegúrate de que el Google Cloud usuario de tu proyecto 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)

Habilita las API para tu proyecto.

Enable the Cloud Composer API.

Enable the API

Crea tu entorno de Cloud Composer

Crea un entorno de Cloud Composer 2.

Importante: Asegúrate de usar el parámetro de configuración predeterminado de entorno pequeño para reproducir los errores cuando uses los ejemplos que se proporcionan más adelante en este instructivo. Este es el parámetro de configuración predeterminado cuando se crea un entorno a través de la consola de Google Cloud.

Como parte de la creación del entorno, otorgas el rol Extensión del agente de servicio de la API de Cloud Composer v2 (roles/composer.ServiceAgentV2Ext) a la cuenta del agente de servicio de Composer. Cloud Composer usa esta cuenta para realizar operaciones en tu Google Cloud proyecto.

Ejemplo: El programador funciona mal y la tarea falla debido a problemas de programación

En este ejemplo, se muestra el mal funcionamiento del programador de depuración y la latencia causada por una alta simultaneidad de tareas.

Sube el DAG de muestra a tu entorno

Sube el siguiente DAG de muestra al entorno que creaste en los pasos anteriores. En este instructivo, este DAG se llama dag_10_tasks_200_seconds_1.

Este DAG tiene 200 tareas. Cada tarea espera 1 segundo y muestra el mensaje "Complete!". El DAG se activa automáticamente una vez que se sube. Cloud Composer ejecuta este DAG 10 veces, y todas las ejecuciones de DAG se realizan en paralelo.

Advertencia: Este DAG simula problemas con la programación de DAG, lo que puede interrumpir las operaciones en tu entorno.

import time
from datetime import datetime, timedelta

from airflow import DAG
from airflow.decorators import task


tasks_amount = 200
seconds = 1
minutes = 5

with DAG(
    dag_id=f"dag_10_tasks_{tasks_amount}_sec_{seconds}",
    start_date=datetime(2023, 11, 22, 20, 0),
    end_date=datetime(2023, 11, 22, 20, 49),
    schedule_interval=timedelta(minutes=minutes),
    catchup=True,
) as dag:

    @task
    def create_subtasks(seconds: int) -> None:
        time.sleep(seconds)

    for i in range(tasks_amount):
        create_subtasks(seconds)

Diagnostica los problemas de funcionamiento incorrecto del programador y de fallas de tareas

Una vez que se complete la ejecución del DAG, abre la IU de Airflow y haz clic en el DAG dag_10_tasks_200_seconds_1. Verás que se ejecutaron correctamente 10 ejecuciones de DAG en total y que cada una tiene 200 tareas que se ejecutaron correctamente.

Revisa los registros de tareas de Airflow:

1. En la consola de Google Cloud, ve a la página Entornos.

   Ir a Entornos

2. En la lista de entornos, haz clic en el nombre de tu entorno. Se abrirá la página Detalles del entorno.

3. Ve a la pestaña Registros y, luego, a Todos los registros > Registros de Airflow > Trabajadores > Ver en el Explorador de registros.

En el histograma de registros, puedes ver los errores y las advertencias indicados con colores rojo y naranja:

Figura 1: Histograma de registros del trabajador de Airflow (haz clic para ampliar)

El DAG de ejemplo generó alrededor de 130 advertencias y 60 errores. Haz clic en cualquier columna que contenga barras amarillas y rojas. Verás algunas de las siguientes advertencias y errores en los registros:

State of this instance has been externally set to success. Terminating
instance.

Received SIGTERM. Terminating subprocesses.

worker: Warm shutdown (MainProcess).

Estos registros pueden indicar que el uso de recursos superó los límites y que el trabajador se reinició.

Si una tarea de Airflow se mantiene en la cola durante demasiado tiempo, el programador la marca como fallida y up_for_retry, y la reprogramará una vez más para su ejecución. Una forma de observar los síntomas de esta situación es mirar el gráfico con la cantidad de tareas en cola y, si los picos de este gráfico no disminuyen en aproximadamente 10 minutos, es probable que haya fallas en las tareas (sin registros).

Revisa la información de supervisión:

1. Ve a la pestaña Supervisión y selecciona Descripción general.

2. Revisa el gráfico de tareas de Airflow.

   

   Figura 2: Gráfico de tareas de Airflow (haz clic para ampliar)

   En el gráfico de tareas de Airflow, hay un aumento repentino en las tareas en cola que dura más de 10 minutos, lo que podría significar que no hay suficientes recursos en tu entorno para procesar todas las tareas programadas.

3. Revisa el gráfico Trabajadores activos:

   

   Figura 3: Gráfico de trabajadores activos (haz clic para agrandar)

   El gráfico Trabajadores activos indica que el DAG activó el ajuste de escala automático al límite máximo permitido de tres trabajadores durante la ejecución del DAG.

4. Los gráficos de uso de recursos pueden indicar la falta de capacidad en los trabajadores de Airflow para ejecutar tareas en cola. En la pestaña Supervisión, selecciona Trabajadores y revisa los gráficos Uso total de CPU de los trabajadores y Uso total de memoria de los trabajadores.

   

   Figura 4: Gráfico del uso total de la CPU de los trabajadores (haz clic para ampliar)

   

   Figura 5. Gráfico del uso total de la memoria de los trabajadores (haz clic para ampliar)

   Los gráficos indican que la ejecución de demasiadas tareas de forma simultánea resultó en alcanzar el límite de la CPU. Los recursos se habían usado durante más de 30 minutos, lo que es incluso más largo que la duración total de 200 tareas en 10 ejecuciones de DAG que se ejecutan una por una.

Estos son los indicadores de que la cola se está llenando y de la falta de recursos para procesar todas las tareas programadas.

Consolida tus tareas

El código actual crea muchos DAG y tareas sin recursos suficientes para procesar todas las tareas en paralelo, lo que hace que la cola se llene. Mantener las tareas en la cola durante demasiado tiempo puede hacer que se reprogramen o fallen. En tales casos, debes elegir una cantidad menor de tareas más consolidadas.

En el siguiente DAG de muestra, se cambia la cantidad de tareas del ejemplo inicial de 200 a 20 y se aumenta el tiempo de espera de 1 a 10 segundos para imitar tareas más consolidadas que realizan la misma cantidad de trabajo.

Sube el siguiente DAG de muestra al entorno que creaste. En este instructivo, este DAG se llama dag_10_tasks_20_seconds_10.

Advertencia: Este DAG simula problemas con la programación de DAG, lo que puede interrumpir las operaciones en tu entorno.

import time
from datetime import datetime, timedelta

from airflow import DAG
from airflow.decorators import task


tasks_amount = 20
seconds = 10
minutes = 5

with DAG(
    dag_id=f"dag_10_tasks_{tasks_amount}_sec_{seconds}",
    start_date=datetime(2021, 12, 22, 20, 0),
    end_date=datetime(2021, 12, 22, 20, 49),
    schedule_interval=timedelta(minutes=minutes),
    catchup=True,
) as dag:

    @task
    def create_subtasks(seconds: int) -> None:
        time.sleep(seconds)

    for i in range(tasks_amount):
        create_subtasks(seconds)

Evalúa el impacto de las tareas más consolidadas en los procesos de programación:

1. Espera a que se completen las ejecuciones de DAG.

2. En la IU de Airflow, en la página DAGs, haz clic en el DAG dag_10_tasks_20_seconds_10. Verás 10 ejecuciones de DAG, cada una con 20 tareas que se completaron correctamente.

3. En la consola de Google Cloud, ve a la página Entornos.

   Ir a Entornos

4. En la lista de entornos, haz clic en el nombre de tu entorno. Se abrirá la página Detalles del entorno.

5. Ve a la pestaña Registros y, luego, a Todos los registros > Registros de Airflow > Trabajadores > Ver en el Explorador de registros.

   El segundo ejemplo con tareas más consolidadas generó aproximadamente 10 advertencias y 7 errores. En el histograma, puedes comparar la cantidad de errores y advertencias en el ejemplo inicial (valores anteriores) y en el segundo ejemplo (valores posteriores).

   

   Figura 6: Histograma de registros del trabajador de Airflow después de que se consolidaron las tareas (haz clic para ampliar)

   Cuando comparas el primer ejemplo con el más consolidado, puedes ver que hay muchos menos errores y advertencias en el segundo ejemplo. Sin embargo, los mismos errores relacionados con el cierre en caliente siguen apareciendo en los registros debido a la sobrecarga de recursos.

6. En la pestaña Supervisión, selecciona Trabajadores y revisa los gráficos.

   Cuando comparas el gráfico de tareas de Airflow del primer ejemplo (valores anteriores) con el gráfico del segundo ejemplo con tareas más consolidadas, puedes ver que el aumento repentino en las tareas en cola duró un período más corto cuando las tareas estaban más consolidadas. Sin embargo, duró cerca de 10 minutos, lo que sigue siendo poco óptimo.

   

   Figura 7: Gráfico de tareas de Airflow después de que se consolidaron las tareas (haz clic para ampliar)

   En el gráfico Trabajadores activos, puedes ver que el primer ejemplo (en el lado izquierdo del gráfico) usó recursos durante un período mucho más extenso que el segundo, a pesar de que ambos ejemplos imitan la misma cantidad de trabajo.

   

   Figura 8: Gráfico de trabajadores activos después de que se consolidaron las tareas (haz clic para ampliar)

   Revisa los gráficos de consumo de recursos de los trabajadores. Aunque la diferencia entre los recursos utilizados en el ejemplo con tareas más consolidadas y el ejemplo inicial es bastante significativa, el uso de la CPU sigue aumentando hasta el 70% del límite.

   

   Figura 9: Gráfico del uso total de CPU de los trabajadores después de que se consolidaron las tareas (haz clic para ampliar)

   

   Figura 10: Gráfico del uso total de la memoria de los trabajadores después de que se consolidaron las tareas (haz clic para ampliar)

Distribuye las tareas de forma más uniforme en el tiempo

Demasiadas tareas simultáneas hacen que la cola se llene, lo que genera que las tareas se bloqueen en la cola o se reprogramen. En los pasos anteriores, reduciste la cantidad de tareas consolidándolas. Sin embargo, los registros de salida y la supervisión indicaron que la cantidad de tareas simultáneas aún es poco óptima.

Puedes controlar la cantidad de ejecuciones de tareas simultáneas implementando un programa o estableciendo límites para la cantidad de tareas que se pueden ejecutar de forma simultánea.

En este instructivo, distribuirás las tareas de manera más uniforme con el tiempo agregando parámetros a nivel del DAG al DAG dag_10_tasks_20_seconds_10:

1. Agrega el argumento max_active_runs=1 al administrador de contexto de DAG. Este argumento establece un límite de una sola instancia de una ejecución de DAG en un momento determinado.

2. Agrega el argumento max_active_tasks=5 al administrador de contexto de DAG. Este argumento controla la cantidad máxima de instancias de tareas que se pueden ejecutar de forma simultánea en cada DAG.

Sube el siguiente DAG de muestra al entorno que creaste. En este instructivo, este DAG se llama dag_10_tasks_20_seconds_10_scheduled.py.

import time
from datetime import datetime, timedelta

from airflow import DAG
from airflow.decorators import task


tasks_amount = 20
seconds = 10
minutes = 5
active_runs = 1
active_tasks = 5


with DAG(
    dag_id=f"dag_10_tasks_{tasks_amount}_sec_{seconds}_runs_{active_runs}_tasks_{active_tasks}",
    start_date=datetime(2021, 12, 22, 20, 0),
    end_date=datetime(2021, 12, 22, 20, 49),
    schedule_interval=timedelta(minutes=minutes),
    max_active_runs=active_runs,
    max_active_tasks=active_tasks,
    catchup=True,
) as dag:

    @task
    def create_subtasks(seconds: int) -> None:
        time.sleep(seconds)

    for i in range(tasks_amount):
        create_subtasks(seconds)

Evalúa el impacto de distribuir tareas a lo largo del tiempo en los procesos de programación:

1. Espera a que se completen las ejecuciones de DAG.

2. En la consola de Google Cloud, ve a la página Entornos.

   Ir a Entornos

3. En la lista de entornos, haz clic en el nombre de tu entorno. Se abrirá la página Detalles del entorno.

4. Ve a la pestaña Registros y, luego, a Todos los registros > Registros de Airflow > Trabajadores > Ver en el Explorador de registros.

5. En el histograma, puedes ver que el tercer DAG con una cantidad limitada de tareas y ejecuciones activas no generó advertencias ni errores, y la distribución de registros se ve más uniforme en comparación con los valores anteriores.

   

   Figura 11: Histograma de registros de trabajadores de Airflow después de que las tareas se consolidaron y distribuyeron a lo largo del tiempo (haz clic para ampliar)

Las tareas del ejemplo de dag_10_tasks_20_seconds_10_scheduled que tienen una cantidad limitada de tareas y ejecuciones activas no causaron presión de recursos porque las tareas se pusieron en cola de manera uniforme.

Después de realizar los pasos descritos, consolidaste tareas pequeñas y las distribuyeste de manera más uniforme con el tiempo para optimizar el uso de recursos.

Optimiza la configuración de los entornos

Puedes ajustar la configuración de tu entorno para asegurarte de que siempre haya capacidad en los trabajadores de Airflow para ejecutar tareas en cola.

Cantidad de trabajadores y simultaneidad de trabajadores

Puedes ajustar la cantidad máxima de trabajadores para que Cloud Composer escale automáticamente tu entorno dentro de los límites establecidos.

El parámetro [celery]worker_concurrency define la cantidad máxima de tareas que un solo trabajador puede recuperar de la lista de tareas en cola. Si cambias este parámetro, se ajusta la cantidad de tareas que un solo trabajador puede ejecutar al mismo tiempo. Puedes cambiar esta opción de configuración de Airflow si la anulas. De forma predeterminada, la simultaneidad del trabajador se establece en un mínimo de lo siguiente: 32, 12 * worker_CPU, 8 * worker_memory, lo que significa que depende de los límites de recursos del trabajador. Consulta Cómo optimizar los entornos para obtener más información sobre los valores de simultaneidad predeterminados de los trabajadores.

La cantidad de trabajadores y la simultaneidad de los trabajadores funcionan en combinación entre sí, y el rendimiento de tu entorno depende en gran medida de ambos parámetros. Puedes usar las siguientes consideraciones para elegir la combinación correcta:

• Varias tareas rápidas que se ejecutan en paralelo Puedes aumentar la simultaneidad de los trabajadores cuando hay tareas en espera en la cola y tus trabajadores usan un porcentaje bajo de sus CPUs y memoria al mismo tiempo. Sin embargo, en ciertos casos, es posible que la cola nunca se llene, lo que provocará que el ajuste de escala automático nunca se active. Si las tareas pequeñas terminan de ejecutarse cuando los trabajadores nuevos están listos, un trabajador existente puede retomar las tareas restantes y no habrá tareas para los trabajadores creados recientemente.

  En estas situaciones, se recomienda aumentar la cantidad mínima de trabajadores y aumentar la simultaneidad de los trabajadores para evitar una escalamiento excesivo.

• Varias tareas largas que se ejecutan en paralelo La alta simultaneidad de los trabajadores prevente que el sistema escale la cantidad de trabajadores. Si varias tareas requieren muchos recursos y tardan mucho tiempo en completarse, una alta simultaneidad de trabajadores puede provocar que la cola nunca se llene y que un solo trabajador tome todas las tareas, lo que genera problemas de rendimiento. En estas situaciones, se recomienda aumentar la cantidad máxima de trabajadores y disminuir la simultaneidad de los trabajadores.

Importante: En algunos casos, configurar el parámetro [celery]worker_concurrency en un valor más alto podría no ser lo más adecuado para tu entorno. Antes de modificar la configuración de simultaneidad del trabajador, consulta las consideraciones de rendimiento.

La importancia del paralelismo

Los programadores de Airflow controlan la programación de las ejecuciones del DAG y las tareas individuales de los DAG. La opción de configuración [core]parallelism de Airflow controla cuántas tareas puede poner en cola el programador de Airflow en la cola del ejecutor después de que se cumplen todas las dependencias para estas tareas.

El paralelismo es un mecanismo de protección de Airflow que determina cuántas tareas se pueden ejecutar al mismo tiempo por cada programador, independientemente de la cantidad de trabajadores. El valor de paralelismo, multiplicado por la cantidad de programadores en tu clúster, es la cantidad máxima de instancias de tareas que tu entorno puede poner en cola.

Por lo general, [core]parallelism se establece como un producto de una cantidad máxima de trabajadores y [celery]worker_concurrency. También se ve afectado por el grupo. Puedes cambiar esta opción de configuración de Airflow si la anulas. Para obtener más información sobre cómo ajustar la configuración de Airflow relacionada con el escalamiento, consulta Cómo escalar la configuración de Airflow.

Encuentra las configuraciones de entorno óptimas

La forma recomendada de solucionar los problemas de programación es consolidar las tareas pequeñas en tareas más grandes y distribuirlas de manera más uniforme en el tiempo. Además de optimizar el código de DAG, también puedes optimizar las configuraciones de entorno para tener una capacidad suficiente para ejecutar varias tareas de forma simultánea.

Por ejemplo, supongamos que consolidas las tareas en tu DAG tanto como sea posible, pero limitar las tareas activas para distribuirlas de manera más uniforme en el tiempo no es una solución preferida para tu caso de uso específico.

Puedes ajustar el paralelismo, la cantidad de trabajadores y los parámetros de simultaneidad de los trabajadores para ejecutar el DAG de dag_10_tasks_20_seconds_10 sin limitar las tareas activas. En este ejemplo, el DAG se ejecuta 10 veces y cada ejecución contiene 20 tareas pequeñas. Si quieres ejecutarlos todos de forma simultánea, haz lo siguiente:

• Necesitarás un tamaño de entorno más grande, ya que controla los parámetros de rendimiento de la infraestructura administrada de Cloud Composer de tu entorno.

• Los trabajadores de Airflow deben poder ejecutar 20 tareas de forma simultánea, lo que significa que debes establecer la simultaneidad de los trabajadores en 20.

• Los trabajadores necesitan suficiente CPU y memoria para controlar todas las tareas. La simultaneidad del trabajador se ve afectada por la CPU y la memoria del trabajador, por lo que necesitarás al menos worker_concurrency / 12 en CPU y least worker_concurrency / 8 en memoria.

• Deberás aumentar el paralelismo para que coincida con la mayor simultaneidad de trabajadores. Para que los trabajadores tomen 20 tareas de la cola, el programador deberá programar primero esas 20 tareas.

Ajusta la configuración de tu entorno de la siguiente manera:

1. En la consola de Google Cloud, ve a la página Entornos.

   Ir a Entornos

2. En la lista de entornos, haz clic en el nombre de tu entorno. Se abrirá la página Detalles del entorno.

3. Ve a la pestaña Configuración del entorno.

4. Busca la configuración de Recursos > Cargas de trabajo y haz clic en Editar.

5. En la sección Trabajador, en el campo Memoria, especifica el nuevo límite de memoria para los trabajadores de Airflow. En este instructivo, usa 4 GB.

6. En el campo CPU, especifica el nuevo límite de CPU para los trabajadores de Airflow. En este instructivo, usa 2 vCPU.

7. Guarda los cambios y espera unos minutos para que los trabajadores de Airflow se reinicien.

A continuación, anula las opciones de configuración de Airflow de paralelismo y simultaneidad de trabajadores:

1. Ve a la pestaña Airflow Configuration Overrides.

2. Haz clic en Editar y, luego, en Agregar anulación de configuración de Airflow.

3. Anula la configuración de parralelismo:

   Sección	Clave	Valor
   core	parallelism	20

4. Haz clic en Agregar anulación de configuración de Airflow y anula la configuración de simultaneidad del trabajador:

   Sección	Clave	Valor
   celery	worker_concurrency	20

5. Haz clic en Guardar y espera a que el entorno actualice su configuración.

Vuelve a activar el mismo DAG de ejemplo con los parámetros de configuración ajustados:

1. En la IU de Airflow, ve a la página DAGs.

2. Busca el DAG de dag_10_tasks_20_seconds_10 y bórralo.

   Después de que se borre el DAG, Airflow verifica la carpeta de DAGs en el bucket de tu entorno y vuelve a ejecutar el DAG automáticamente.

Una vez que se completen las ejecuciones de DAG, vuelve a revisar el histograma de registros. En el diagrama, puedes ver que el ejemplo de dag_10_tasks_20_seconds_10 con tareas más consolidadas no generó errores ni advertencias cuando se ejecutó con la configuración del entorno ajustada. Compara los resultados con los datos anteriores del diagrama, en los que el mismo ejemplo generó errores y advertencias cuando se ejecutaba con la configuración de entorno predeterminada.

Figura 12: Histograma de registros del trabajador de Airflow después de ajustar la configuración del entorno (haz clic para ampliar)

Las configuraciones de entorno y de Airflow desempeñan un papel fundamental en la programación de tareas. Sin embargo, no es posible aumentar las configuraciones más allá de ciertos límites.

Te recomendamos que optimices el código de DAG, consolides las tareas y uses la programación para optimizar el rendimiento y la eficiencia.

Ejemplo: errores de análisis y latencia de DAG debido a un código de DAG complejo

En este ejemplo, investigarás la latencia de análisis de un DAG de muestra que imita un exceso de variables de Airflow.

Crea una nueva variable de Airflow

Antes de subir el código de muestra, crea una nueva variable de Airflow.

1. En la consola de Google Cloud, ve a la página Entornos.

   Ir a Entornos

2. En la columna Servidor web de Airflow, sigue el vínculo de Airflow para tu entorno.

3. Ve a Administrador > Variables > Agregar un registro nuevo.

4. Configura los siguientes valores:

   • clave: example_var
   • val: test_airflow_variable

Sube el DAG de muestra a tu entorno

Sube el siguiente DAG de muestra al entorno que creaste en los pasos anteriores. En este instructivo, este DAG se llama dag_for_loop_airflow_variable.

Este DAG contiene un bucle for que se ejecuta 1,000 veces y imita un exceso de variables de Airflow. Cada iteración lee la variable example_var y genera una tarea. Cada tarea contiene un comando que imprime el valor de la variable.

Advertencia: Este DAG simula problemas con la programación de DAG, lo que puede interrumpir las operaciones en tu entorno.

from datetime import datetime
from airflow import DAG
from airflow.operators.bash_operator import BashOperator
from airflow.models import Variable

default_args = {
    'owner': 'airflow',
    'depends_on_past': False,
    'start_date': datetime(2023, 2, 17),
    'retries': 0
}

dag = DAG(
    'dag_for_loop_airflow_variable',
    default_args=default_args,
    catchup=False,
    schedule_interval="@daily"
)

for i in range(1000):
    a = Variable.get('example_var', 'N/A')
    task = BashOperator(
        task_id=f'task_{i}',
        bash_command='echo variable foo=${foo_env}',
        dag=dag,
        env={'foo_env': a}
    )

Diagnostica los problemas de análisis

El tiempo de análisis del DAG es la cantidad de tiempo que tarda el programador de Airflow en leer un archivo DAG y analizarlo. Antes de que el programador de Airflow pueda programar cualquier tarea desde un DAG, el programador debe analizar el archivo del DAG para descubrir la estructura del DAG y las tareas definidas.

Si un DAG tarda mucho en analizarse, esto consume la capacidad del programador y podría reducir el rendimiento de las ejecuciones de DAG.

Para supervisar el tiempo de análisis del DAG, haz lo siguiente:

1. Ejecuta el comando de la CLI de Airflow dags report en la CLI de gcloud para ver el tiempo de análisis de todos tus DAG:

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

   Reemplaza lo siguiente:

   • ENVIRONMENT_NAME: Es el nombre de tu entorno.
   • LOCATION: Es la región en la que se encuentra el entorno.

2. En el resultado del comando, busca el valor de duración del DAG de dag_for_loop_airflow_variables. Un valor alto puede indicar que este DAG no se implementa de manera óptima. Si tienes varios DAG, en la tabla de resultados, puedes identificar qué DAG tienen un tiempo de análisis prolongado.

   Ejemplo:

   file                | duration       | dag_num | task_num | dags
   ====================+================+=========+==========+=====================
   /dag_for_loop_airfl | 0:00:14.773594 | 1       | 1000     | dag_for_loop_airflow
   ow_variable.py      |                |         |          | _variable
   /airflow_monitoring | 0:00:00.003035 | 1       | 1        | airflow_monitoring
   .py
   
   

3. Inspecciona los tiempos de análisis del DAG en la consola de Google Cloud:

   1. En la consola de Google Cloud, ve a la página Entornos.

   Ir a Entornos

4. En la lista de entornos, haz clic en el nombre de tu entorno. Se abrirá la página Detalles del entorno.

5. Ve a la pestaña Registros y, luego, a Todos los registros > Administrador de procesadores DAG.

6. Revisa los registros de dag-processor-manager y, luego, identifica los posibles problemas.

   

   Figura 13: Los registros del administrador del procesador de DAG muestran los tiempos de análisis del DAG (haz clic para ampliar).

Si el tiempo total de análisis de DAG supera los 10 segundos, es posible que tus programadores estén sobrecargados con el análisis de DAG y no puedan ejecutarlos de manera eficaz.

Optimiza el código del DAG

Se recomienda evitar el código de Python "de nivel superior" innecesario en tus DAG. Los DAG con muchas importaciones, variables y funciones fuera del DAG introducen tiempos de análisis más largos para el programador de Airflow. Esto reduce el rendimiento y la escalabilidad de Cloud Composer y Airflow. El exceso de lectura de variables de Airflow genera un tiempo de análisis largo y una carga alta de la base de datos. Si este código está en un archivo DAG, estas funciones se ejecutan en cada indicador de actividad del programador, lo que puede ser lento.

Los campos de plantillas de Airflow te permiten incorporar valores de las variables de Airflow y las plantillas de Jinja en tus DAG. Esto evita la ejecución de funciones innecesarias durante los mensajes de confirmación del programador.

Para implementar el ejemplo de DAG de una mejor manera, evita usar variables de Airflow en el código de Python de nivel superior de los DAG. En su lugar, pasa las variables de Airflow a los operadores existentes a través de una plantilla de Jinja, que retrasará la lectura del valor hasta la ejecución de la tarea.

Sube la versión nueva del DAG de ejemplo a tu entorno. En este instructivo, este DAG se llama dag_for_loop_airflow_variable_optimized.

Advertencia: Este DAG simula problemas con la programación de DAG, lo que puede interrumpir las operaciones en tu entorno.

from datetime import datetime
from airflow import DAG
from airflow.operators.bash_operator import BashOperator

default_args = {
    'owner': 'airflow',
    'depends_on_past': False,
    'start_date': datetime(2023, 2, 17),
    'retries': 0
}

dag = DAG(
    'dag_for_loop_airflow_variable_optimized',
    default_args=default_args,
    catchup=False,
    schedule_interval='@daily'
)

for i in range(1000):
    task = BashOperator(
        task_id=f'bash_use_variable_good_{i}',
        bash_command='echo variable foo=${foo_env}',
        dag=dag,
        env={'foo_env': '{{ var.value.get("example_var") }}'},
    )

Inspecciona el nuevo tiempo de análisis del DAG:

1. Espera a que se complete la ejecución del DAG.

2. Vuelve a ejecutar el comando dags report para ver el tiempo de análisis de todos tus DAG:

   file                | duration       | dag_num | task_num | dags
   ====================+================+=========+==========+=====================
   /dag_for_loop_airfl | 0:00:37.000369 | 1       | 1000     | dag_for_loop_airflow
   ow_variable.py      |                |         |          | _variable
   /dag_for_loop_airfl | 0:00:01.109457 | 1       | 1000     | dag_for_loop_airflow
   ow_variable_optimiz |                |         |          | _variable_optimized
   ed.py               |                |         |          |
   /airflow_monitoring | 0:00:00.040510 | 1       | 1        | airflow_monitoring
   .py                 |                |         |          |
   

3. Revisa los registros de dag-processor-manager nuevamente y analiza la duración del análisis.

   

   Figura 14: Los registros del administrador de procesadores del DAG muestran los tiempos de análisis del DAG después de que se optimizó el código del DAG (haz clic para ampliar).

Cuando reemplazaste las variables de entorno por plantillas de Airflow, simplificaste el código de DAG y redujiste la latencia de análisis unas diez veces.

Optimiza las configuraciones de entorno de Airflow

El programador de Airflow intenta activar tareas nuevas y analizar todos los DAGs de tu bucket de entorno de forma constante. Si tus DAG tienen un tiempo de análisis largo y el programador consume muchos recursos, puedes optimizar las configuraciones del programador de Airflow para que use los recursos de manera más eficiente.

En este instructivo, los archivos DAG tardan mucho tiempo en analizarse y los ciclos de análisis comienzan a superponerse, lo que agota la capacidad del programador. En nuestro ejemplo, el primer DAG de ejemplo tarda más de 5 segundos en analizarse, por lo que configurarás el programador para que se ejecute con menos frecuencia y usar los recursos de manera más eficiente. Anularás la opción de configuración de Airflow scheduler_heartbeat_sec. Esta configuración define la frecuencia con la que se debe ejecutar el programador (en segundos). De forma predeterminada, el valor se establece en 5 segundos. Puedes cambiar esta opción de configuración de Airflow si la anulas.

Anula la opción de configuración de Airflow scheduler_heartbeat_sec:

1. En la consola de Google Cloud, ve a la página Entornos.

   Ir a Entornos

2. En la lista de entornos, haz clic en el nombre de tu entorno. Se abrirá la página Detalles del entorno.

3. Ve a la pestaña Airflow Configuration Overrides.

4. Haz clic en Editar y, luego, en Agregar anulación de configuración de Airflow.

5. Anula la opción de configuración de Airflow:

   Sección	Clave	Valor
   scheduler	scheduler_heartbeat_sec	10

6. Haz clic en Guardar y espera a que el entorno actualice su configuración.

Verifica las métricas del programador:

1. Ve a la pestaña Monitoring y selecciona Programadores.

2. En el gráfico Scheduler heartbeat, haz clic en el botón Más opciones (tres puntos) y, luego, en Ver en el Explorador de métricas.

Figura 15: Gráfico de la señal de monitoreo de funcionamiento del programador (haz clic para ampliar)

En el gráfico, verás que el programador se ejecuta con la mitad de frecuencia después de que cambiaste la configuración predeterminada de 5 segundos a 10 segundos. Cuando reduces la frecuencia de los mensajes de estado, te aseguras de que el programador no comience a ejecutarse mientras el ciclo de análisis anterior está en curso y no se agota la capacidad de recursos del programador.

Asigna más recursos al programador

En Cloud Composer 2, puedes asignar más recursos de CPU y memoria al planificador. De esta manera, puedes aumentar el rendimiento de tu programador y acelerar el tiempo de análisis de tu DAG.

Asignar CPU y memoria adicionales al programador:

1. En la consola de Google Cloud, ve a la página Entornos.

   Ir a Entornos

2. En la lista de entornos, haz clic en el nombre de tu entorno. Se abrirá la página Detalles del entorno.

3. Ve a la pestaña Configuración del entorno.

4. Busca la configuración de Recursos > Cargas de trabajo y haz clic en Editar.

5. En la sección Programador, en el campo Memoria, especifica el nuevo límite de memoria. En este instructivo, usa 4 GB.

6. En el campo CPU, especifica el nuevo límite de CPU. En este instructivo, usa 2 vCPU.

7. Guarda los cambios y espera unos minutos para que se reinicien los programadores de Airflow.

8. Ve a la pestaña Registros y, luego, a Todos los registros > Administrador de procesadores DAG.

9. Revisa los registros de dag-processor-manager y compara la duración del análisis de los DAGs de ejemplo:

   

   Figura 16: Los registros del administrador de procesadores de DAG muestran los tiempos de análisis de DAG después de que se asignaron más recursos al programador (haz clic para ampliar).

Cuando asignaste más recursos al programador, aumentaste su capacidad y redujiste la latencia de análisis de forma significativa en comparación con las configuraciones predeterminadas del entorno. Con más recursos, el programador puede analizar los DAG más rápido. Sin embargo, los costos asociados con los recursos de Cloud Composer también aumentarán. Además, no es posible aumentar los recursos más allá de un límite determinado.

Recomendamos asignar recursos solo después de implementar las posibles optimizaciones de código DAG y configuración de Airflow.

Limpia

Para evitar que se apliquen cargos a tu Google Cloud cuenta por los recursos usados en este instructivo, borra el proyecto que contiene los recursos o conserva el proyecto y borra los recursos individuales.

Borra el proyecto

Precaución: Borrar un proyecto tiene las siguientes consecuencias:

• Se borra todo en el proyecto. Si usaste un proyecto existente para las tareas de este documento, cuando lo borres, también se borrará cualquier otro trabajo que hayas realizado en el proyecto.
• Se pierden los ID personalizados de proyectos. Cuando creaste este proyecto, es posible que hayas creado un ID del proyecto personalizado que desees utilizar en el futuro. Para conservar las URL que utilizan el ID del proyecto, como una URL appspot.com, borra los recursos seleccionados dentro del proyecto en lugar de borrar todo el proyecto.

Si planeas explorar varias infraestructuras, 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.

1. In the Google Cloud console, go to the Manage resources page.

   Go to Manage resources

2. In the project list, select the project that you want to delete, and then click Delete.
3. 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ás el bucket del entorno durante este procedimiento.

¿Qué sigue?

• Optimiza el rendimiento y los costos de tu entorno
• Escalar entornos
• Soluciona problemas de los DAG