Orquestando un DAG I

Contenido del curso

Introducción a Apache Airflow

Conceptos básicos

Instalación y configuración

Implementando un DAG

Orquestar y monitorizar procesos

Sensores

0:0010:31

Orquestando un DAG I

Curso de Fundamentos de Apache Airflow

Clase 16 de 29

Clase 16 / 29

Comentarios (12)12 comentarios

Héctor Arturo Can Bacab

Hace 4 años

Para aquellas personas a las que solo se ejecuto una vez y no todo el mes como en la clase. Esto es debido a que posiblemente pusieron la fecha de inicio al día actual, ejecutaron y luego lo cambiaron a una fecha más atrás para hacer la simulación de tareas secuenciales por día.

La única forma en la que se me resolvío fue al darle al botón de eliminar, no estoy completamente seguro, pero parece ser que si ya ejecutaste un día, Airflow no va a ejecutar días anteriores, aunque el start_date y end_date esten correctos, sino hasta que se haya limpiado el registro de ese DAG.

Marvin Alvarenga

Hace 4 años

Airflow por defecto usa UTC como zona horaria, a menos que se le cambie directamente en el archivo de configuración la zona horaria por defecto a utilizar.

Una forma de manejar zonas horarias es con Python de la siguiente forma:

from datetime import datetime, timedelta, timezone

TIMEZONE = timezone(timedelta(hours=-6))  # UTC-6

with DAG(dag_id='orquestation',
         start_date=datetime(2023, 1, 1, tzinfo=TIMEZONE),
         end_date=datetime(2023, 2, 1, tzinfo=TIMEZONE),
         schedule_interval='@daily') as dag:

	...

También podría utilizarse la libreria pytz para hacerlo más intuitivo pero puede ser logrado sin instalar nada.

Stanley Melgar

Hace 4 años

Crontab es una herramienta muy útil de Linux, permite automatizar tareas en tu PC fácilmente. Por ejemplo, hacer copias de seguridad, actualizar repositorios mediante un pull etc. La imaginación es el límite. En este repositorio está un script de Python para ordenar los archivos de un directorio en concreto (por ejemplo Downloads, donde siempre va a parar todo si no eres ordenado), genera logs de los archivos que se movieron y hacia donde con fecha y hora. Todo con base en la extensión de los archivos ~

Oscar Camilo Luna Feo

Hace 4 años

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

with DAG(dag_id="5.1-orquestacion",
         description="Probando la orquestacion",
         schedule_interval="@daily",
         start_date=datetime(2022, 5, 1),
         end_date=datetime(2023, 6, 1),
         default_args={"depends_on_past": True},
         max_active_runs=1) as dag:

    t1 = BashOperator(task_id="tarea1",
                      bash_command="sleep 2 && echo 'Tarea 1'")

    t2 = BashOperator(task_id="tarea2",
                      bash_command="sleep 2 && echo 'Tarea 2'")

    t3 = BashOperator(task_id="tarea3",
                      bash_command="sleep 2 && echo 'Tarea 3'")

    t4 = BashOperator(task_id="tarea4",
                      bash_command="sleep 2 && echo 'Tarea 4'")

    t1 >> t2 >> [t3,t4]

Santiago Zuluaga Saldarriaga

Hace 3 años

Por qué a el le funciona en la clase sin el uso del default_args={"depends_on_past": True}? En mi caso sí no se especifica ese argumento falla.

Norberto Iván Tolaba

Hace 2 años

Notar que este código está para un año.

Gerardo Mayel Fernández Alamilla

Hace 4 años

No entendi por qué se simuló todo el mes? si lo parámetros dicen un mes pero no transcurrió

Eric Bellet

Profesor

Hace 4 años

¿En qué minuto del video? Según entiendo la pregunta, Airflow ejecuta cuando el final del intervalo ha llegado. Por ejemplo en un intervalo mensual, el mes de Febrero se va a ejecutar cuando ya sea Marzo

Royer Guerrero Pinilla

Hace 4 años

Eso pasa porque Airflow detecta si la fecha de inicio es mayor que a la actual entonces el ejecuta todas las tareas que no fueron ejecutadas si te das cuenta el lo esta ejecutando el día 02-08-2022 y el start_date del dag esta dede una fecha menor el 01-06-2022, si deseas evitar ese comportamiento debes usar el parámetro catchup al momento de inicializar el DAG

