Định nghĩa một @task.branch
Sau khi tìm hiểu về sức mạnh của logic điều kiện trong Airflow, bạn muốn thử decorator @task.branch. Bạn muốn chạy nhánh mã khác nếu ngày thực thi hiện tại rơi vào năm mới (ví dụ, 2026 so với 2025).
Dag đã được định nghĩa sẵn, cùng với các task liên quan. Nhiệm vụ hiện tại của bạn là triển khai @task.branch.
Bài tập này là một phần của khóa học
Giới thiệu về Apache Airflow bằng Python
Hướng dẫn bài tập
- Thêm các biến template ngày tháng của Airflow phù hợp làm tham số cho
year_checkđể so sánh năm thực thi hiện tại và trước đó. - Gán
current_yearvàprevious_yearbằng cách cắt 4 ký tự đầu tiên từ mỗi tham số ngày. - Thiết lập phụ thuộc cho
current_year_taskvànew_year_task.
Bài tập tương tác thực hành trực tiếp
Hãy thử làm bài tập này bằng cách hoàn thành đoạn mã mẫu này.
@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