Definiowanie @task.branch

Po zapoznaniu się z możliwościami logiki warunkowej w Airflow chcesz przetestować dekorator @task.branch. Zależy ci na uruchomieniu innej ścieżki kodu, jeśli bieżąca data wykonania oznacza nowy rok (np. 2026 zamiast 2025).

DAG jest już dla ciebie zdefiniowany, wraz z odpowiednimi zadaniami. Twoim zadaniem jest teraz zaimplementowanie @task.branch.

To ćwiczenie jest częścią kursu

Wprowadzenie do Apache Airflow w Pythonie

Zobacz kurs

Instrukcje do ćwiczenia

Dodaj odpowiednie zmienne szablonów dat Airflow jako parametry funkcji year_check, aby porównać bieżący i poprzedni rok wykonania.
Przypisz current_year i previous_year, wycinając pierwsze 4 znaki z każdego parametru daty.
Ustaw zależności dla current_year_task i new_year_task.

Interaktywne ćwiczenie praktyczne

Spróbuj tego ćwiczenia, uzupełniając ten przykładowy kod.

@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

Edytuj i uruchom kod