@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