En este capítulo, tendrás una introducción completa a los componentes de Apache Airflow y aprenderás cómo y por qué deberías usarlos.

Introducción a Apache Airflow

Probar una tarea en Airflow

Examinar los comandos de Airflow

DAGs de Airflow

Definir un DAG sencillo

Trabajar con DAGs y la shell de Airflow

Solucionar problemas al crear un DAG

Interfaz web de Airflow

Iniciar el servidor web de Airflow

Cómo moverte por la interfaz de Airflow

Examinar DAGs con la interfaz de Airflow

Introducción a Airflow

¿Qué tal, DAG? Ahora toca aprender lo básico para implementar DAGs de Airflow. Con actividades prácticas, aprenderás a configurar y desplegar operadores, tareas y programación.

Operadores de Airflow

Definir una tarea con BashOperator

Varios BashOperators

Tareas de Airflow

Define el orden de los BashOperators

Determinar el orden de las tareas

Solución de problemas con dependencias de DAG

Operadores adicionales

Usar el PythonOperator

Más PythonOperators

EmailOperator y dependencias

Programación en Airflow

Programa un DAG con Python

Descifrando las planificaciones de Airflow

Solución de problemas en ejecuciones de DAG

Implementación de DAGs de Airflow

En este capítulo, aprenderás a ahorrarte tiempo con componentes de Airflow como sensores y ejecutores, a la vez que monitorizas y resuelves problemas en flujos de trabajo de Airflow.

Sensores de Airflow

Options

Sensors

Both

Operators

Sensores vs operadores

Privación sensorial

Ejecutores de Airflow

Determinar el ejecutor

from airflow import DAG
from airflow.operators.bash import BashOperator
from airflow.sensors.filesystem import FileSensor
from datetime import datetime

report_dag = DAG(
    dag_id = 'execute_report',
    schedule_interval = "0 0 * * *"
)

precheck = FileSensor(
    task_id='check_for_datafile',
    filepath='salesdata_ready.csv',
    start_date=datetime(2024,1,20),
    mode='poke',
    dag=report_dag
)

generate_report_task = BashOperator(
    task_id='generate_report',
    bash_command='generate_report.sh',
    start_date=datetime(2024,1,20),
    dag=report_dag
)

precheck >> generate_report_task


from airflow import DAG
from airflow.operators.bash import BashOperator
from airflow.sensors.filesystem import FileSensor
from datetime import datetime

report_dag = DAG(
    dag_id = 'execute_report',
    schedule_interval = "0 0 * * *"
)

precheck = FileSensor(
    task_id='check_for_datafile',
    filepath='salesdata_ready.csv',
    start_date=datetime(2024,1,20),
    mode='reschedule',
    dag=report_dag
)

generate_report_task = BashOperator(
    task_id='generate_report',
    bash_command='generate_report.sh',
    start_date=datetime(2024,1,20),
    dag=report_dag
)

precheck >> generate_report_task


Implicaciones del executor

Depuración y resolución de problemas en Airflow

DAGs a punto

from airflow import DAG
#from airflow.operators.bash import BashOperator
from airflow.sensors.filesystem import FileSensor
from datetime import datetime

report_dag = DAG(
    dag_id = 'execute_report',
    schedule_interval = "0 0 * * *"
)

precheck = FileSensor(
    task_id='check_for_datafile',
    filepath='salesdata_ready.csv',
    start_date=datetime(2023,2,20),
    mode='poke',
    dag=report_dag)

generate_report_task = BashOperator(
    task_id='generate_report',
    bash_command='generate_report.sh',
    start_date=datetime(2023,2,20),
    dag=report_dag
)

precheck >> generate_report_task


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

sample_dag = DAG(
    dag_id = 'sample_dag',
    schedule_interval = "0 0 * * *"
)

sample_task = BashOperator(
    task_id='sample',
    bash_command='generate_sample.sh',
    start_date=datetime(2023,2,20),
    dag=sample_dag
)


from airflow import DAG
from airflow.operators.bash import BashOperator
from airflow.sensors.filesystem import FileSensor
from datetime import datetime

report_dag = DAG(
    dag_id = 'execute_report',
    schedule_interval = "0 0 * * *"
)

