定義一個 @task.branch
在學完 Airflow 中條件邏輯的威力後，你想試試 @task.branch 裝飾器。你希望當目前的執行日期代表進入新的一年時（例如 2026 相較於 2025），走不同的程式碼路徑。
Dag 和相關任務都已替你定義好。你現在的工作是實作 @task.branch。
本練習屬於課程
Python 中的 Apache Airflow 入門
練習說明
- 在
year_check中加入合適的 Airflow 日期樣板變數作為參數，用來比較目前與前一次執行的年份。 - 透過擷取每個日期參數的前 4 個字元，分別指定
current_year與previous_year。 - 設定
current_year_task與new_year_task的相依關係。
動手互動練習
試著完成這個範例程式碼，體驗一下這個練習。
@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