In this chapter, you’ll gain a complete introduction to the components of Apache Airflow and learn how and why you should use them.

Introduction to Apache Airflow

Testing a task in Airflow

Examining Airflow commands

Airflow DAGs

Defining a simple DAG

Working with DAGs and the Airflow shell

Troubleshooting DAG creation

Airflow web interface

Starting the Airflow webserver

Navigating the Airflow UI

Examining DAGs with the Airflow UI

Intro to Airflow

What’s up DAG? Now it’s time to learn the basics of implementing Airflow DAGs. Through hands-on activities, you’ll learn how to set up and deploy operators, tasks, and scheduling.

Airflow operators

Defining a BashOperator task

Multiple BashOperators

Airflow tasks

Define order of BashOperators

Determining the order of tasks

Troubleshooting DAG dependencies

Additional operators

Using the PythonOperator

More PythonOperators

EmailOperator and dependencies

Airflow scheduling

Schedule a DAG via Python

Deciphering Airflow schedules

Troubleshooting DAG runs

Implementing Airflow DAGs

In this chapter, you’ll learn how to save yourself time using Airflow components such as sensors and executors while monitoring and troubleshooting Airflow workflows.

Capteurs Airflow

Options

Sensors

Both

Operators

Capteurs vs opérateurs

Privation sensorielle

Les exécuteurs Airflow

Déterminer l’executor

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


Conséquences liées à l’exécuteur

Débogage et résolution des problèmes dans Airflow

DAGs in the bag

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 manquant

SLA et reporting dans Airflow

Définir un SLA

Définir un SLA de tâche

Générer et envoyer un rapport par e‑mail

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


Ajouter des e-mails de statut

Maintaining and monitoring Airflow workflows

Put it all together. In this final chapter, you’ll apply everything you've learned to build a production-quality workflow in Airflow.

Working with templates

Creating a templated BashOperator

Templates with multiple arguments

More templates

Using lists with templates

Understanding parameter options

Sending templated emails

Branching

Define a BranchPythonOperator

Branch troubleshooting

Creating a production pipeline

Creating a production pipeline #1

Creating a production pipeline #2

Adding the final changes to your pipeline

Congratulations!

Building production pipelines in Airflow

Mise à jour vers Apache Airflow 2.7 – Livrer des données selon un calendrier peut vite devenir manuel et fastidieux. Vous écrivez des scripts, ajoutez des tâches cron complexes et essayez différentes approches pour répondre à des exigences qui évoluent sans cesse — et c’est encore plus délicat à gérer en équipe. Apache Airflow peut vous simplifier la vie en intégrant la planification, la gestion des erreurs et le reporting à vos workflows. Dans ce cours, vous allez maîtriser les bases d’Apache Airflow et apprendre à mettre en production des pipelines de data engineering complexes. Vous verrez aussi comment utiliser des graphes acycliques orientés (DAG), automatiser des workflows de data engineering et implémenter des tâches de manière simple et reproductible — de quoi préserver votre sérénité.

Intermediate Python

Introduction to Shell

Apprenez à planifier des workflows de data engineering avec Airflow et DAGs en Python.

Introduction à Apache Airflow en Python

Optimisez vos flux d’ingénierie des données en apprenant à les mettre en œuvre et à les planifier efficacement.

Ingénieur de données en Python

DAG manquant

Introduction à Apache Airflow en Python

Exercice interactif pratique