Scheduler

Clase 7 de 29 • Curso de Fundamentos de Apache Airflow

Mauro Ezequiel Bravo

student•

DAG => Flujo de datos TASK => Componentes dentro del DAG que se puede anidar con otros TASKs. Operator => Objeto con el que se crea la TASK. Scheduler => Encargado de ejecutar los DAGs. Cuando ejecuta, en que intevalo de tiempo, etc.

Royer Guerrero Pinilla

student•

✨ Existen 2 maneras de definir esto

Usando expresiones CRON

# ┌─────── minute (0 - 59)
# │ ┌────── hour (0 - 23)
# │ │ ┌───── day of the month (1 - 31)
# │ │ │ ┌───── month (1 - 12)
# │ │ │ │ ┌──── day of the week (0 - 6) (Sunday to Saturday;
# │ │ │ │ │      7 is also Sunday on some systems)
# * * * * *

Usando las opciones pre-configuradas @once Una y solo una vez @hourly Cada hora al comienzo de la hora @dialy Diariamente a medianoche @weekly Una vez a la semana el domingo @monthly Cada primer dia del mes a medianoche @yearly Cada 1 de Enero a medianoche

⚠️ Hay que tener cuidado porque el la vista de grid se organiza por fecha de programación schedule date y no por la fecha de ejecución execution_date lo que puede confundir y se puede llegar a pensar que no corrió

Sergio Waldemar Garzón Mariño

student•

Vengo de trabajar mucho con Control-M, es muy parecido en cuando al core de la aplicación. Airflow cuenta con la ventaja de programar en Python, en cambio, para programar los jobs en Control-M usamos XML o la interfaz gráfica.

Mario Alexander Vargas Celis

student•

El Scheduler en Apache Airflow es el componente central encargado de gestionar la ejecución de los DAGs (Directed Acyclic Graphs) y las tareas definidas en ellos. Es responsable de planificar, desencadenar y supervisar las tareas según las dependencias y los intervalos de tiempo especificados.

Funciones principales del Scheduler

Planificación de DAGs:
- Detecta automáticamente los DAGs disponibles y calcula los intervalos de ejecución para ellos.
- Identifica las tareas que están listas para ejecutarse según sus dependencias y programación.
Asignación de tareas a los workers:
- Determina qué tareas deben ejecutarse y las asigna a los workers para su ejecución.
Supervisión del estado de las tareas:
- Supervisa continuamente el estado de las tareas: queued, running, success, failed, etc.
- Reintenta tareas en caso de fallos si está configurado.
Ejecuta tareas programadas o manuales:
- Procesa ejecuciones automáticas basadas en el parámetro schedule_interval del DAG.
- Maneja ejecuciones manuales iniciadas por usuarios desde la interfaz de usuario o la línea de comandos.

Flujo de trabajo del Scheduler

Carga de DAGs:
- El Scheduler analiza los DAGs definidos en los archivos de Python dentro del directorio especificado (dags_folder).
- Verifica si hay nuevas ejecuciones pendientes basadas en las definiciones de los DAGs.
Determinación de las tareas ejecutables:
- Evalúa las dependencias entre tareas para determinar cuáles están listas para ejecutarse.
- Considera las configuraciones como:
  - start_date: Fecha desde la cual debe comenzar a ejecutarse el DAG.
  - schedule_interval: Frecuencia de ejecución del DAG.
  - catchup: Si debe ejecutarse para intervalos pasados o solo para el más reciente.
Cola de tareas:
- Las tareas listas se colocan en una cola para ser recogidas por los workers disponibles.
Monitoreo continuo:
- El Scheduler sigue monitoreando los DAGs y tareas para desencadenar nuevas ejecuciones y manejar fallos.

Configuración del Scheduler

El Scheduler puede configurarse desde el archivo airflow.cfg bajo la sección [scheduler]. Algunas opciones clave incluyen:

scheduler_heartbeat_sec:
- Intervalo en segundos en que el Scheduler verifica el estado de los DAGs.
min_file_process_interval:
- Tiempo mínimo entre análisis de los archivos del directorio de DAGs.
num_runs:
- Número máximo de ejecuciones antes de reiniciar el proceso del Scheduler.
max_threads:
- Número de subprocesos que puede utilizar el Scheduler para procesar tareas simultáneamente.
dag_dir_list_interval:
- Intervalo para buscar cambios en el directorio de DAGs.

