Define un @task.branch
Después de conocer el poder de la lógica condicional en Airflow, quieres probar el decorador @task.branch. Te gustaría ejecutar una ruta de código diferente si la fecha de ejecución actual corresponde a un año nuevo (p. ej., 2026 frente a 2025).
El Dag ya está definido para ti, junto con las tareas en cuestión. Tu tarea ahora es implementar el @task.branch.
Este ejercicio forma parte del curso
Introducción a Apache Airflow en Python
Instrucciones del ejercicio
- Añade las variables de plantilla de fecha de Airflow adecuadas como parámetros de
year_checkpara comparar el año de ejecución actual y el anterior. - Asigna
current_yearyprevious_yearcortando los primeros 4 caracteres de cada parámetro de fecha. - Establece las dependencias de
current_year_taskynew_year_task.
Ejercicio interactivo práctico
Prueba este ejercicio y completa el código de muestra.
@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