ExternalTaskSensor

Clase 22 de 29 • Curso de Fundamentos de Apache Airflow

Mauro Ezequiel Bravo

student•

Buenas!

7.1-ExternalTaskSensor .py

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


with DAG(dag_id="7.1-externalTaskSensor",
    description="DAG principal",
    schedule_interval="@daily",
    start_date=datetime(2022, 8, 10),
    end_date=datetime(2022, 8, 25)
) as dag:

    t1 = BashOperator(task_id="tarea_1",
                      bash_command="sleep 10 && echo 'DAG finalizado!'",
                      depends_on_past=True)

    t1

7.2-ExternalTaskSensor .py

from datetime import datetime
from airflow import DAG
from airflow.operators.bash import BashOperator
from airflow.sensors.external_task import ExternalTaskSensor


with DAG(dag_id="7.2-externalTaskSensor",
    description="DAG Secundario",
    schedule_interval="@daily",
    start_date=datetime(2022, 8, 20),
    end_date=datetime(2022, 8, 25),
    max_active_runs=1
) as dag:

    t1 = ExternalTaskSensor(task_id="waiting_dag",
							external_dag_id="7.1-externalTaskSensor",
							external_task_id="tarea_1",
							poke_interval=10 # Cada 10 segundos pregunta si ya termino la tarea
							)

    t2 = BashOperator(task_id="tarea_2",
					  bash_command="sleep 10 && echo 'DAG 2 finalizado!'",
					  depends_on_past=True)

    t1 >> t2

Saludos

Federico Martinez

student•

¿Cómo orquestar DAGs con ExternalTaskSensor?

El ExternalTaskSensor te permite establecer dependencias entre tareas que residen en DAGs diferentes. Esto es crucial cuando un DAG necesita esperar la finalización exitosa de una tarea específica en otro DAG antes de comenzar sus propias operaciones, asegurando una orquestación coordinada.

Santiago Ahumada Lozano

student•

La utilidad de los task sensors incluye dividir un worflow complejo en DAGs mas modulares, de modo que un DAG se ejecute cuando su sensor detercte que otro se ejecuta

Manuel CABRERA

student•

Buen dia, al momento de realizar este ejercicio mi scheduler se pausa automaticamente, agradezco su ayuda inmensamente para lograr solucionar este problema.

The scheduler does not appear to be running. Last heartbeat was received 3 minutes ago.

The DAGs list may not update, and new tasks will not be scheduled.

Eric Bellet

teacher•

Buenas Andres, por lo que veo tienes un error levantando los servicios. Con Docker esto puedo pasar si tienes poca memoria, la solución suele ser limpiar los contenedores, imágenes, y ampliar los recursos de Docker. Lo qie te esta pasando es que el contenedor probablemente no tenga suficiente memoria

Jorge Aurelio Valdez Osorio

student•

Si estamos esperando por el main dag, pero este dag se ejecuta varias veces, una cada dia, como sabe airflow que dia esperar?

Eric Bellet

teacher•

Airflow lo maneja a nivel de ejecuciones. Es decir, el día Lunes no se ejecuta hasta que el Lunes del main DAG no se ejecuta

Pablo Ponte

student•

el sensor como sabe contra que instancia de ejecución disparar? con la primera que se ejecute? con la del mismo día?

Eric Bellet

teacher•

Con la del mismo instante, es decir, día y hora

Manuel Felipe Jiménez Cortés

student•

Esta clase no se puede ver: "The media could not be loaded, either because the server or network failed or because the format is not supported"

Sura Naileth Cedeño Lopez

Team Platzi•

Ya está listo

Norberto Iván Tolaba

student•

Hoy 11-6-24 sí puedo ver está clase

Emilio Nicolás Mendoza Patti

student•

Al fin me salio

Edwin Uldarico Hernandez Osorio

student•

Esta Clase no funciona, @platziteam

Maximiliano Negrillo Salcedo

student•

Buenos días, no es posible ver la clase del ExternalTaskSensor.

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


with DAG(dag_id="7.1-externalTaskSensor",
    description="DAG principal",
    schedule_interval="@daily",
    start_date=datetime(2022, 8, 10),
    end_date=datetime(2022, 8, 25)
) as dag:

    t1 = BashOperator(task_id="tarea_1",
                      bash_command="sleep 10 && echo 'DAG finalizado!'",
                      depends_on_past=True)

    t1 

from datetime import datetime
from airflow import DAG
from airflow.operators.bash import BashOperator
from airflow.sensors.external_task import ExternalTaskSensor


with DAG(dag_id="7.2-externalTaskSensor",
    description="DAG Secundario",
    schedule_interval="@daily",
    start_date=datetime(2022, 8, 20),
    end_date=datetime(2022, 8, 25),
    max_active_runs=1
) as dag:

    t1 = ExternalTaskSensor(task_id="waiting_dag",
							external_dag_id="7.1-externalTaskSensor",
							external_task_id="tarea_1",
							poke_interval=10 # Cada 10 segundos pregunta si ya termino la tarea
							)

    t2 = BashOperator(task_id="tarea_2",
					  bash_command="sleep 10 && echo 'DAG 2 finalizado!'",
					  depends_on_past=True)

    t1 >> t2

ExternalTaskSensor

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