Airflow

Dag 변경 직후 트리거를 피해야 하는 이유

Dag를 변경한 직후, 또는 Dag 폴더 안의 다른 동반 파일을 변경한 직후 바로 그 Dag를 트리거하는 것은 피해야 한다. 시스템이 변경된 파일을 처리할 충분한 시간을 줘야 한다.

이 과정은 여러 단계를 거친다. 먼저 파일이 스케줄러로 배포되어야 한다(보통 분산 파일시스템이나 Git-Sync를 통해). 그다음 스케줄러가 Python 파일을 파싱해 데이터베이스에 저장해야 한다. 설정, 분산 파일시스템 속도, 파일 개수, Dag 개수, 파일 내 변경 개수, 파일 크기, 스케줄러 개수, CPU 속도에 따라 이 과정은 몇 초에서 몇 분, 극단적인 경우 수십 분까지 걸릴 수 있다. Dag를 트리거하기 전에는 UI에 그 Dag가 나타날 때까지 기다려야 한다.

변경 후 반영까지 지연이 크다면 다음 설정 파라미터들을 조정할 수 있다: :ref:`scheduler_idle_sleep_time<config:scheduler__scheduler_idle_sleep_time>`, :ref:`min_file_process_interval<config:dag_processor__min_file_process_interval>`, :ref:`refresh_interval<config:dag_processor__refresh_interval>`, :ref:`parsing_processes<config:dag_processor__parsing_processes>`, :ref:`file_parsing_sort_mode<config:dag_processor__file_parsing_sort_mode>`.

Watcher 패턴 - trigger rule로 실패를 놓치지 않게 하기

Watcher 패턴은 다른 태스크들의 상태를 "감시"하는 태스크를 Dag에 두는 방식을 말한다. 주된 목적은 다른 어떤 태스크든 실패하면 Dag Run을 실패시키는 것이며, Airflow의 시스템 테스트(여러 태스크로 구성된, 테스트의 각 단계와 비슷한 Dag)에서 이런 필요가 생겨났다.

보통은 어떤 태스크가 실패하면 다른 모든 태스크는 실행되지 않고 Dag Run 전체가 실패 상태가 된다. 그런데 trigger rule을 쓰면 태스크 실행의 정상적인 흐름을 깨뜨릴 수 있어 Dag가 예상과 다른 상태를 나타낼 수 있다. 예를 들어 ``TriggerRule.ALL_DONE``으로 설정된 teardown 태스크(다른 태스크들의 상태와 무관하게 항상 실행되어 자원을 정리하는 역할)를 둔 경우, teardown이 항상 실행되어 성공하면 Dag Run은 teardown의 상태를 따라가 버려 실패 정보를 놓칠 수 있다. 이런 teardown 태스크가 있는 Dag가 실제로 어떤 태스크든 실패하면 함께 실패하길 원한다면 watcher 패턴이 필요하다.

watcher 태스크는 트리거되면 항상 실패하는 태스크지만, 다른 어떤 태스크가 실패했을 때만 트리거되어야 한다. 이를 위해 trigger rule을 ``TriggerRule.ONE_FAILED``로 설정하고, Dag 내 다른 모든 태스크의 다운스트림 태스크로 만들어야 한다. 이렇게 하면 다른 모든 태스크가 통과하면 watcher는 skip되지만, 무언가 실패하면 watcher가 실행되어 실패하고 Dag Run도 함께 실패한다.

trigger rule은 직접 상위(parent) 태스크에만 의존한다는 점에 유의해야 한다 - 예를 들어 ``TriggerRule.ONE_FAILED``는 파라미터화된 태스크의 직접 부모가 아닌 실패(또는 upstream_failed) 태스크는 무시한다.

예시 Dag에서는 ``failing_task``(항상 실패), ``passing_task``(항상 성공), ``teardown``(``ALL_DONE``으로 항상 트리거), ``watcher``(모든 다른 태스크의 다운스트림, ``ONE_FAILED``)가 등장한다. watcher가 없다면 유일하게 실패하는 ``failing_task``가 leaf 태스크가 아니고 ``teardown``이 성공으로 끝나므로 Dag Run 전체는 ``success``가 되어버린다. watcher를 다른 모든 태스크에 개별적으로 의존하게 만들어야 어떤 태스크든 실패 시 Dag Run을 실패시킬 수 있다. 반대로 teardown 태스크가 없다면애초에 ``failing_task``의 ``failed`` 상태가 다운스트림 ``passing_task``로 전파되어 Dag Run이 자연히 ``failed``가 되므로 watcher가 필요 없다.

AirflowClusterPolicySkipDag - 멀티 클러스터에서 특정 Dag 제외하기

.. versionadded:: 2.7

Airflow Dag는 보통 ``git-sync``를 통해 Git 저장소의 특정 브랜치로 배포·갱신된다. 그런데 운영상의 이유로 여러 Airflow 클러스터를 운영해야 할 때, 여러 Git 브랜치를 유지하는 것은 매우 번거롭다. 특히 ``prod``와 ``beta``처럼 분리된 두 브랜치를 적절한 브랜칭 전략으로 주기적으로 동기화하는 데는 어려움이 따른다 - cherry-pick은 Git 저장소를 유지하기에는 너무 번거롭고, hard-reset은 GitOps 방식으로는 권장되지 않는다.

대안으로, 여러 Airflow 클러스터를 동일한 Git 브랜치(예: ``main``)에 연결하고, 서로 다른 환경 변수와 서로 다른(하지만 동일한 ``connection_id``를 갖는) 커넥션 설정으로 이들을 유지하는 방식을 고려할 수 있다. 또한 cluster policy에서 :class:`~airflow.exceptions.AirflowClusterPolicySkipDag` 예외를 발생시켜, 필요하다면 특정 Airflow 배포에서만 특정 Dag를 :class:`~airflow.models.dagbag.DagBag`에 로드하도록 만들 수 있다.

예시로 ``dag_policy`` 함수에서 Dag의 ``tags``에 ``"only_for_beta"``가 포함돼 있으면 ``AirflowClusterPolicySkipDag`` 예외를 발생시켜, 그 Dag가 프로덕션 클러스터에는 로드되지 않도록 스킵할 수 있다.