precheck = FileSensor(
    task_id='check_for_datafile',
    filepath='salesdata_ready.csv',
    start_date=datetime(2023,2,20),
    mode='reschedule',
    dag=report_dag)

generate_report_task = BashOperator(
    task_id='generate_report',
    bash_command='generate_report.sh',
    start_date=datetime(2023,2,20),
    dag=report_dag
)

precheck >> generate_report_task


DAG ausente

SLA y generación de informes en Airflow

Definir un SLA

Definir un SLA de tarea

Generar y enviar por email un informe

from airflow import DAG
from airflow.operators.bash import BashOperator
from airflow.sensors.filesystem import FileSensor
from datetime import datetime

default_args={
    '____': [____],
    '____': True,
    'email_on_success': ____
}

report_dag = DAG(
    dag_id = 'execute_report',
    schedule_interval = "0 0 * * *",
    default_args=default_args
)

precheck = FileSensor(
    task_id='check_for_datafile',
    filepath='salesdata_ready.csv',
    start_date=datetime(2023,2,20),
    mode='reschedule',
    dag=report_dag)

generate_report_task = BashOperator(
    task_id='generate_report',
    bash_command='generate_report.sh',
    start_date=datetime(2023,2,20),
    dag=report_dag
)

precheck >> generate_report_task


from airflow import DAG
from airflow.operators.bash import BashOperator
from airflow.sensors.filesystem import FileSensor
from datetime import datetime

default_args={
    'email': ['airflowalerts@datacamp.com', 'airflowadmin@datacamp.com'],
    'email_on_failure': True,
    'email_on_success': True
}

report_dag = DAG(
    dag_id = 'execute_report',
    schedule_interval = "0 0 * * *",
    default_args=default_args
)

precheck = FileSensor(
    task_id='check_for_datafile',
    filepath='salesdata_ready.csv',
    start_date=datetime(2023,2,20),
    mode='reschedule',
    dag=report_dag)

generate_report_task = BashOperator(
    task_id='generate_report',
    bash_command='generate_report.sh',
    start_date=datetime(2023,2,20),
    dag=report_dag
)

precheck >> generate_report_task


Añadir correos de estado

Mantenimiento y monitorización de flujos de trabajo en Airflow

Junta todas las piezas. En este último capítulo, aplicarás todo lo aprendido para construir en Airflow un flujo de trabajo con calidad de producción.

Trabajar con plantillas

Crear un BashOperator con plantilla

Plantillas con varios argumentos

Más plantillas

Usar listas con plantillas

Entender las opciones de parámetros

Enviar correos con plantillas

Ramificación

Define un BranchPythonOperator

Solución de problemas de ramificación

Crear una canalización de producción

Creación de un pipeline de producción #1

Creación de un pipeline de producción #2

Añadiendo los últimos cambios a tu pipeline

¡Enhorabuena!

Creación de canalizaciones de producción en Airflow

Actualizado a Apache Airflow 2.7: entregar datos con una programación puede convertirse en un proceso manual. Escribes scripts, añades tareas cron complejas y pruebas distintas formas de cumplir unos requisitos que cambian constantemente, y gestionar todo esto es aún más complicado cuando trabajas en equipo. Apache Airflow puede quitarte este dolor de cabeza añadiendo planificación, gestión de errores e informes a tus flujos de trabajo. En este curso, dominarás lo básico de Apache Airflow y aprenderás a implementar en producción canalizaciones complejas de ingeniería de datos. También aprenderás a usar grafos acíclicos dirigidos (DAGs), automatizar flujos de trabajo de ingeniería de datos e implementar tareas de ingeniería de datos de forma sencilla y repetible, ayudándote a mantener la calma.

Intermediate Python

Introduction to Shell

Aprende a implementar y programar flujos de trabajo de ingeniería de datos con Airflow y DAGs en Python.

Introducción a Apache Airflow en Python

Aprende a implantar y programar flujos de trabajo de ingeniería de datos.

Ingeniero de datos en Python

Solo se ejecuta una vez por ejecución del DAG.

Sensores vs operadores

Introducción a Apache Airflow en Python

Ejercicio interactivo práctico