[Airflow] Duplicate TaskIDFound exception 에러 해결 방법

 

 

 

[문제 상황]

DAG 생성 후 아래와 같은 에러가 발생했고, 에러 메세지를 읽어보니 TaskID가 겹친다는 것을 확인했다.

 

Error message
: DAG Import Error with Duplicate TaskIDFound exception

 

Error message

 

 

 

[해결 방법]

Airflow task object 선언 시에 중복되었던 task_id를 찾아서, 서로 중복되지 않도록 id를 수정하여 문제를 해결할 수 있었다.

 

# 수정전(t1_orange task_id 중복)

# 수정전 코드

from airflow import DAG
import pendulum
import datetime
from airflow.operators.bash import BashOperator

with DAG(
    dag_id="dags_bash_select_fruit",
    schedule="10 0 * * 6#1",    # 매월 첫번째주 토요일 0시 10분
    start_date=pendulum.datetime(2024, 1, 4, tz="Asia/Seoul"),
    catchup=False
) as dag:
    
    t1_orange=BashOperator(
        task_id="t1_orange",
        bash_command="/opt/Airflow/plugins/shell/select_fruit.sh ORANGE"
    )

    t2_avocado=BashOperator(
        task_id="t1_orange",
        bash_command="/opt/Airflow/plugins/shell/select_fruit.sh AVOCADO"
    )

    # 실행순서
    t1_orange >> t2_avocado

 

 

# 수정후(중복되었던 task_id 수정)

# 수정후 코드
from airflow import DAG
import pendulum
import datetime
from airflow.operators.bash import BashOperator

with DAG(
    dag_id="dags_bash_select_fruit",
    schedule="10 0 * * 6#1",    # 매월 첫번째주 토요일 0시 10분
    start_date=pendulum.datetime(2024, 1, 4, tz="Asia/Seoul"),
    catchup=False
) as dag:
    
    t1_orange=BashOperator(
        task_id="t1_orange",
        bash_command="/opt/Airflow/plugins/shell/select_fruit.sh ORANGE"
    )

    t2_avocado=BashOperator(
        task_id="t2_avocado",   # <------------------ 수정한 부분
        bash_command="/opt/Airflow/plugins/shell/select_fruit.sh AVOCADO"
    )

    # 실행순서
    t1_orange >> t2_avocado

 

 

# 오류해결

 

 

 

코드를 다시 확인해보니 task object 선언 시에 설정했던 task_id가 중복되어 있음을 확인했고, task_id가 중복되지 않도록 수정하여 해결할 수 있었다. Airflow task를 만드는 과정에서 이러한 작은 실수에도 매우 민감하다는 것을 알 수 있었고, 고유한 task_id를 만들기 위해 주의를 기울여야겠다고 생각하는 계기가 되었다.