定义一个 @task.branch
在学习了 Airflow 中条件逻辑的强大作用后，您希望试用一下 @task.branch 装饰器。您想在当前执行日期代表新年时（例如 2026 vs 2025）走不同的代码路径。
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