[Airlfow] Trigger Dag Run 오퍼레이터

1. DAG 간 의존관계 설정 방법

• TriggerDagRun 오퍼레이터
  어떤 DAG를 트리거 할지 지정할 수 있음

 

• ExternalTask 센서
  어떤 DAG에 어떤 Task가 완료되었는지 감지하는 센서

 

 

비교	TriggerDagRun 오퍼레이터	ExternalTask 센서
방식	실행할 다른 DAG의 ID를 지정하여 수행	본 Task가 수행되기 전 다른 DAG의 완료를 기다린 후 수행
권고 사용시점	Trigger 되는 DAG의 선행 DAG이 하나만 있을 경우	Trigger 되는 DAG의 선행 DAG이 2개 이상인 경우

 

 

 

 

 

1. Trigger Dag Run 오퍼레터 사용 예제코드

'''
dags_trigger_dag_run_operator.py
'''

from airflow import DAG
from airflow.operators.bash import BashOperator
from airflow.operators.trigger_dagrun import TriggerDagRunOperator
import pendulum


with DAG(
    dag_id = 'dags_trigger_dag_run_operator',
    start_date = pendulum.datetime(2023, 4, 1, tz='Asia/Seoul'),
    schedule = '30 9 * * *',
    catchup = False
) as dag:

    start_task = BashOperator(
        task_id = 'start_task',
        bash_command = 'echo "start!"',
    )

    trigger_dag_task = TriggerDagRunOperator(
        task_id = 'trigger_dag_task',                     # task의 id (필수값)
        trigger_dag_id = 'dags_python_operator',          # 어떤 DAG을 trigger 할지 명시(필수값)
        trigger_run_id = None,                            # DAG의 수행 방식과 시간을 유일하게 식별해주는 키
        execution_date = '{{data_interval_start}}',       # manual_{{execution_date}}로 수행
        reset_dag_run = True,                             # DAG 수행 이력이 있어도 수행할 것인지 여부
        wait_for_completion=False,                        # 트리거 DAG이 완료 될 때까지 기다릴 것인지 여부
        poke_interval=60,                                 # 트리거 DAG의 모니터링 주기
        allowed_states=['success'],                       # 트리거덱테스크가 success로 끝나려면 트리거 DAG이 어떤 상태로 끝나야 하는지 명시 (리스트 형태로 복수의 상태 명시 가능)
        failed_states=None                                # 트리거덱테스크가 fail로 끝나려면 트리거 DAG이 어떤 상태로 끝나야 하는지 명시 (리스트 형태로 복수의 상태 명시 가능)
    )


    start_task >> trigger_dag_task

 

 

실행결과 Graph

 

 

트리거 DAG으로 설정한 DAG의 태스크 실행결과 Run ID가 메뉴얼로 명시된것을 볼 수 있음

(trigger_run_id 값을 None으로 주고 execution_date 값만 지정해줬기 때문에 )