Airflow

외부 이벤트 기반 스케줄링: Push vs Pull

Dag가 다른 Dag의 Asset 갱신으로 트리거되는 것(내부 이벤트 기반 스케줄링)만으로는 시스템 신호, 메시지, 실시간 데이터 변경 같은 '외부' 이벤트를 다루기 어렵다. Airflow는 외부 이벤트 기반 스케줄링을 위해 두 가지 접근을 지원한다.

Push 기반(REST API, 2.9에서 추가)은 외부 시스템이 REST API로 Asset 이벤트를 직접 밀어넣는 방식이다. 예를 들어 ``waiting_for_asset_1_and_2`` Dag가 ``schedule=[Asset("asset-1"), Asset("asset-2")]``로 정의되어 있을 때, "asset-1"이 먼저 갱신되면 Airflow는 이를 기록해두고(이런 레코드를 'queued asset event'라 부른다) "asset-2"가 갱신되는 순간 Dag를 트리거해야 함을 알 수 있다. 이 레코드들은 ``queuedEvent`` 계열 REST 엔드포인트로 조회·삭제할 수 있다 (Dag별 조회/삭제, Asset별 조회/삭제 등).

Pull 기반(Asset Watcher)은 외부 시스템이 이벤트를 밀어주기를 기다리는 대신 Airflow가 외부 이벤트 소스를 직접 폴링(pull)하는 방식이다. ``AssetWatcher`` 클래스가 외부 소스(큐, 스토리지 등)를 모니터링하다가 관련 이벤트가 발생하면 대응하는 Asset을 갱신하고 Dag 실행을 트리거한다. 무한 재스케줄링을 피하기 위해 ``BaseEventTrigger``를 상속한 트리거만 이 용도로 쓸 수 있다(자세한 내용은 이벤트 기반 스케줄링 모듈에서 다룬다).

조건식 연산자로 복합 Asset 의존성 구성하기

Airflow는 Asset 조건을 조합하는 두 논리 연산자를 지원한다: AND(``&``)는 지정된 모든 Asset이 갱신되어야 트리거되고, OR(``|``)는 지정된 Asset 중 어느 하나만 갱신되어도 트리거된다.

# 둘 다 갱신되어야 트리거
schedule=(dag1_asset & dag2_asset)

# 둘 중 하나만 갱신되어도 트리거
schedule=(dag1_asset | dag2_asset)

# asset 1이 갱신되거나, asset 2와 3이 둘 다 갱신되면 트리거
schedule=(dag1_asset | (dag2_asset & dag3_asset))

이 연산자들을 조합하면 '하나가 갱신되거나 다른 두 개가 모두 갱신될 때' 같은 더 복잡한 조건도 표현할 수 있다.

Asset Alias 기반 스케줄링과 재파싱 타이밍

다운스트림 Dag는 alias 이름을 직접 참조해 스케줄링될 수도 있다(``schedule=AssetAlias("example-alias")``). Alias는 Dag 파싱 시점에 실제 Asset으로 해석(resolve)되므로, ``min_file_process_interval`` 설정값이 크면 alias 해석이 지연될 수 있고, 이를 해결하려면 Dag 파싱을 트리거해야 한다.

with DAG(dag_id="asset-producer"):

    @task(outlets=[Asset("example-alias")])
    def produce_asset_events():
        pass


with DAG(dag_id="asset-alias-producer"):

    @task(outlets=[AssetAlias("example-alias")])
    def produce_asset_events(*, outlet_events):
        outlet_events[AssetAlias("example-alias")].add(Asset("s3://bucket/my-task"))


with DAG(dag_id="asset-consumer", schedule=Asset("s3://bucket/my-task")):
    ...

with DAG(dag_id="asset-alias-consumer", schedule=AssetAlias("example-alias")):
    ...

``asset-alias-producer``가 처음 실행되면 ``AssetAlias("example-alias")``는 ``Asset("s3://bucket/my-task")``로 해석된다. 하지만 ``asset-alias-consumer``는 다음 Dag 재파싱이 일어나야 그 스케줄이 갱신된다. Airflow는 alias가 이전에 의존하지 않던 새로운 Asset으로 해석될 때, 그 alias에 의존하는 Dag들을 재파싱하도록 처리한다 — 그 결과 ``asset-consumer``와 ``asset-alias-consumer`` 모두 ``asset-alias-producer`` 실행 이후 트리거된다.

Asset과 시간 기반 스케줄 결합 (개요)

조건식 Asset 표현식과 시간 기반 스케줄을 함께 쓰고 싶을 때는 ``AssetOrTimeSchedule``을 쓴다. 이는 Dag가 데이터 갱신에도 반응하면서 동시에 정기적으로도 실행되어야 하는 상황(예: Asset이 갱신될 때마다 실행하되, 그것과 무관하게 매일 한 번은 정기 점검을 실행)에 유용하다. 구체적인 문법과 예시는 Timetable을 다루는 모듈의 AssetOrTimeSchedule 절에서 이어서 다룬다.