[Airflow] trigger 시도 시 DagRunAlreadyExists: A Dag Run already exists for dag id at {execution_date} with run id

개요

모든 Task에 대해 clear 처리한 DAG를 다시 실행시키고자 trigger 명령을 사용했다.

airflow dags trigger \
    -r scheduled__2022-11-12T00:00:00+00:00 \
    test_dag

하지만 아래와 같은 DagRunAlreadyExists 에러가 발생하면서 트리거가 되지 않았다.

Traceback (most recent call last):
  File "/opt/airflow/conda/bin/airflow", line 8, in <module>
    sys.exit(main())
  File "/opt/airflow/conda/lib/python3.7/site-packages/airflow/__main__.py", line 48, in main
    args.func(args)
  File "/opt/airflow/conda/lib/python3.7/site-packages/airflow/cli/cli_parser.py", line 48, in command
    return func(*args, **kwargs)
  File "/opt/airflow/conda/lib/python3.7/site-packages/airflow/utils/cli.py", line 92, in wrapper
    return f(*args, **kwargs)
  File "/opt/airflow/conda/lib/python3.7/site-packages/airflow/cli/commands/dag_command.py", line 138, in dag_trigger
    dag_id=args.dag_id, run_id=args.run_id, conf=args.conf, execution_date=args.exec_date
  File "/opt/airflow/conda/lib/python3.7/site-packages/airflow/api/client/local_client.py", line 32, in trigger_dag
    dag_id=dag_id, run_id=run_id, conf=conf, execution_date=execution_date
  File "/opt/airflow/conda/lib/python3.7/site-packages/airflow/api/common/trigger_dag.py", line 124, in trigger_dag
    replace_microseconds=replace_microseconds,
  File "/opt/airflow/conda/lib/python3.7/site-packages/airflow/api/common/trigger_dag.py", line 74, in _trigger_dag
    f"A Dag Run already exists for dag id {dag_id} at {execution_date} with run id {run_id}"
airflow.exceptions.DagRunAlreadyExists: A Dag Run already exists for dag id test_dag at 2022-11-14 11:48:49+00:00 with run id scheduled__2022-11-12T00:00:00+00:00

 

해결 방법을 정리해둔다.

 

원인

지정한 run_id로 생성된 DAG run이 이미 존재하고 있는 것이 원인이다.

execution_date를 함께 지정하는 경우, 지정한 execution_date로 실행한 DAG run이 존재하면, run_id가 달라고 발생하는 것 같다.

 

 

 

해결 방법

이런 경우에는 기존에 실행했던 DAG run 정보를 삭제한 후, 트리거하면 된다.

DAG run을 삭제할 때는 Web UI를 통하거나 Airflow Meta DB의 dagrun 테이블에서 관련 정보를 삭제하면 된다. 다만 Meta DB에서 직접 정보를 삭제하는 건 권장하기 어려우므로(……) 이 글에서는 Web UI를 통한 방법만을 정리한다.

 

1. 상단 메뉴바 > Browse > DAG Runs로 접근

 

2. DAG run을 삭제한 dag 검색

적절한 필터링 조건을 걸어서 찾는다.

 

3. 삭제한 DAG run의 Delete record 버튼 클릭

 

4. 선택한 DAG run을 삭제하는 게 맞는지 확인한다.

 

 

 

 

 

 

 

 

참고 문서

https://stackoverflow.com/questions/58459855/how-to-delete-a-dag-run-in-apache-airflow