Definisci un @task.branch
Dopo aver imparato la potenza della logica condizionale in Airflow, vuoi provare il decoratore @task.branch. Vorresti eseguire un percorso di codice diverso se la data di esecuzione corrente rappresenta un nuovo anno (ad es., 2026 vs 2025).
Il Dag è già definito per te, insieme ai task in questione. Il tuo compito ora è implementare @task.branch.
Questo esercizio fa parte del corso
Introduzione ad Apache Airflow in Python
Istruzioni dell'esercizio
- Aggiungi le variabili template di data di Airflow appropriate come parametri di
year_checkper confrontare l'anno di esecuzione corrente e quello precedente. - Assegna
current_yeareprevious_yearestraendo (slice) i primi 4 caratteri da ciascun parametro di data. - Imposta le dipendenze su
current_year_taskenew_year_task.
Esercizio pratico interattivo
Prova a risolvere questo esercizio completando il codice di esempio.
@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