Airflow

AssetWatcher와 외부 이벤트

Airflow 3.0부터 이벤트 기반 스케줄링이 도입되어, 미리 정해진 시간 스케줄이 아니라 외부 이벤트를 기준으로 Dag를 트리거할 수 있게 되었다. 이는 워크플로가 실시간 데이터 변경, 메시지, 시스템 신호에 반응해야 하는 현대적인 데이터 아키텍처에서 특히 유용하다.

이 기능은 Asset을 기반으로 한다: Asset을 통해 외부 이벤트와 Dag 실행 사이의 의존 관계를 세울 수 있다. 여기서 핵심 역할을 하는 것이 ``AssetWatcher`` 클래스다. 이는 메시지 큐 같은 외부 이벤트 소스를 모니터링하다가 관련 이벤트가 발생하면 Asset 갱신을 트리거한다. ``Asset`` 정의의 ``watchers`` 파라미터에 여러 ``AssetWatcher`` 인스턴스를 연결하면, 하나의 Asset이 여러 이벤트 소스에 반응하게 만들 수 있다.

BaseEventTrigger로 호환 트리거 만들기

Airflow의 모든 트리거(``deferring``에서 쓰이는)가 이벤트 기반 스케줄링에 쓰일 수 있는 것은 아니다. ``BaseTrigger``를 상속하는 트리거 전체 중, ``BaseEventTrigger``를 상속하는 부분집합만 호환된다. 이렇게 제한하는 이유는 일부 트리거가 애초에 이벤트 기반 패러다임을 염두에 두고 설계되지 않았고, 그런 트리거를 스케줄링에 쓰면 의도치 않은 결과를 낳을 수 있기 때문이다 — ``BaseEventTrigger``는 스케줄링에 쓰이는 트리거가 외부 이벤트 변화에 적절히 반응하고 예기치 않은 Dag 동작을 유발하지 않도록 보장하는 역할을 한다.

트리거를 이벤트 기반 스케줄링과 호환되게 만드는 데는 세 가지 시나리오가 있다. (1) 지원되지 않는 이벤트 소스를 위한 새 트리거가 필요하면, ``BaseEventTrigger``를 상속하는 새 클래스를 만들고 로직을 구현한다. (2) 기존의 ``BaseTrigger`` 서브클래스가 이미 이벤트 기반 스케줄링과 호환됨이 검증되었다면, 베이스 클래스를 ``BaseTrigger``에서 ``BaseEventTrigger``로 바꾸기만 하면 된다. (3) 기존 트리거가 호환되지 않아 보인다면, ``BaseEventTrigger``를 상속하는 새 트리거를 만들어야 한다(공통 코드가 있다면 기존 트리거를 상속할 수도 있다).

무한 재스케줄링 피하기

일부 트리거가 이벤트 기반 스케줄링과 호환되지 않는 이유는, 그런 트리거들이 외부 리소스가 '특정 상태에 도달하기'를 기다리기 때문이다. 예: 스토리지 서비스에 파일이 존재하기를 기다림, 어떤 작업이 성공 상태가 되기를 기다림, DB에 특정 행이 존재하기를 기다림. 이런 조건 위에서 스케줄링하면 무한 재스케줄링으로 이어질 수 있다 — 조건이 한 번 참이 되면 대개 그 이후로도 계속 참인 채로 남기 때문이다.

예를 들어 특정 작업이 'success' 상태가 되면 트리거되는 Dag를 생각해보자. 그 작업이 한 번 성공하면 보통 그 상태가 계속 유지되므로, 트리거러가 조건을 확인할 때마다 매번 Dag가 다시 트리거된다. ``S3KeyTrigger``도 비슷한 예다 — S3 버킷에 특정 파일이 있는지 확인하는데, 파일이 한 번 생성되면 '파일 X가 존재하는가'라는 조건은 계속 참으로 남아 트리거 메커니즘이 동작할 때마다 Dag가 끝없이 트리거된다.

커스텀 트리거를 만들 때는 한 번 충족되면 영구적으로 참으로 남는 조건을 쓰지 않도록 주의해야 한다. 그렇지 않으면 의도치 않게 Dag가 무한히 실행되어 시스템에 과부하를 줄 수 있다.

이벤트 기반 Dag의 활용 사례

이벤트 기반 스케줄링이 특히 유용한 대표적인 활용 사례는 다음과 같다.

* 데이터 수집 파이프라인 — 스토리지 시스템에 새 데이터가 도착하면 ETL 워크플로를 트리거. * 머신러닝 워크플로 — 새 데이터셋이 준비되면 모델 학습을 시작. * IoT와 실시간 분석 — 센서 데이터, 로그, 애플리케이션 이벤트에 실시간으로 반응. * 마이크로서비스/이벤트 기반 아키텍처 — 서비스 간 메시지를 기준으로 워크플로를 오케스트레이션.