Definiere ein @task.branch

Nachdem du die Möglichkeiten von bedingter Logik in Airflow kennengelernt hast, möchtest du den @task.branch-Decorator ausprobieren. Du willst einen anderen Codepfad ausführen, wenn das aktuelle Ausführungsdatum ein neues Jahr darstellt (z. B. 2026 vs. 2025).

Das Dag ist für dich definiert, ebenso die betreffenden Tasks. Deine aktuelle Aufgabe ist es, das @task.branch zu implementieren.

Diese Übung ist Teil des Kurses

<Kurs>Einführung in Apache Airflow mit Python</Kurs>

Kurs ansehen

Übungsanweisungen

Füge die passenden Airflow-Datentemplate-Variablen als Parameter zu year_check hinzu, um das aktuelle mit dem vorherigen Ausführungsjahr zu vergleichen.
Weise current_year und previous_year zu, indem du die ersten 4 Zeichen aus jedem Datumsparameter ausschneidest.
Setze die Abhängigkeiten zu current_year_task und new_year_task.

Interaktive praktische Übung

Versuche dich an dieser Übung, indem du diesen Beispielcode vervollständigst.

@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

Code bearbeiten und ausführen