[Airflow] TriggerDagRunOperator 사용할 때 duplicate key value violates unique constraint "dag_run_dag_id_execution_date_key"

 

개요

Airflow을 활용하여 크롤링을 진행하고 있었다.

하나의 DAG가 특정 DAG를 한번에 여러번 trigger 하는 프로세스인데

지금 계속 크롤링 사이트가 늘어나는 과정에 있었다.

 

크롤링 사이트가 300개 가까이 늘어나고, 하나의 DAG를 300개 가까이 호출하다가 오류가 발생했다.

 

 

에러는 trigger_internal_dag task가 Fail

 

확인해보자

error message: run_id 중복 오류

sqlalchemy.exc.IntegrityError: (psycopg2.errors.UniqueViolation) duplicate key value violates unique constraint "dag_run_dag_id_execution_date_key"

 

 

• trigger_internal_dag은 매우 짧은 시간 안에 동일한 DAG를 여러 번 트리거(trigger)하려고 시도
• 이때 Airflow 내부에서 실행하는 시간으로 run_id를 자동으로 부여하는데,
• 동시에 여러개가 실행되며 같은 run_id를 가지는 경우가 발생 → 그래서 중복 오류 발생!

 

이렇게 변경

변경 전:

trigger_internal = TriggerDagRunOperator.partial(
    task_id="trigger_internal_dag",
    trigger_dag_id="execute_crawler",
    wait_for_completion=False,
).expand(conf=internal_merged_confs)

 

변경 후:

@task
def trigger_internal_dag_dynamically(payload: dict):
    category_id = payload.get("category_id")
    run_id = f"internal_{category_id}_{timezone.utcnow().strftime('%Y%m%dT%H%M%S.%f')}"
    print(f"Triggering execute_crawler with run_id: {run_id}")
    trigger_dag(
        dag_id="execute_crawler",
        run_id=run_id,
        conf=payload,
        execution_date=timezone.utcnow(),
        replace_microseconds=False
    )

...

trigger_internal = trigger_internal_dag_dynamically.expand(payload=internal_merged_confs)

 

뭘 변경했을까

1. Task 데코레이터 적용
   1. 이전: Airflow에 내장된 TriggerDagRunOperator 사용 → .expand()와 함께 사용할 때 run_id를 동적으로 만드는 데 문제 발생
   2. 변경: trigger_internal_dag_dynamically라는 이름의 파이썬 함수를 생성 → @task 데코레이터를 붙여 Airflow가 이 함수를 하나의 task로 인식하게 함
2. run_id 생성 방식
   1. 이전: Airflow가 내부 규칙에 따라 run_id을 자동으로 만들도록 함, 이 과정에서 동시에 trigger 되는 경우 같은 run_id을 생성하며 에러가 발생
   2. 현재: run_id = f"internal_{category_id}_{timezone.utcnow()...}" 코드를 통해 각 task마다 고유한 run_id를 직접 명시적으로 생성 → category_id와 현재 시간을 조합해서 절대 겹치지 않게 함
3. DAG 실행 호출 방식
   1. 이전: TriggerDagRunOperator가 내부적으로 DAG 실행
   2. 현재: Airflow의 API인 trigger_dag() 함수를 직접 호출 → dag_id, run_id, conf를 전달

 

결론

동일한 Dag를 동시에 300여개 정도 호출해도 각각의 run_id를 가지고 있기에 → 중복 호출 오류가 사라짐

 

 

실제로 운영하면서 생기는 문제를 발생하는 건 재밌다

DAG를 더 많이 호출하면 또 어떤 문제가 발생할까

내가 또 놓친 부분은 어디일까