# Trigger Rule에 따라 Task 실행 여부를 결정하기

# Graph View

다음처럼 간단한 Task 의존성을 가지는 DAG을 작성해볼 것입니다.

간단해 보이지만 조금 특이한 점은 task_3가 실패한 상태임에도 task_4가 실행되어 성공한 상태로 남아있다는 것입니다. Task Instance의 실행 조건을 Trigger Rule이라고 부르는데, 기본적으로 의존하는 모든 Task Instance가 성공 상태여야 해당 Task Instance가 실행됩니다.

여기서는 Trigger Rule의 기본 값이 아닌 다른 옵션으로 수정해보겠습니다.

# Code

from datetime import datetime, timedelta
from time import sleep

from airflow import DAG
from airflow.operators.python import PythonOperator
from pendulum.tz.timezone import Timezone

with DAG(
    dag_id="07_conditional_tasks_with_trigger_rule",
    description="Task 의존성에 trigger_rule을 통해 조건문을 가지는 DAG 예제입니다.",
    default_args={
        "owner": "heumsi",
        "retries": 0,
        "retry_delay": timedelta(minutes=1),
    },
    start_date=datetime(2022, 1, 20, tzinfo=Timezone("Asia/Seoul")),
    schedule_interval="@once",
    tags=["examples", "01_writing_various_task_flows"],
) as dag:

    def dump() -> None:
        sleep(3)

    def must_fail() -> None:
        raise

    t1 = PythonOperator(task_id="task_1", python_callable=dump)
    t2 = PythonOperator(task_id="task_2", python_callable=dump)
    t3 = PythonOperator(task_id="task_3", python_callable=must_fail)
    t4 = PythonOperator(
        task_id="task_4", python_callable=dump, trigger_rule="one_success"
    )

    t1 >> [t2, t3] >> t4

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34

PythonOperator의 trigger_rule 파라미터 값으로 "one_success"을 넘겨주었습니다.
- 이는 의존하는 Task Instance의 상태가 단 하나라도 성공 상태이면 해당 Task Instance를 실행한다는 의미입니다.
- trigger_rule의 기본 값은 "all_success" 입니다.

TIP

더 다양한 Trigger Rule 옵션을 보고 싶으면 공식 문서 (opens new window)를 확인해보세요.

# Web UI

← 직전 DAG Run의 상태에 따라 Task 실행 여부를 결정하기 SubDAG 만들기 →