Definește un @task.branch

După ce ai aflat despre puterea logicii condiționale în Airflow, vrei să testezi decoratorul @task.branch. Îți dorești să rulezi un traseu de cod diferit dacă data curentă de execuție reprezintă un an nou (de exemplu, 2026 față de 2025).

DAG-ul este deja definit pentru tine, împreună cu taskurile relevante. Sarcina ta acum este să implementezi @task.branch.

Acest exercițiu face parte din cursul

Introducere în Apache Airflow în Python

Vezi cursul

Instrucțiuni pentru exercițiu

Adaugă variabilele de șablon de dată Airflow corespunzătoare ca parametri ai funcției year_check, pentru a compara anul de execuție curent cu cel anterior.
Atribuie valorile current_year și previous_year prin extragerea primelor 4 caractere din fiecare parametru de dată.
Setează dependențele față de current_year_task și new_year_task.

Exercițiu interactiv practic

Încearcă acest exercițiu completând acest cod de exemplu.

@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

Editează și rulează codul