Airflow

dag.test()로 단일 프로세스에서 Dag 실행하기

Dag 파일에 dag.test() 호출을 설정해두면, 단일 직렬화된 파이썬 프로세스 안에서 Dag 전체를 실행하며 IDE에서 디버깅할 수 있다. 이 방식은 로컬 SQLite를 포함해 어떤 지원 DB에서도 쓸 수 있고, 모든 태스크가 한 프로세스에서 실행되므로 빠르게 실패(fail fast)한다.

dag.test()는 실제 Dag run을 실행하므로 두 가지 전제조건이 있다. 첫째, 초기화된 메타데이터 데이터베이스(airflow db migrate)가 있어야 한다. 둘째, Airflow가 직렬화할 수 있는 Dag여야 한다 — 즉 Dag가 설정된 Dag 번들(기본값은 Dags 폴더) 안의 파일에 정의되어 있어야 한다. dag.test()는 번들을 다시 파싱해 Dag를 직렬화해준다.

메모리 안에서만 만들어진 Dag(예: pytest 함수 안이나 Dags 폴더 밖의 파일)는 어떤 번들에도 속하지 않으므로 아무것도 직렬화되지 않고, `Cannot create DagRun for DAG <dag_id> because the dag is not serialized`라는 에러와 함께 호출이 실패한다. Dag run 자체 없이 오퍼레이터 동작만 테스트하려면 오퍼레이터를 직접 호출하는 방식(유닛 테스트)을 쓴다.

설정 방법은 Dag 파일 맨 아래에 두 줄을 추가하는 것뿐이다: `if __name__ == "__main__": dag.test()`. 세부 튜닝을 위한 선택 인자로는 execution_date(특정 날짜의 Dag run을 테스트), use_executor(기본적으로 dag.test는 실행기 없이 모든 태스크를 로컬에서 실행하지만, 이 인자를 주면 Airflow 환경에 설정된 실행기를 통해 Dag를 실행한다)가 있다.

특정 태스크 건너뛰기 (mark_success_pattern)

로컬 환경에서 일부 태스크(의존성 체크용 센서나 정리 단계 등)를 실행하고 싶지 않다면, mark_success_pattern 인자에 해당 task_id에 매칭되는 패턴을 넘겨 자동으로 성공 처리할 수 있다. 예시 Dag는 두 개의 ExternalTaskSensor로 상류 Dag 완료를 기다리고, PythonOperator로 통계를 추출한 뒤 cleanup 태스크로 중간 산출물을 지운다. 테스트 시에는 ingest_testing_data()로 테스트 데이터를 수동으로 채워 넣고, `dag.test(mark_success_pattern="wait_for_.*|cleanup")`로 두 센서와 cleanup 단계를 건너뛰게 해, 중간 산출물인 csv를 검사용으로 남겨둘 수 있다.

커맨드라인 디버깅 (pdb)

같은 두 줄(`if __name__ == "__main__": dag.test()`)을 추가해두면 `python -m pdb <Dag파일>.py`로 커맨드라인에서 대화형 디버깅도 할 수 있다. 특정 줄에 브레이크포인트를 걸고(b 45) 실행을 계속하면(c) 그 지점에서 멈춰 태스크 객체 등을 조사할 수 있다. IDE에서 디버깅하려면 같은 main 블록을 Dag 파일 끝에 추가한 뒤 그 파일을 실행/디버그하면 된다.