Tasks y Operators

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:002:13

Tasks y Operators

Curso de Fundamentos de Apache Airflow

Clase 6 de 29

Clase 6 / 29

Comentarios (6)6 comentarios

Royer Guerrero Pinilla

Hace 4 años

🏗️ Architecture

🧰 Tech Stack

🌿 Celery - Airflow Workers
🍄 Redis - Queue
💽 SQLite - Airflow Metastore (Really se puede cambiar por cualquier base de datos relacional)

Eric Bellet

Profesor

Hace 4 años

Gracias por compartirlo!

Víctor Mazo

Hace 3 años

En resumen, una task es la acción o el paso que deseamos realizar, mientras que el operator dictamina la forma de realizar dicha acción, así por ejemplo: Deseo mostrar un 'hola mundo' (task), y la forma en que lo podría hacer sería mediante un Bash Operator (valga la redundancia, el operator).

Lucas Gonzalez

Hace 4 años

Un DAG está compuesto por tareas:

Operators.
Sensors.
TaskFlow-decorated@task. Entre los operators podemos utilizar( entre otros) BashOperators o PythonOperators, algunas veces se pueden utilizar ambos

Mario Alexander Vargas Celis

Hace 2 años

En Apache Airflow, tasks y operators son conceptos clave para construir y ejecutar flujos de trabajo.

Tasks (Tareas)

Una task es una unidad individual de trabajo dentro de un DAG (Directed Acyclic Graph). Cada tarea representa una operación específica que se ejecuta como parte del flujo de trabajo. Las tareas son instancias de operadores, y juntas conforman las actividades que ocurren en un DAG.

Características de una Task

Independencia: Cada tarea es independiente y realiza una acción específica.
Dependencias: Las tareas pueden depender unas de otras para garantizar que se ejecuten en el orden correcto.
Configuración de reintentos:
- Puedes configurar cuántas veces intentará ejecutarse una tarea si falla.
- Ejemplo: retries=3
Estado de ejecución:
- Los estados posibles incluyen: success, failed, running, skipped.

Operators (Operadores)

Un operator es una plantilla predefinida en Airflow que define lo que hace una tarea. Los operadores proporcionan la lógica para ejecutar una acción específica, como ejecutar un script de Python, interactuar con una API o copiar datos entre bases de datos.

Tipos de Operadores

Operadores de acción:
- Ejecutan una acción concreta, como un script o comando.
- Ejemplos:
  - BashOperator: Ejecuta comandos Bash.
  - PythonOperator: Ejecuta funciones de Python.
Operadores de transferencia:
- Manejan transferencias de datos entre sistemas.
- Ejemplos:
  - S3ToGCSOperator: Copia datos de S3 a Google Cloud Storage.
  - MySqlToPostgresOperator: Transfiere datos entre bases de datos.
Operadores de sensores:
- Esperan a que ocurra un evento antes de continuar.
- Ejemplos:
  - S3KeySensor: Espera a que un archivo específico esté disponible en S3.
  - HttpSensor: Verifica que una URL esté activa.
Operadores personalizados:
- Puedes crear operadores personalizados mediante herencia de clases base como BaseOperator.

Ejemplo de Operators en uso

BashOperator

Ejecuta comandos en el sistema operativo:

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

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

bash_task = BashOperator( task_id='print_date', bash_command='date', dag=dag, )

PythonOperator

Ejecuta funciones en Python:

from airflow import DAG from airflow.operators.python import PythonOperator from datetime import datetime

def my_python_function(): print("Hello from Python!")

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

python_task = PythonOperator( task_id='run_python', python_callable=my_python_function, dag=dag, )

Sensor

Espera a que un archivo exista en un sistema S3:

from airflow import DAG from airflow.providers.amazon.aws.sensors.s3_key import S3KeySensor from datetime import datetime

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

s3_sensor = S3KeySensor( task_id='wait_for_file', bucket_name='my-bucket', bucket_key='path/to/file.csv', aws_conn_id='my_aws_connection', dag=dag, )

Relación entre Tasks y Operators

Una task es una instancia de un operator.
Un operator define la lógica de ejecución (qué hace).
Una task utiliza esa lógica y la aplica en un DAG específico (cómo y cuándo lo hace).

Por ejemplo:

bash_task = BashOperator( task_id='show_date', bash_command='date', dag=dag, )

En este caso:

BashOperator define cómo ejecutar un comando Bash.
bash_task es la tarea específica que ejecuta el comando date.

Configuración Avanzada de Tasks

Dependencias:task1 >> task2 # task2 se ejecuta después de task1 task3 << task1 # task1 se ejecuta antes de task3
- Se configuran para controlar el orden de ejecución.
Propiedades comunes:
- retry_delay: Tiempo entre reintentos.
- timeout: Límite de tiempo para completar la tarea.
- execution_timeout: Tiempo máximo permitido para que la tarea se ejecute.
Ejemplo con varias dependencias:start = DummyOperator(task_id='start', dag=dag) process = PythonOperator(task_id='process', python_callable=my_function, dag=dag) end = DummyOperator(task_id='end', dag=dag)

start >> process >> end

Resumen

Las tasks son las actividades que ejecuta un DAG.
Los operators son plantillas predefinidas que implementan la lógica de ejecución de las tareas.
Juntos, permiten crear flujos de trabajo dinámicos, reutilizables y escalables en Airflow.

Silvia Veronica Noriega

Hace 4 años

Èse codigo python me quedo dando vueltas en la cabeza.. nunca lo habia visto asi, hasta ahora... que es eso de callable?

Jhony Urbano Diaz Quispe

Hace 4 años

en python puedes pasar una funcion como argumento y es lo que estan haciendo

Alfonso Andres Zapata Guzman

Hace 3 años

En python la funciones son pasajeros/ciudadanos de primera clase.

Por el contrario es muy normal pasar funciones a otras funciones, ya sea funciones que creamos nosotros y se la pasamos a funciones de librerias que exportamos. Como funciones exportadas de una libreria a otra funcion exportada de otra libreria distinta. En los cursos de python aca en platzi se trabaja un modulo creando funciones y pasandolas a otras funciones que creamos, viendo alcance del scope, etc, y en cursos de la ruta de ciencia de datos se ve varias veces como pasamos funciones de sklearn a otras funciones de sklearn, o una que es muy representativa, pasar de sklearn a imblearn, no hay que saltarse los cursos de la ruta, al comienzo del curso se indica que se requiere python.

~ Por cierto, ya conectamos en LinkedIn? ~

Conectemos en LinkedIn, GitHub, Medium o Redes sociales

Camilo Corredor

Hace 2 años

El decorador @task en Apache Airflow se utiliza para marcar funciones como tareas dentro de un DAG (Directed Acyclic Graph). Esto permite que Airflow las trate como tareas independientes en el flujo de trabajo. Al usar @task, se simplifica la creación de tareas, ya que no es necesario definir explícitamente un operador. Además, proporciona una forma más clara y concisa de gestionar la lógica de las tareas en el flujo de trabajo. Es parte de la funcionalidad de "TaskFlow", que facilita la orquestación de procesos en Airflow.

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

Tasks y Operators