Airflow

동적 Dag와 시각화(TaskGroup, Edge Label)

Dag는 Python 코드로 정의되므로 순수하게 선언적일 필요가 없고, 반복문이나 함수를 자유롭게 써서 정의할 수 있다. 다만 일반적으로는 Dag 태스크의 토폴로지(레이아웃)는 비교적 안정적으로 유지하고, 동적 Dag는 설정 옵션을 동적으로 불러오거나 오퍼레이터 옵션을 바꾸는 용도로 쓰는 것이 권장된다.

Dag를 시각적으로 보려면 Airflow UI에서 Dag로 이동해 Graph를 선택하거나, ``airflow dags show`` 명령으로 이미지 파일을 렌더링할 수 있다. Graph 뷰는 선택한 Dag Run 안의 모든 Task Instance 상태도 함께 보여주므로 일반적으로 권장된다.

TaskGroup은 Graph 뷰에서 태스크를 계층적으로 묶어 시각적 잡음을 줄이는 데 쓰인다. TaskGroup 안의 태스크들은 원래 Dag에 속하며 Dag의 설정과 풀(pool) 설정을 그대로 따른다. ``>>``/``<<``로 TaskGroup 전체에 걸친 의존성을 적용할 수 있다. TaskGroup도 Dag처럼 ``default_args``를 지원하며, TaskGroup 레벨의 default_args가 Dag 레벨의 것을 덮어쓴다. 기본적으로 자식 태스크/TaskGroup의 ID에는 부모 TaskGroup의 group_id가 접두어로 붙어 고유성을 보장하며, ``prefix_group_id=False``로 이를 끌 수 있지만 그러면 ID 유일성을 사용자가 직접 책임져야 한다.

Edge Label은 태스크 사이의 의존성 엣지에 라벨을 붙이는 기능으로, 특히 분기 구간에서 어떤 조건에 따라 어떤 경로가 실행되는지 표시하는 데 유용하다. ``my_task >> Label("조건") >> other_task``처럼 ``>>``/``<<``와 함께 인라인으로 쓰거나, ``set_upstream``/``set_downstream``에 Label 객체를 전달할 수 있다.

문서화와 패키징

Dag, TaskGroup, 태스크 객체에는 웹 UI에 렌더링되는 문서·노트를 추가할 수 있다. ``doc``(모노스페이스), ``doc_json``(json), ``doc_yaml``(yaml), ``doc_md``(마크다운), ``doc_rst``(reStructuredText) 속성이 이런 용도로 쓰이며, Dag와 TaskGroup에는 이 중 ``doc_md``만 해석된다. ``doc_md``는 문자열이거나 ``.md``로 끝나는 마크다운 파일 참조일 수 있고, 펜스드 코드 블록·표 등 마크다운 구문을 지원하며 ``math`` 언어의 코드 블록은 KaTeX로, ``mermaid`` 언어의 코드 블록은 Mermaid 다이어그램으로 렌더링된다. 상대 경로를 지정하면 스케줄러/Dag 파서가 시작된 경로를 기준으로 로드되고, 파일이 없으면 예외 없이 파일명 자체가 텍스트로 쓰인다. 마크다운 파일 내용의 변경은 Dag 파싱 주기 한 번을 거쳐야 반영된다.

간단한 Dag는 보통 하나의 Python 파일에 들어가지만, 복잡한 Dag는 여러 파일로 나뉘고 함께 배포(vendor)해야 할 의존성을 가질 수 있다. Dag 번들 안에 표준 파일시스템 레이아웃으로 두거나, Dag와 그에 딸린 모든 Python 파일을 하나의 zip 파일로 패키징할 수 있다. 패키징된 Dag에는 몇 가지 제약이 있다: 직렬화에 피클링(pickling)을 쓰는 경우 사용할 수 없고, ``libz.so`` 같은 컴파일된 라이브러리는 포함할 수 없으며 순수 Python만 가능하고, ``sys.path``에 삽입되어 Airflow 프로세스 내 다른 코드에서도 import 가능해지므로 패키지 이름이 시스템에 이미 설치된 다른 패키지와 충돌하지 않도록 해야 한다. 복잡한 컴파일 의존성이 많다면 zip 패키징보다는 Python ``virtualenv``와 ``pip`` 설치를 쓰는 편이 낫다.

.airflowignore와 Dag 간 의존성

``.airflowignore`` 파일은 Dag 번들이나 ``PLUGINS_FOLDER`` 안에서 Airflow가 의도적으로 무시해야 할 디렉터리·파일을 지정한다. ``DAG_IGNORE_FILE_SYNTAX`` 설정(Airflow 2.3에서 추가)으로 ``regexp``와 ``glob`` 두 문법을 선택할 수 있으며, Airflow 3 이상의 기본값은 ``glob``이다(이전 버전은 ``regexp``였다). glob 문법에서는 ``*``가 ``/``를 제외한 임의 문자수를 매칭하고, ``?``는 단일 문자를, ``[a-zA-Z]`` 같은 범위 표기를 쓸 수 있으며, ``!``로 패턴을 부정할 수 있고(순서대로 평가되어 이전 패턴을 덮어쓸 수 있음), ``**``는 여러 디렉터리 계층을 가로질러 매칭한다. 패턴 시작이나 중간에 ``/``가 있으면 그 .airflowignore 파일이 있는 디렉터리 기준 상대 경로가 되고, 없으면 그 아래 어느 레벨에서도 매칭될 수 있다. regexp 문법에서는 각 줄이 정규식이며 파일/디렉터리 '이름'(Dag id 아님)에 대해 ``Pattern.search()``로 매칭한다. .airflowignore의 적용 범위는 그 파일이 위치한 디렉터리와 모든 하위 폴더다.

