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

Dag counter

Running your first Dag

Airflow Dags

Creating a simple Dag

Working with Dags and the Airflow shell

Examining Airflow commands

Testing a task in Airflow

Airflow web interface

Navigating the Airflow UI

Examining Dags with the Airflow UI

Dag errors

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

Decorating a task

Defining a BashOperator with @task.bash

Define order of Tasks

More @tasks

Scheduling Dags

Schedule a Dag via Python

Deciphering Airflow schedules

Troubleshooting Dag runs

Passing data between tasks with XCom

Passing data with XCom

XCom variables

Airflow Sensors

Determining the order of tasks

Creating a FileSensor

Sensory Deprivation

Building Dags in Airflow

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.

Monitoring, Alerting, and Callbacks

Implementing a callback function

Implementing an SmtpNotifier

What's in the logs?

Templating with Jinja

Writing with Jinja

Subbing out

Airflow Variables

The answer is Variable

Reading a variable in Python

Variables not quite right

Debugging and troubleshooting in Airflow

Dag troubleshooting

Missing Dag

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.

Déclencheurs et tolérance aux pannes

Implémenter une règle de déclenchement

Ajouter des relances (retries)

Déclencher un Dag enfant

Branchement

Définir un @task.branch

Dépannage de branchement

L’opérateur HITL

Ajouter une approbation humaine

Tâche ApprovalOperator

Créer un pipeline de production

# Imports
from airflow.sdk import ____, ____, ____
from airflow.providers.standard.sensors.filesystem import ____
from airflow.providers.smtp.notifications.smtp import SmtpNotifier

from pendulum import datetime

# Dag definition
@dag(
  start_date=datetime(2026,5,1),
  schedule='____',
  on_success_callback=____(
    to="airflow_team@mycompany.com",
    from_email="airflow_production@mycompany.com",
    subject="sales_etl run succeeded!"
  )
)
def sales_etl():
  
  @task
  def initialize(ds):
    print('Cleaning data directories, prepping for processing...')

    # Use a variable to set the current data path
    Variable.set("sales_etl_data_path", f'/data/sales_etl/{ds}')
  
  
  # Setup a file sensor to wait for the filescopied.txt file to be present
  filewatcher = FileSensor(
    task_id="WaitForDataFiles",
    filepath="{{var.value.____}}/filescopied.txt",
    poke_interval=15,
    timeout=300
  )
    
  initialized = initialize()
  initialized >> filewatcher
  
sales_etl()

# Imports
from airflow.sdk import dag, task, Variable
from airflow.providers.standard.sensors.filesystem import FileSensor
from airflow.providers.smtp.notifications.smtp import SmtpNotifier

from pendulum import datetime

# Dag definition
@dag(
  start_date=datetime(2026,5,1),
  schedule='@monthly',
  on_success_callback=SmtpNotifier(
    to="airflow_team@mycompany.com",
    from_email="airflow_production@mycompany.com",
    subject="sales_etl run succeeded!"
  )
)
def sales_etl():
  
  @task
  def initialize(ds):
    print('Cleaning data directories, prepping for processing...')

    # Use a variable to set the current data path
    Variable.set("sales_etl_data_path", f'/data/sales_etl/{ds}')
  
  
  # Setup a file sensor to wait for the filescopied.txt file to be present
  filewatcher = FileSensor(
    task_id="WaitForDataFiles",
    filepath="{{var.value.sales_etl_data_path}}/filescopied.txt",
    poke_interval=15,
    timeout=300
  )
      
  initialized = initialize()
  initialized >> filewatcher 
  
sales_etl()

Mise en place du pipeline ETL des ventes

# Imports
from airflow.sdk import dag, task, Variable
from airflow.providers.standard.sensors.filesystem import FileSensor
from airflow.providers.smtp.notifications.smtp import SmtpNotifier

from pendulum import datetime

