자식 Dag 트리거하기
일부 워크플로우에서 유사한 구성 요소를 사용하고 있다는 것을 발견하고, 공통 태스크를 별도의 Dag으로 분리할 수 있다는 것을 깨달았습니다. 이렇게 하면 여러 복사본을 유지 관리하지 않고도 필요할 때마다 해당 구성 요소를 실행할 수 있습니다. 현재 워크플로우 내의 태스크를 통해 자식 Dag을 실행하기로 결정했습니다.
dag, task, datetime 컴포넌트는 이미 가져와져 있습니다.
이 연습은 강의의 일부입니다
Python으로 배우는 Apache Airflow 입문
연습 안내
- 워크플로우 내에서 Dag을 시작하는 데 필요한 오퍼레이터를 가져오세요.
child_pipeline이라는 이름의 Dag을 트리거하도록 오퍼레이터를 설정하세요.- 트리거된 Dag이 완료될 때까지 부모 Dag이 기다린 후 계속 진행하도록 설정하세요.
- 오퍼레이터가 자식 Dag의 완료 여부를 확인하는 빈도를 설정하세요.
실습형 인터랙티브 연습
이 예제를 이 샘플 코드를 완성하여 풀어보세요.
# Import TriggerDagRunOperator
from airflow.providers.standard.operators.trigger_dagrun import ____
@dag(start_date=datetime(2026, 1, 1))
def parent_orchestrator_dag():
# Trigger child_pipeline and wait for it to complete
trigger_child = TriggerDagRunOperator(
task_id="trigger_child_pipeline",
trigger_dag_id="____",
____=True,
____=30,
conf={"source": "s3://my-bucket/raw/"})
validate() >> trigger_child >> post_trigger_summary()
parent_orchestrator_dag()