En este capítulo, obtendrás una introducción completa a los componentes de Apache Airflow y aprenderás cómo y por qué debes utilizarlos.

Introducción a Apache Airflow

Probar una tarea en Airflow

Examinar los comandos de flujo de aire

Flujo de aire DAGs

Definir una simple DAG

Trabajar con DAGs y la carcasa Airflow

Solución de problemas en la creación de DAG

Interfaz web Airflow

Iniciar el servidor web Airflow

Navegar por el flujo de aire UI

Examinar DAGs con el Flujo de aire UI

Introducción al flujo de aire

¿Qué pasa DAG? Ahora es el momento de aprender los fundamentos de la aplicación de Airflow DAGs. Mediante actividades prácticas, aprenderás a configurar y desplegar operadores, tareas y programación.

Operadores de flujo de aire

Definir una tarea BashOperator

Múltiples BashOperators

Tareas de flujo de aire

Definir el orden de los BashOperators

Determinar el orden de las tareas

Solución de problemas DAG dependencias

Operadores adicionales

Utilizar el Operador Python

Más PythonOperators

EmailOperator y dependencias

Programación del flujo de aire

Programa un DAG mediante Python

Descifrar los horarios del flujo de aire

Solución de problemas DAG runs

Implementación del flujo de aire DAGs

En este capítulo, aprenderás a ahorrarte tiempo utilizando componentes de Airflow, como sensores y ejecutores, mientras supervisas y solucionas problemas de los flujos de trabajo de Airflow.

Sensores de flujo de aire

Opciones

Sensores

Ambos

Operarios

Sensores frente a operadores

Privación sensorial

Ejecutores de flujo de aire

Determinación del albacea

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 para el albacea

Depuración y resolución de problemas en Airflow

DAGs en la bolsa

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


Falta DAG

SLAs e informes en Airflow

Definición de una SLA

Definir una tarea SLA

Generar y enviar por correo electrónico 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 electrónicos de estado

Mantener y supervisar los flujos de trabajo de Airflow

Ponlo todo junto. En este capítulo final, aplicarás todo lo que has aprendido para crear un flujo de trabajo de calidad de producción en Airflow.

Trabajar con plantillas

Crear un BashOperator con plantilla

Plantillas con múltiples argumentos

Más plantillas

Utilizar listas con plantillas

Comprender las opciones de los parámetros

Enviar correos electrónicos con plantillas

Ramificación

Definir un BranchPythonOperator

Solución de problemas de la sucursal

Crear un canal de producción

Crear un canal de producción nº 1

Crear una cadena de producción nº 2

Añadir los cambios finales a tu pipeline

¡Enhorabuena!

Construir conductos de producción en Airflow

Ahora actualizado a Apache Airflow 2.7 - Entregar datos de forma programada puede ser un proceso manual. Escribes secuencias de comandos, añades complejas tareas cron e intentas varias formas de cumplir una serie de requisitos siempre cambiantes, y es aún más complicado gestionarlo todo cuando se trabaja con compañeros de equipo. Apache Airflow puede eliminar este quebradero de cabeza añadiendo programación, gestión de errores e informes a tus flujos de trabajo. En este curso, dominarás los fundamentos de Apache Airflow y aprenderás a implementar canalizaciones complejas de ingeniería de datos en producción. También aprenderás a utilizar grafos acíclicos dirigidos (DAGs), a automatizar los flujos de trabajo de ingeniería de datos y a realizar tareas de ingeniería de datos de forma fácil y repetible, lo que te ayudará a mantener la cordura.

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 implementar y programar flujos de trabajo de ingeniería de datos.

Ingeniero de datos en Python

DAGs en la bolsa

Introducción a Apache Airflow en Python

Ejercicio interactivo práctico