@task.branch の定義
Airflow における条件分岐ロジックの仕組みを学んだあと、@task.branch デコレータを実際に試してみましょう。現在の実行日が新しい年（たとえば 2025 年から 2026 年）を表している場合に、異なるコードパスを実行したいと思います。
DAG はあらかじめ定義されており、関連するタスクも用意されています。ここでは @task.branch の実装に取り組みましょう。
この演習はコースの一部です
Python で学ぶ Apache Airflow 入門
演習の手順
- 現在の実行年と前回の実行年を比較するために、適切な Airflow の日付テンプレート変数を
year_checkのパラメータとして追加しましょう。 - 各日付パラメータの先頭 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