Defina um @task.branch
Depois de aprender sobre o poder da lógica condicional no Airflow, você quer testar o decorator @task.branch. Você deseja executar um caminho de código diferente se a data de execução atual representar um novo ano (por exemplo, 2026 vs 2025).
O Dag já está definido para você, junto com as tarefas em questão. Sua tarefa agora é implementar o @task.branch.
Este exercício faz parte do curso
Introdução ao Apache Airflow em Python
Instruções do exercício
- Adicione as variáveis de template de data apropriadas do Airflow como parâmetros de
year_checkpara comparar o ano de execução atual e o anterior. - Atribua
current_yeareprevious_yearcortando os primeiros 4 caracteres de cada parâmetro de data. - Configure as dependências em
current_year_taskenew_year_task.
Exercício interativo prático
Experimente este exercício completando este código de exemplo.
@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