[TroubleShooting] 동일한 dag_id로 여러가지의 dag를 만들었을 때 문제 상황
Issue Background
최근 스마트팩토리 온도 센서 데이터를 MySQL로 적재하는 ETL 파이프라인 실습을 진행하며 겪은 치명적인 오류와 그 해결 과정을 공유합니다.
비슷한 형태의 파이프라인 여러 개를 테스트하기 위해 코드를 여러 파이썬 파일(file_A.py, file_B.py)로 나누어 작성했습니다. 이때, "동일한 dag_id를 부여하면 AWS의 로드밸런서(ELB)처럼 스케줄러가 여러 DAG를 라운드로빈 방식으로 분산 처리해 주지 않을까?"라는 가설을 세웠습니다.
하지만 결과는 처참한 파이프라인 붕괴였습니다.
Symptoms (발생한 증상)
대시보드와 실행 로그에서 다음과 같은 기괴한 현상들이 발생했습니다.
- 대시보드의 깜빡임(Flickering) 현상: 웹 UI에서 새로고침(F5)을 누를 때마다 DAG의 구조나 태스크 목록이 계속 다른 파일의 형태로 바뀌어 보였습니다.
- Task Failed 및 무한 Retry 지옥: 워커가 태스크를 실행하려다 말고 갑자기
Task not found in DAG라는 에러를 뱉으며 실패했습니다. Airflow는 재시도(Retry)를 지시하지만, 영원히 성공하지 못하고 상태가 꼬여버렸습니다.
Root Cause Analysis (원인 분석)
결론부터 말하자면, 저의 가설은 Airflow의 시스템 아키텍처를 오판한 치명적인 착각이었습니다. Airflow는 DAG 단위로 로드밸런싱을 수행하는 네트워크 도구가 아닙니다.
이 사태의 근본적인 원인은 '메타데이터 DB의 고유키 충돌'과 '스케줄러의 파싱(Parsing) 방식'에 있었습니다.
1. dag_id는 데이터베이스의 Primary Key(고유키)다
Airflow의 모든 설정과 상태는 메타데이터 DB(Postgres, MySQL 등)에 저장됩니다. 이때 dag 테이블에서 dag_id는 무조건 고유해야 하는 Primary Key(PK)입니다.
동일한 식별자를 가진 파일이 여러 개 존재한다는 것은, 분산 처리가 아니라 데이터 충돌(Collision)을 의미합니다.
2. Scheduler의 덮어쓰기(Overwrite) 경합 (Race Condition)
Airflow 스케줄러는 백그라운드에서 dags/ 폴더 내의 파일들을 주기적으로 순회하며 파싱(Parsing)합니다.
- T1 시점: 스케줄러가
file_A.py를 파싱합니다. DB의dag_id="my_etl"정보가file_A의 구조로 덮어써집니다. (UI에는 A가 보임) - T2 시점: 몇 초 뒤, 스케줄러가
file_B.py를 파싱합니다. DB의my_etl정보가file_B의 구조로 다시 덮어써집니다. (새로고침하면 UI에 B가 보임)
즉, 로드밸런싱이 되는 것이 아니라 스케줄러가 파싱할 때마다 DB 레코드가 엎치락뒤치락하며 덮어쓰기 전쟁을 하고 있었던 것입니다.
3. 고아 태스크(Orphan Task)의 발생
이 상태에서 Worker가 file_A의 task_1을 실행하려는데, 찰나의 순간 스케줄러가 file_B를 파싱해버렸다고 가정해 봅시다.
DB 상의 DAG 구조는 file_B로 바뀌었으므로, Worker는 자신이 실행해야 할 task_1을 DB에서 찾지 못합니다. 그 결과 "DAG 안에 해당 Task가 존재하지 않는다"며 비정상 종료(Failed) 처리되는 것입니다.
Solution & Action Item (해결책)
시스템의 원리를 이해했다면 해결책은 간단합니다.
1. 파일별로 독립적인 dag_id 부여
각 파이썬 파일마다 반드시 고유한 dag_id를 사용해야 합니다. (예: 05_mysql_etl_sensor_A, 05_mysql_etl_sensor_B)
2. Dynamic DAG (동적 DAG) 활용
만약 수십 개의 센서에서 들어오는 데이터를 동일한 로직으로 처리해야 해서 하드코딩이 어렵다면, 하나의 파이썬 파일 내에서 for 문을 활용해 동적 DAG를 생성하는 방식을 채택해야 합니다.
# Dynamic DAG 생성 예시
sensors = ['sensor_A', 'sensor_B', 'sensor_C']
for sensor in sensors:
dag_id = f"05_mysql_etl_{sensor}"
with DAG(dag_id=dag_id, ...) as dag:
# 태스크 정의
# ...
# 글로벌 네임스페이스에 DAG 등록 (Airflow가 인식하게 함)
globals()[dag_id] = dag📝 Conclusion (마치며)
이번 트러블슈팅을 통해 Airflow의 스케줄러가 파일을 파싱하는 루프 방식과, 메타데이터 DB가 전체 시스템의 중심에서 어떻게 작동하는지 명확히 깨달았습니다.
