Orquestando un DAG II

Solo quiero decir GRACIAS! por el excelente uso del zoom en este curso, enfocando en lo que hay que enfocarse. Además de que en otros cursos es imposible verlo en el celular porque terminas viendo una pantalla negra con unas letritas moviendose, aquí sin ningun problema.

La cosa se pone un poco más confusa si no tienes claro en que franja horaria esta siendo planificada jaja, porque tienes tu hora local y la hora del servidor donde tengas Aiflow si corre en la nube, lo más recomendable es hacerlo en hora UTC.

Con respecto a los intervalos, efectivamente me tomo un tiempo entenderlo, por ejempo en mi trabajo los dag procesan los datos a dia cumplido, es decir, si mi próxima ejecucion esta programada para el 24/10/22 a las 00:00:00, en la realidad esto va a ejecutar cuando termine ese día, es decir el 25/10/22 00:00:00, justamente para ingestar los datos con todo el comportamiento que tuvo durante el día 24

)

Libro: Ruiter-Data Pipelines with Apache Airflow

Primer ejemplo

from airflow import DAG
from airflow.operators.empty import EmptyOperator
from datetime import datetime

with DAG(dag_id="5.2-orquestation",
         description="Probando la orquestacion",
         schedule_interval="0 7 * * 1",
         start_date=datetime(2022, 1, 1),
         end_date=datetime(2022, 6, 1)) as dag:

    t1 = EmptyOperator(task_id = "tarea1")

    t2 = EmptyOperator(task_id = "tarea2")

    t3 = EmptyOperator(task_id = "tarea3")

    t4 = EmptyOperator(task_id = "tarea4")

    t1 >> t2 >> t3 >> t4

Segundo ejemplo

from airflow import DAG
from airflow.operators.empty import EmptyOperator
from datetime import datetime

with DAG(dag_id="5.3-orquestation",
         description="Probando la orquestacion",
         schedule_interval="@monthly",
         start_date=datetime(2022, 1, 1),
         end_date=datetime(2022, 6, 1)) as dag:

    t1 = EmptyOperator(task_id = "tarea1")

    t2 = EmptyOperator(task_id = "tarea2")

    t3 = EmptyOperator(task_id = "tarea3")

    t4 = EmptyOperator(task_id = "tarea4")

    t1 >> t2 >> t3 >> t4

Mi practica,

Le puse algunas pausas a los procesos,

y este es el código

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



with DAG(dag_id = "5.4-orquestation-tarea",
         description          = "Probando la orquestacion 3",
         schedule_interval    = "0 0 * * *", #"se ejecuta cada dia a las 00:00 horas
         start_date           = datetime(2022, 5, 1),
         end_date             = datetime(2023, 5, 30),
         default_args         = {"depends_on_past": True},
         #max_active_runs      = 1 
         ) as dag:
    # "depends_on_past" por defecto es False, por lo que pueden surgir los procesos todos los días independientemente. 
    # "depends_on_past" hace que se ejecute solo si el anterior proceso terminó, así se realiza uno a la vez secuencial
    # max_active_runs, se ejecute un modulo, evitado que haya paralelismo, y cuando se termine todo le grupo, se pasa al sgte día.
 
 
    t1 = EmptyOperator(task_id="tarea1")
    t2 = BashOperator(task_id="tarea2", bash_command="sleep 2 && echo 'tarea2'")
    t3 = EmptyOperator(task_id="tarea3")
    t4 = EmptyOperator(task_id="tarea4")
    t5 = BashOperator(task_id="tarea5", bash_command="sleep 2 && echo 'tarea5'")
    t6 = EmptyOperator(task_id="tarea6")
    t7 = BashOperator(task_id="tarea7", bash_command="sleep 3 && echo 'tarea7'")
    t8 = BashOperator(task_id="tarea8", bash_command="sleep 2 && echo 'tarea8'")
    
    t1 >> [ t2, t3, t4] 
    t3 >> [t5, t6]
    [t4, t5] >> t7
    [t4, t7] >> t8 

https://crontab.guru/