sebastian felipe herrera sanchez

Hace 2 años

El argumento depends_on_past=True se aplica a cada tarea individualmente y no al DAG completo. Esto significa que cada tarea esperará a que su ejecución anterior se complete antes de comenzar, pero no necesariamente esperará a que todas las tareas del DAG se completen.

Ejemplo: tarea1 del día 2 no comenzará hasta que tarea1 del día 1 se haya completado con éxito. tarea2 del día 2 no comenzará hasta que tarea1 del día 2 se haya completado con éxito.

sebastian felipe herrera sanchez

Hace 2 años

Tengo una pregunta, ¿Como tengo que organizar mis archivos a la hora de empezar un proyecto en airflow?

Eric Bellet

Profesor

Hace 2 años

Realmente depende del proyecto o empresa en la que estés trabajando. En mi empresa organizamos los dags por carpetas relacionadas a cada proyecto, y el nombre de cada DAG sigue una sintaxis interna

José Daniel López Ramírez

Hace un año

@sebastian, Esto tal vez te ayude:

Marvin Alvarenga

Hace 4 años

En ese caso que habían ejecuciones pendientes, cual sería un ejemplo real o más concreto del por qué no quisieramos que se ejecutaran en paralelo y de golpe las pendientes? Gracias.

Eric Bellet

Profesor

Hace 4 años

Normalemente si tienes un proceso que depende del resultado del anterior. Por ejemplo, de los datos que puede generar

Carlos Plaza

Hace 4 años

Por que se ejecutan varias tareas si el parametro schedule_interval esta dado como "@daily". No deberia ser una sola ejecucion? O es tal vez porque los parametros de fecha estan dados como start_date=1/may/2022 y end_date =1/jun/2022, pero la ejecucion del ejercicio se lo esta realizando el 22/ago/2022. Es decir airflow estaria como ejecutando un conjunto de tareas de dias acumulados, y es por eso que parecia que todo se realizada el mismo dia?

Eric Bellet

Profesor

Hace 4 años

Exacto, por defecto ejecuta todos los días acumulados al mismo tiempo. Si no quieres eso, puedes utilizar el max_active_runs y el depends_on_past

Mario Alexander Vargas Celis

Hace 2 años

Un DAG (Directed Acyclic Graph) en el contexto de la orquestación de tareas (como Apache Airflow) es una estructura que define la secuencia y las dependencias entre las tareas que se ejecutan como parte de un flujo de trabajo.

Carlos Perilla

Hace 3 años

Para limpiar también es útil seleccionar en la sección Task Actions las opciones Past y Future, luego clickear en Clear.

Con esto borramos las tareas pasadas y futuras. Recomiendo primero detener la ejecución del DAG.

Julián Esteban Oliveros Forero

Hace 4 años

Usualmente Airflow está en UTC entonces cuando se crea un crontab y quieren que se ejecute en su hora local deben hacer el cambio horario por ejemplo:

Si se quiere ejecutar una tarea a las 7 de Colombia en el schedule interval se deberá poner que se ejecute a las 11

from datetime import datetime, timedelta, timezone

TIMEZONE = timezone(timedelta(hours=-6))  # UTC-6

with DAG(dag_id='orquestation',
         start_date=datetime(2023, 1, 1, tzinfo=TIMEZONE),
         end_date=datetime(2023, 2, 1, tzinfo=TIMEZONE),
         schedule_interval='@daily') as dag:

	...

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

with DAG(dag_id="5.1-orquestacion",
         description="Probando la orquestacion",
         schedule_interval="@daily",
         start_date=datetime(2022, 5, 1),
         end_date=datetime(2023, 6, 1),
         default_args={"depends_on_past": True},
         max_active_runs=1) as dag:

    t1 = BashOperator(task_id="tarea1",
                      bash_command="sleep 2 && echo 'Tarea 1'")

    t2 = BashOperator(task_id="tarea2",
                      bash_command="sleep 2 && echo 'Tarea 2'")

    t3 = BashOperator(task_id="tarea3",
                      bash_command="sleep 2 && echo 'Tarea 3'")

    t4 = BashOperator(task_id="tarea4",
                      bash_command="sleep 2 && echo 'Tarea 4'")

    t1 >> t2 >> [t3,t4]

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

Orquestando un DAG I