Dag 간 의존성(Airflow 2.1에서 추가)은 태스크 간 의존성과 달리 좀 더 복잡하다. 한 Dag가 다른 Dag에 의존하는 방식은 두 가지다: ``TriggerDagRunOperator``로 다른 Dag를 트리거하거나, ``ExternalTaskSensor``로 다른 Dag의 완료를 기다리는 것이다. 한 Dag가 다른 Dag의 여러 실행(서로 다른 데이터 인터벌)을 기다리거나 트리거할 수도 있어 더 복잡해지며, 이 의존성은 스케줄러가 Dag 직렬화 과정에서 계산한다. 의존성 탐지기(dependency detector)는 설정 가능하므로 ``DependencyDetector``의 기본 로직과 다른 커스텀 로직을 구현할 수도 있다.

Dag의 일시정지, 비활성화, 삭제 — 서로 다른 세 가지 상태

Dag가 '실행되지 않는' 상태에는 여러 층위가 있다. 일시정지(pause)는 UI에서 할 수 있으며, ``DAGS_FOLDER``에 파일이 존재하고 스케줄러가 DB에 저장한 상태에서 사용자가 UI로 비활성화를 선택한 경우다. Pause/Unpause는 UI와 API로 가능하다. 일시정지된 Dag는 스케줄러가 스케줄링하지 않지만 UI에서 수동으로 트리거할 수는 있다. Dag가 일시정지되면 실행 중인 태스크는 완료가 허용되고 모든 다운스트림 태스크는 'Scheduled' 상태에 놓이며, 재개(unpause)되면 이 'scheduled' 태스크들이 Dag 로직에 따라 실행을 시작한다.

비활성화(deactivation, UI의 'Active' 태그와 혼동하지 말 것)는 Dag를 ``DAGS_FOLDER``에서 제거함으로써 이루어진다. 스케줄러가 폴더를 파싱하다가 이전에 봤고 DB에 저장했던 Dag를 더 이상 찾지 못하면 비활성화 상태로 표시한다. 비활성화된 Dag의 메타데이터와 히스토리는 보존되며, 파일이 다시 추가되면 다시 활성화되고 히스토리도 다시 보인다. UI나 API로 직접 activate/deactivate를 할 수는 없고 오직 ``DAGS_FOLDER``에서 파일을 추가/제거하는 방법으로만 가능하다. 비활성화된 Dag는 UI에서 보이지 않지만(과거 실행 기록은 가끔 보일 수 있음), 상세 정보를 보려고 하면 Dag가 없다는 오류가 표시된다.

Dag 메타데이터는 UI나 API로 삭제할 수 있지만, ``DAGS_FOLDER``에 Dag 파일이 여전히 있다면 스케줄러가 다시 파싱하면서 Dag가 재등장하고 히스토리 정보만 사라진다. 따라서 Dag와 그 모든 히스토리 메타데이터를 완전히 삭제하려면 세 단계가 필요하다: 첫째 Dag를 일시정지하고, 둘째 UI나 API로 히스토리 메타데이터를 DB에서 삭제하고, 셋째 ``DAGS_FOLDER``에서 Dag 파일을 삭제해 비활성화될 때까지 기다린다.

자동 일시정지, Deadline Alert, Dag 테스트

Dag는 자동으로 일시정지되도록 설정할 수도 있다(실험적 기능). ``[core] max_consecutive_failed_dag_runs_per_dag`` 설정으로 Dag가 연속으로 N번 실패하면 자동으로 비활성화되도록 할 수 있고, Dag 인자 ``max_consecutive_failed_dag_runs``로 이 설정을 개별 Dag 단위로 오버라이드할 수 있다.

Deadline Alert(Airflow 3.1에서 추가)는 Dag Run에 시간 임계값을 설정하고 초과 시 자동으로 대응하는 기능이다. 고정된 datetime을 기준으로 삼거나, Dag 큐잉 시각·시작 시각 같은 사전 계산된 참조(reference)를 쓰거나, 커스텀 참조를 구현할 수 있다. 임계값을 초과하면 콜백이 트리거되어 알림 등의 조치를 취할 수 있다. 예를 들어 ``DeadlineReference.DAGRUN_QUEUED_AT`` 기준으로 ``interval=timedelta(minutes=30)``을 설정하면, 큐잉된 지 30분이 지나도 끝나지 않은 Dag Run에 대해 이메일 알림 콜백이 실행된다.

Dag 객체는 ``test()`` 메서드를 제공해 문법/기능을 검증할 수 있다. Dag 정의 모듈 끝에 ``if __name__ == "__main__": dag.test()``를 두고 스크립트로 실행하면, LocalExecutor 실행 세션과 유사한 시뮬레이션된 실행 흐름으로 Dag 내용을 실행해볼 수 있다. 실제 Airflow Executor로 테스트하고 싶다면 ``dag.test(use_executor=True)``를 호출하면 현재 적용된 Airflow 설정의 Executor가 사용된다. pytest 스위트 안에서 사용할 때는 Airflow pytest 플러그인의 ``conf_vars`` 픽스처로 설정값을 손쉽게 바꿔가며 테스트할 수 있다.