# Dag definition
@dag(
  start_date=datetime(2026,5,1),
  schedule='@monthly',
  on_success_callback=SmtpNotifier(
    to="airflow_team@mycompany.com",
    from_email="airflow_production@mycompany.com",
    subject="sales_etl run succeeded!"
  )
)
def sales_etl():
  
  @task
  def initialize(ds):
    print('Cleaning data directories, prepping for processing...')

    # Use a variable to set the current data path
    Variable.set("sales_etl_data_path", f'/data/sales_etl/{ds}')
  
  
  # Setup a file sensor to wait for the filescopied.txt file to be present
  filewatcher = FileSensor(
    task_id="WaitForDataFiles",
    filepath="{{var.value.sales_etl_data_path}}/filescopied.txt",
    poke_interval=15,
    timeout=300
  )
  
  
  # Create a branching task to validate the data. 
  @____
  def validate_data():
    sales_path = Variable.____('sales_etl_data_path')
    print(f"Parsing data at {sales_path}")
    next_task = '____'
    if '-09-' in sales_path:
      next_task = '____'
    return next_task
  
  
  initialized = initialize()
  validated = validate_data()
  
  initialized >> filewatcher >> validated
  
  
sales_etl()

# Imports
from airflow.sdk import dag, task, Variable
from airflow.providers.standard.sensors.filesystem import FileSensor
from airflow.providers.smtp.notifications.smtp import SmtpNotifier

from pendulum import datetime

# Dag definition
@dag(
  start_date=datetime(2026,5,1),
  schedule='@monthly',
  on_success_callback=SmtpNotifier(
    to="airflow_team@mycompany.com",
    from_email="airflow_production@mycompany.com",
    subject="sales_etl run succeeded!"
  )
)
def sales_etl():
  
  @task
  def initialize(ds):
    print('Cleaning data directories, prepping for processing...')

    # Use a variable to set the current data path
    Variable.set("sales_etl_data_path", f'/data/sales_etl/{ds}')
  
  
  # Setup a file sensor to wait for the filescopied.txt file to be present
  filewatcher = FileSensor(
    task_id="WaitForDataFiles",
    filepath="{{var.value.sales_etl_data_path}}/filescopied.txt",
    poke_interval=15,
    timeout=300
  )
  
  
  # Create a branching task to validate the data. 
  @task.branch
  def validate_data():
    sales_path = Variable.get('sales_etl_data_path')
    print(f"Parsing data at {sales_path}")
    next_task = 'regular_monthly_task'
    if '-09-' in sales_path:
      next_task = 'yearend_approval_task'
    return next_task
  
  
  initialized = initialize()
  validated = validate_data()
  
  initialized >> filewatcher >> validated
  
  
sales_etl()

Ajouter une branche mensuelle

# Imports
from airflow.sdk import dag, task, Variable
from airflow.providers.standard.sensors.filesystem import FileSensor
from airflow.providers.smtp.notifications.smtp import SmtpNotifier
from airflow.providers.standard.operators.hitl import ____

from pendulum import datetime

# Dag definition
@dag(
  start_date=datetime(2026,5,1),
  schedule='@monthly',
  on_success_callback=SmtpNotifier(
    to="airflow_team@mycompany.com",
    from_email="airflow_production@mycompany.com",
    subject="sales_etl run succeeded!"
  )
)
def sales_etl():
  
  @task
  def initialize(ds):
    print('Cleaning data directories, prepping for processing...')

    # Use a variable to set the current data path
    Variable.set("sales_etl_data_path", f'/data/sales_etl/{ds}')
  
  
  # Setup a file sensor to wait for the filescopied.txt file to be present
  filewatcher = FileSensor(
    task_id="WaitForDataFiles",
    filepath="{{var.value.sales_etl_data_path}}/filescopied.txt",
    poke_interval=15,
    timeout=300
  )
  
  
  # Create a branching task to validate the data. 
  @task.branch
  def validate_data():
    sales_path = Variable.get('sales_etl_data_path')
    print(f"Parsing data at {sales_path}")
    next_task = 'regular_monthly_task'
    if '-09-' in sales_path:
      next_task = 'yearend_approval_task'
    return next_task
  
  @task
  def regular_monthly_task():
    print(f'Processing data and automatically updating dataset')
    
  yearend_approval_task = ____(
    task_id="____",
    subject="Sales data processing - Approval Required",
    body=(
            "Please review the sales data produced by the *validate_data* task "
            "on {{ds}}."
            "Approve to push to the data warehouse, or Reject to halt the run."
    ),
  )
  
  @task
  def push_to_warehouse():
    print(f'Pushing processed data to warehouse')
  
  initialized = initialize()
  validated = validate_data()
  
  initialized >> filewatcher >>validated
  
  validated >> [regular_monthly_task(), ____] >> push_to_warehouse()
  
  
