Définir un @task.branch
Après avoir vu la puissance de la logique conditionnelle dans Airflow, vous souhaitez essayer le décorateur @task.branch. Vous voulez exécuter un chemin de code différent si la date d'exécution actuelle correspond à une nouvelle année (p. ex., 2026 vs 2025).
Le Dag est déjà défini, ainsi que les tâches visées. Votre tâche actuelle est d'implanter le @task.branch.
Cette activité fait partie du cours
Introduction à Apache Airflow en Python
Instructions de l’exercice
- Ajoutez les variables de gabarit de date d'Airflow appropriées comme paramètres de
year_checkpour comparer l'année d'exécution actuelle et la précédente. - Attribuez
current_yearetprevious_yearen découpant les 4 premiers caractères de chaque paramètre de date. - Définissez les dépendances sur
current_year_tasketnew_year_task.
Exercice interactif pratique
Essayez cet exercice en complétant ce code d’exemple.
@dag(start_date=datetime(2026,5,1), schedule='@monthly')
def process_yearly_expenses():
# Create a function to determine if years are different
@task.branch
def year_check(____, ____):
current_year = int(____[0:4])
previous_year = int(____[0:4])
if current_year == previous_year:
return 'current_year_task'
else:
return 'new_year_task'
# Define the dependencies
branch_task __ current_year_task
____ >> new_year_task