Inicio del Scheduler

El Scheduler se ejecuta como un servicio continuo que procesa los DAGs y las tareas. Para iniciar el Scheduler, puedes usar el siguiente comando en la línea de comandos:

airflow scheduler

Este comando:

Comienza a analizar los DAGs en el directorio especificado.
Gestiona las tareas programadas en función de sus dependencias y disponibilidad de recursos.

Ejemplo práctico

Supongamos que tienes un DAG que ejecuta una tarea diaria para procesar datos. El Scheduler:

Verifica si es necesario ejecutar el DAG en función de su schedule_interval.
Evalúa las dependencias entre tareas en el DAG.
Coloca las tareas listas en la cola.
Asigna las tareas a los workers disponibles para ejecutarlas.

Código de ejemplo:

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

dag = DAG( dag_id='daily_processing', schedule_interval='@daily', start_date=datetime(2023, 1, 1), catchup=False, )

task = BashOperator( task_id='process_data', bash_command='echo "Procesando datos..."', dag=dag, )

En este caso:

El Scheduler verificará diariamente si la tarea process_data debe ejecutarse.
Si está lista, la colocará en la cola para que un worker la procese.

Interacción con otros componentes

Workers:
- El Scheduler delega la ejecución de tareas a los workers disponibles.
Base de datos metadata:
- El Scheduler consulta y actualiza continuamente el estado de los DAGs y tareas en la base de datos.
Webserver:
- Permite a los usuarios monitorear y controlar las ejecuciones desde la interfaz web, interactuando indirectamente con el Scheduler.

Buenas prácticas para el Scheduler

Optimizar la cantidad de DAGs:
- Evita tener demasiados DAGs pequeños o con tareas innecesarias que sobrecarguen al Scheduler.
Usar dependencias correctamente:
- Configura las dependencias de las tareas para evitar bloqueos innecesarios.
Configurar el número adecuado de workers:
- Asegúrate de que los workers disponibles puedan procesar las tareas programadas eficientemente.
Monitorear el rendimiento del Scheduler:
- Usa herramientas de monitoreo para verificar si el Scheduler está procesando los DAGs a tiempo y eficientemente.

Con estas consideraciones, el Scheduler puede manejar flujos de trabajo complejos y escalar según las necesidades del proyecto.

Camilo Corredor

student•

¿El operador viene siendo un worker dentro del work flow?

Santiago Ahumada Lozano

student•

Hola Camilo

Lo que entiendo es que no. El worker es el componente que ejecuta las tareas. Mientras que el operador lo que hace es definir qué tipo de tareas existen.

Si ves la arquitectura que el profe muestra podras notar que los workers estan directamente conectados desde el servidor web. Sin embargo es el executor quien corre las tareas. En otras palabras, el scheduler le dice al executor qué tareas se deben ejecutar y cuando y este simplemente se encarga de usar los workers para que las tareas que se ponen a correr se ejecuten.

te recomiendo estas lecturas ya que me han ayudado a entender los componentes

Espero te sea de ayuda! Saludos:)

Scheduler

Introducción a Apache Airflow

Introducción al curso

¿Para qué sirve airflow?

¿Por qué usar airflow?

Resumen módulo 1

Conceptos básicos

DAG

Tasks y Operators

Scheduler

Instalación y configuración

Instalación de Apache Airflow usando Docker

Posibles configuraciones

Variables y conexiones

Implementando un DAG

Implementando un DAG

Bash Operator

Python Operator

Definiendo dependencias entre tareas

Custom Operator

Orquestar y monitorizar procesos

Orquestando un DAG I

Orquestando un DAG II

Monitoring

Task Actions

Trigger Rules

Sensores

¿Qué son los sensores?

ExternalTaskSensor

FileSensor

Templates con Jinja

¿Qué son los templates con Jinja?

XComs

¿Qué son los Xcoms?

BranchPythonOperator

BranchPythonOperator

Proyecto

Definición del proyecto

Cierre del curso

Cierre del curso

Comparte tu proyecto de Platzi explora el espacio con Airflow y certifícate