sales_etl()

# Imports
from airflow.sdk import dag, task, Variable
from airflow.providers.standard.sensors.filesystem import FileSensor
from airflow.providers.smtp.notifications.smtp import SmtpNotifier
from airflow.providers.standard.operators.hitl import ApprovalOperator

from pendulum import datetime

# Dag definition
@dag(
  start_date=datetime(2026,5,1),
  schedule='@monthly',
  on_success_callback=SmtpNotifier(
    to="airflow_team@mycompany.com",
    from_email="airflow_production@mycompany.com",
    subject="sales_etl run succeeded!"
  )
)
def sales_etl():
  
  @task
  def initialize(ds):
    print('Cleaning data directories, prepping for processing...')

    # Use a variable to set the current data path
    Variable.set("sales_etl_data_path", f'/data/sales_etl/{ds}')
  
  
  # Setup a file sensor to wait for the filescopied.txt file to be present
  filewatcher = FileSensor(
    task_id="WaitForDataFiles",
    filepath="{{var.value.sales_etl_data_path}}/filescopied.txt",
    poke_interval=15,
    timeout=300
  )
  
  
  # Create a branching task to validate the data. 
  @task.branch
  def validate_data():
    sales_path = Variable.get('sales_etl_data_path')
    print(f"Parsing data at {sales_path}")
    next_task = 'regular_monthly_task'
    if '-09-' in sales_path:
      next_task = 'yearend_approval_task'
    return next_task
  
  @task
  def regular_monthly_task():
    print(f'Processing data and automatically updating dataset')
    
  yearend_approval_task = ApprovalOperator(
    task_id="yearend_approval_task",
    subject="Sales data processing - Approval Required",
    body=(
            "Please review the sales data produced by the *validate_data* task "
            "on {{ds}}."
            "Approve to push to the data warehouse, or Reject to halt the run."
    ),
  )
  
  @task
  def push_to_warehouse():
    print(f'Pushing processed data to warehouse')
  
  initialized = initialize()
  validated = validate_data()
  
  initialized >> filewatcher >>validated
  
  validated >> [regular_monthly_task(), yearend_approval_task] >> push_to_warehouse()
  
  
sales_etl()

Ajouter une étape d’approbation humaine

Félicitations !

Controlling Dag Logic

Mise à jour vers Apache Airflow 3.1.6 – Livrer des données selon un planning peut vite devenir un processus manuel. Vous écrivez des scripts, ajoutez des tâches cron complexes et tentez diverses approches pour répondre à des besoins qui évoluent sans cesse — et c’est encore plus difficile à gérer en équipe. Apache Airflow élimine ce casse-tête en ajoutant la planification, la gestion des erreurs et le reporting à vos workflows. Dans ce cours, vous maîtriserez les bases d’Apache Airflow et apprendrez à mettre en production des pipelines de data engineering complexes. Vous verrez aussi comment utiliser des graphes orientés acycliques (DAGs), 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

Ajouter une branche mensuelle

Introduction à Apache Airflow en Python

Exercice interactif pratique