Deferrable Operator와 Trigger 직접 작성하기
Apache Airflow Official Documentation (in-repo snapshot) — Apache Software Foundation authoring-and-scheduling/deferring.rst - Writing Deferrable Operators, Writing Triggers, Sensitive information in triggers (약 70-183줄)
이 모듈을 다 읽으면
- 커스텀 Deferrable Operator를 작성할 때 지켜야 할 설계 원칙을 나열할 수 있다
- BaseTrigger의 필수 메서드(run/serialize)와 선택 훅(cleanup/on_kill)의 역할을 구분할 수 있다
- Trigger 작성 시 지켜야 할 비동기/멱등성 제약을 설명할 수 있다
커스텀 Deferrable Operator는 트리거로 defer하되 재개 시 상태가 자동으로 유지되지 않는다는 점을 감안해 설계해야 하고, Trigger는 BaseTrigger를 상속해 비동기 run()과 serialize()를 구현하며 여러 곳에서 중복 실행될 수 있음을 전제로 부작용 없는 코드로 작성해야 한다.
Deferrable Operator 작성 원칙
Operator는 반드시 트리거와 함께 defer해야 하며, 이때 코어 Airflow에 포함된 트리거를 쓰거나 직접 커스텀 트리거를 작성할 수 있다. Operator가 defer되면 워커에서 완전히 내려가며, 로컬 변수나 `self`에 저장한 속성 등 상태가 자동으로 보존되지 않는다 - 상태를 다음 실행으로 넘기려면 `method_name`과 `kwargs`를 통해 명시적으로 전달해야 한다.
한 오퍼레이터는 여러 번 defer할 수 있고, 의미 있는 작업 전이든 후든 defer할 수 있으며, 특정 조건이 충족될 때만 조건부로 defer할 수도 있다 - defer 여부와 시점은 전적으로 오퍼레이터 작성자의 통제 아래 있다. 어떤 오퍼레이터든 클래스에 특별한 표시를 하지 않아도 defer할 수 있으며, 센서에 국한되지 않는다.
Deferrable/비-Deferrable 두 모드를 모두 지원하는 오퍼레이터나 센서를 만들고 싶다면, `__init__`에 `deferrable: bool = conf.getboolean("operators", "default_deferrable", fallback=False)` 패턴을 두어 `[operators] default_deferrable` 설정으로 전체 오퍼레이터의 기본값을 한 곳에서 통제할 수 있게 만드는 것이 권장된다.
핵심 포인트
- Operator가 defer되면 워커에서 완전히 내려가며 로컬 변수나 self 속성 등 상태가 자동으로 보존되지 않는다 - 상태를 넘기려면 method_name과 kwargs를 명시적으로 사용해야 한다
- 어떤 Operator든 특별한 표시 없이 defer할 수 있으며, 센서에 국한되지 않고 여러 번 defer하거나 조건부로 defer할 수도 있다
- 두 모드를 모두 지원하는 오퍼레이터는 `deferrable: bool = conf.getboolean("operators", "default_deferrable", fallback=False)` 패턴으로 `[operators] default_deferrable` 설정을 통해 전역 기본값을 통제하게 만드는 것이 권장된다
Trigger 클래스 작성하기 (BaseTrigger)
Trigger는 `BaseTrigger`를 상속하는 클래스로 작성하며 세 메서드를 구현한다. `__init__`은 오퍼레이터로부터 인자를 받는 메서드로, 이 안의 모든 인자는 직렬화 가능해야 한다 (2.10.0부터는 미리 정의된 트리거로 태스크 실행을 곧바로 시작하는 것도 가능해졌다). `run`은 하나 이상의 `TriggerEvent`를 비동기 제너레이터로 yield하는 비동기 메서드다. `serialize`는 이 트리거를 재구성하는 데 필요한 정보를 classpath와 `__init__`에 넘길 키워드 인자의 튜플로 반환한다.
선택적인 두 생명주기 훅도 있다. `cleanup`은 `run`이 어떤 이유(성공, 타임아웃, 트리거러 종료, 사용자 kill)로든 끝난 뒤 항상 호출되며, 트리거 인스턴스가 붙잡고 있던 로컬 리소스(열린 연결, 임시 파일 등)를 해제하는 데 쓴다. `on_kill`은 사용자가 명시적으로 태스크를 kill(마크 실패, clear, 마크 성공)했을 때만 호출된다 - BigQuery job이나 Databricks run처럼 계속 남겨두고 싶지 않은 외부 작업을 취소하는 데 쓴다. `cleanup`과 달리 `on_kill`은 트리거러 재시작이나 재배치 시에는 호출되지 않으므로, 롤링 배포 중 진행 중인 작업을 실수로 취소할 위험 없이 외부 취소 로직을 안전하게 넣을 수 있다.
핵심 포인트
- `run`은 반드시 `async def`로 작성해 asyncio 기반 비동기 대기(`asyncio.sleep` 등)를 써야 하며, 블로킹 `time.sleep`을 쓰면 트리거러 프로세스 전체가 멈출 수 있다
- `cleanup`은 run이 어떤 이유로든 끝난 뒤 항상 호출되고, `on_kill`은 사용자가 명시적으로 태스크를 kill했을 때만 호출된다 - 트리거러 재시작이나 재배치 때는 on_kill이 불리지 않으므로, 외부 작업(BigQuery job 취소 등) 취소 로직을 on_kill에 안전하게 둘 수 있다
- `__init__`과 `serialize`는 쌍으로 동작한다 - 트리거는 defer 요청 시 한 번 인스턴스화된 뒤 직렬화되어, 실제로 실행할 트리거러 프로세스에서 재인스턴스화된다
Trigger 작성 시 설계 제약
`run`은 반드시 비동기여야 하고, 블로킹 연산을 수행할 때마다 올바르게 `await`해야 한다. `run`은 TriggerEvent를 return이 아니라 반드시 yield해야 한다 - 이벤트를 하나도 yield하지 않고 반환하면 Airflow는 이를 에러로 간주해 이를 기다리던 태스크 인스턴스들을 실패시키며, 예외가 발생해도 마찬가지로 의존 태스크들이 실패한다.
트리거 인스턴스는 한 곳 이상에서 실행될 수 있다고 전제해야 한다 - 네트워크 파티션으로 Airflow가 분리된 머신에서 트리거를 재실행할 수 있기 때문이다. 따라서 부작용에 신중해야 하며, 예를 들어 트리거 안에서 DB에 row를 insert하는 식의 로직은 피해야 한다. 여러 이벤트를 발생시키도록 설계한 트리거라면(현재는 지원되지 않지만) 각 이벤트의 payload가 여러 곳에서 실행 중일 때 중복을 제거할 수 있는 정보를 담아야 한다 - 하나의 이벤트만 발생시키고 오퍼레이터에 정보를 되돌려줄 필요가 없다면 payload를 그냥 `None`으로 둬도 된다.
트리거는 어느 한 트리거러 서비스에서 갑자기 제거되어 새 트리거러에서 다시 시작될 수 있다 (서브넷 변경으로 네트워크 파티션이 발생하거나 배포가 있을 때 등). 이런 경우를 위해 `cleanup`을 구현할 수 있고, 이는 트리거가 정상적으로 끝나든 아니든 항상 `run` 이후에 호출된다. 트리거에 변경이 생기면 반드시 트리거러를 재시작해야 반영되고, 트리거 코드는 Dag 번들이 아니라 `sys.path`의 다른 위치에 있어야 한다 (트리거러는 트리거 실행 시 번들을 초기화하지 않는다).
비동기 파이썬 작성이 익숙하지 않다면 `run()` 작성 시 특히 주의해야 한다 - 블로킹 연산에서 올바르게 `await`하지 않으면 프로세스 전체가 막힐 수 있다. `PYTHONASYNCIODEBUG=1` 환경변수로 추가 검사를 켤 수 있고, 네트워크 기반 파일시스템 호출처럼 겉보기에 평범한 코드도 블로킹일 수 있으므로 특히 조심해야 한다.
트리거에 전달하는 kwargs는 Airflow 2.9.0부터 DB에 암호화되어 저장되고, 읽을 때 복호화된다.
핵심 포인트
- 같은 트리거 인스턴스가 네트워크 파티션 등으로 두 곳 이상에서 동시에 실행될 수 있다는 전제하에 작성해야 하며, DB row insert처럼 부작용이 있는 로직은 피해야 한다
- 트리거 코드가 바뀌면 트리거러를 재시작해야 반영되고, 트리거는 Dag 번들이 아니라 sys.path의 다른 위치에 있어야 한다 (트리거러는 번들을 초기화하지 않는다)
- 트리거에 전달하는 kwargs는 Airflow 2.9.0부터 DB에 암호화되어 저장된다