Sensor와 event-driven orchestration: deferrable operator, dataset scheduling

시간 기반 스케줄에서 이벤트 기반 오케스트레이션으로

Airflow는 오랫동안 "매일 새벽 2시에 실행"처럼 시간 기반 스케줄을 중심으로 설계됐다. 그러나 현실의 데이터 파이프라인은 시계보다 데이터에 반응해야 한다. 업스트림 테이블이 채워졌을 때, S3 파일이 도착했을 때, 외부 API가 완료 상태를 반환했을 때 — 이런 조건을 효율적으로 기다리는 방법이 Sensor와 Deferrable Operator, 그리고 Dataset/Asset 스케줄링이다.

이 편에서는 세 가지 레이어를 순서대로 다룬다.

1. Sensor: 조건을 폴링해 대기하는 기본 메커니즘
2. Deferrable Operator: worker 슬롯을 점유하지 않는 비동기 대기
3. Dataset/Asset 스케줄링: DAG 간 데이터 의존성을 선언적으로 표현

1. Sensor: 조건 대기의 기본 단위

Sensor는 BaseSensorOperator를 상속하며, poke() 메서드가 True를 반환할 때까지 반복 호출되는 특수 Operator다. Airflow에서 많이 쓰이는 내장 Sensor는 아래와 같다.

1.1 ExternalTaskSensor: DAG 간 의존 표현

한 DAG의 완료를 다른 DAG에서 기다릴 때 가장 직접적인 방법이다.

from airflow.sensors.external_task import ExternalTaskSensor
from datetime import timedelta

wait_for_upstream = ExternalTaskSensor(
    task_id="wait_for_upstream",
    external_dag_id="upstream_dag",
    external_task_id="final_task",       # 생략하면 DAG 전체 완료 대기
    allowed_states=["success"],
    failed_states=["failed", "skipped"], # 이 상태면 sensor도 즉시 실패
    execution_delta=timedelta(hours=1),  # 1시간 전 run 기준
    mode="reschedule",
    timeout=3600,
)

execution_delta는 현재 run의 logical date에서 얼마나 과거 run을 바라볼지 결정한다. 복잡한 경우에는 execution_date_fn(callable)을 사용하지만, 두 파라미터는 동시에 지정할 수 없다.

Airflow 2.4+ 이후에는 ExternalTaskSensor보다 Dataset 스케줄링으로 DAG 간 의존을 표현하는 것이 권장된다.

2. Sensor 실행 모드: poke vs reschedule vs deferrable

Sensor 설계에서 가장 중요한 운영 결정은 실행 모드다. 세 가지 모드는 worker 슬롯 점유 방식이 근본적으로 다르다.

2.1 poke 모드

Sensor가 worker 슬롯을 잡은 채 sleep(poke_interval) → poke() → 반복한다. 대기 시간 동안에도 슬롯을 점유하므로, 오래 기다리는 Sensor가 많아지면 슬롯이 모두 차서 다른 task가 실행되지 못하는 슬롯 고갈 데드락이 발생한다.

사용 기준: 대기 시간이 수 분 이내이고, poke_interval이 매우 짧은 경우에만 적합하다.

2.2 reschedule 모드

poke()가 False를 반환하면 슬롯을 즉시 반환하고 now + poke_interval 시점에 재스케줄된다. 대기 구간에는 슬롯이 비어 있어 다른 task가 사용할 수 있다.

from airflow.sensors.filesystem import FileSensor

wait_for_file = FileSensor(
    task_id="wait_for_file",
    filepath="/data/landing/{{ ds }}/input.csv",
    mode="reschedule",       # 핵심 설정
    poke_interval=120,       # 2분마다 체크
    timeout=7200,            # 2시간 초과 시 실패
)

운영 주의: pool 슬롯은 reschedule 상태에서도 여전히 소모된다. task는 "실행 중" 상태이므로 pool 용량 설계 시 대기 중인 Sensor 수를 포함해야 한다.

2.3 주요 공통 파라미터

함정: 기본 timeout이 7일이다. 설정을 빠뜨리면 실수로 방치된 Sensor가 일주일 동안 리소스를 점유한다. 항상 현실적인 timeout을 명시한다.

3. Deferrable Operator: worker 없이 기다리는 방법

3.1 도입 배경

Airflow 2.2.0(2021년 10월, AIP-40)에서 도입됐다. reschedule 모드도 poke 순간마다 worker 슬롯을 잠깐 점유하며, 수백 개의 Sensor가 동시에 재스케줄되면 스케줄러 부하도 증가한다. deferrable operator는 이 문제를 근본적으로 해결한다. 조건 체크 자체를 Triggerer라는 별도 프로세스에 위임하고, worker 슬롯을 완전히 반환한다.

3.2 Triggerer 아키텍처

처리 흐름:

1. Worker에서 execute() 호출 → self.defer(trigger=..., method_name="execute_complete") 호출
2. Airflow가 TaskDeferred 예외를 잡아 Trigger를 메타데이터 DB에 직렬화 저장, worker 슬롯 즉시 반환
3. Triggerer 프로세스가 DB에서 Trigger를 로드해 asyncio 이벤트 루프에서 비동기 실행
4. 조건 충족 시 TriggerEvent 발생, DB 기록
5. Scheduler가 task를 재스케줄, Worker에서 execute_complete(context, event) 호출로 완료

3.3 내장 Sensor에서 deferrable 사용하기

대부분의 공식 Provider Sensor는 deferrable=True 파라미터로 deferrable 모드를 활성화한다.

from airflow.providers.amazon.aws.sensors.s3 import S3KeySensor

wait_for_s3 = S3KeySensor(
    task_id="wait_for_s3",
    bucket_name="my-bucket",
    bucket_key="data/{{ ds }}/file.parquet",
    aws_conn_id="aws_default",
    deferrable=True,         # Triggerer 방식으로 전환
    poke_interval=60,
    timeout=7200,
)

3.4 커스텀 Deferrable Operator 작성

Trigger(Triggerer에서 실행)와 Operator(Worker에서 잠깐 실행)를 함께 작성한다.

import asyncio
from airflow.triggers.base import BaseTrigger, TriggerEvent
from airflow.models import BaseOperator

class MyFileTrigger(BaseTrigger):
    def __init__(self, filepath: str, poll_interval: float = 30.0):
        self.filepath = filepath
        self.poll_interval = poll_interval

    def serialize(self):
        # 메타데이터 DB 직렬화를 위해 반드시 구현
        return (
            "my_package.triggers.MyFileTrigger",
            {"filepath": self.filepath, "poll_interval": self.poll_interval},
        )

    async def run(self):
        import aiofiles.os
        while True:
            if await aiofiles.os.path.exists(self.filepath):
                yield TriggerEvent({"status": "found", "path": self.filepath})
                return
            await asyncio.sleep(self.poll_interval)

class MyDeferrableFileSensor(BaseOperator):
    def __init__(self, filepath: str, **kwargs):
        super().__init__(**kwargs)
        self.filepath = filepath

    def execute(self, context):
        self.defer(
            trigger=MyFileTrigger(filepath=self.filepath),
            method_name="execute_complete",
            timeout=timedelta(hours=2),
        )

    def execute_complete(self, context, event):
        if event["status"] != "found":
            raise AirflowException(f"Unexpected event: {event}")
        return event["path"]

3.5 Triggerer 운영

# Triggerer 프로세스 시작 (airflow.cfg 또는 환경변수로 용량 설정)
airflow triggerer

# 동시 처리 가능한 Trigger 수 조정
# AIRFLOW__TRIGGERER__DEFAULT_CAPACITY=1000

운영 권장사항:

• 프로덕션에서는 Triggerer를 2개 이상 실행해 고가용성 확보 (자동 failover)
• 기본 용량 1,000은 대부분의 환경에 충분; 수천 개 동시 대기가 필요하면 증설
• TriggerEvent 페이로드는 메타데이터 DB를 통해 전달되므로 대용량 데이터를 event에 포함하지 말 것

4. Dataset/Asset 스케줄링: 데이터 의존성 선언

4.1 아이디어: 시간이 아니라 데이터로 트리거

Airflow 2.4.0(2022년 9월, AIP-48)에서 도입됐다. 업스트림 DAG의 task가 성공적으로 데이터를 생성했을 때 다운스트림 DAG를 자동으로 트리거하는 선언적 데이터 의존성 표현 방식이다.

ExternalTaskSensor보다 훨씬 간결하고, 오케스트레이터가 의존 관계를 그래프로 시각화해준다.

4.2 Producer: outlets로 데이터 생성 선언

from airflow.datasets import Dataset   # Airflow 2.x

orders_dataset = Dataset("s3://my-bucket/data/orders/daily/")

with DAG(dag_id="producer_dag", schedule="@daily", ...):

    @task(outlets=[orders_dataset])
    def write_orders():
        # S3에 데이터 쓰기
        # task가 성공하면 orders_dataset update 이벤트 자동 발생
        pass

핵심 동작: task가 성공해야만 dataset update 이벤트가 발생한다. 실패하거나 skipped 되면 이벤트 없음 → 다운스트림 DAG는 트리거되지 않는다.

4.3 Consumer: schedule로 Dataset 구독

with DAG(
    dag_id="consumer_dag",
    schedule=[orders_dataset],   # orders_dataset이 갱신될 때마다 실행
    catchup=False,
    ...
):
    @task
    def aggregate_orders():
        ...

여러 Dataset을 나열하면 모두 갱신된 후에 한 번 실행되는 AND 조건이 된다.

4.4 Dataset/Asset 스케줄링 흐름

4.5 inlets로 이벤트 메타데이터 접근

@task(inlets=[orders_dataset])
def aggregate_with_metadata(*, inlet_events):
    events = inlet_events[orders_dataset]
    last_event = events[-1]
    row_count = last_event.extra.get("row_count", 0)
    ...

# Producer에서 extra 메타데이터 첨부
@task(outlets=[orders_dataset])
def write_orders_with_meta(**context):
    count = write_to_s3()
    context["outlet_events"][orders_dataset].extra = {"row_count": count}

4.6 시간 + Dataset 복합 스케줄

데이터가 오지 않더라도 특정 시각에는 반드시 실행해야 하는 경우 DatasetOrTimeSchedule을 사용한다.

from airflow.timetables.datasets import DatasetOrTimeSchedule
from airflow.timetables.trigger import CronTriggerTimetable

with DAG(
    dag_id="reporting_dag",
    schedule=DatasetOrTimeSchedule(
        timetable=CronTriggerTimetable("0 8 * * *", timezone="Asia/Seoul"),
        datasets=[orders_dataset],
    ),
    catchup=False,
    ...
):
    ...

이렇게 하면 데이터가 도착하면 즉시 실행, 데이터가 없어도 매일 오전 8시에 한 번 실행된다.

5. Airflow 3.0: Dataset → Asset으로 개명

Airflow 3.0(2025년 4월 22일 GA)에서 Dataset이 Asset으로 이름이 바뀌었다. S3 파일 이외에도 ML 모델, 대시보드, 설정 파일 등 다양한 아티팩트를 표현해야 하기 때문이다.

# Airflow 2.x
from airflow.datasets import Dataset
orders_asset = Dataset("s3://bucket/orders/")

# Airflow 3.0+
from airflow.sdk import Asset
orders_asset = Asset("s3://bucket/orders/")

5.1 논리 연산자: AssetAll / AssetAny

Airflow 3.0에서는 AND/OR 조합이 명시적으로 지원된다.

from airflow.sdk import Asset, AssetAll, AssetAny

asset_a = Asset("s3://bucket/data/a/")
asset_b = Asset("s3://bucket/data/b/")

# AND: 둘 다 갱신되어야 실행
with DAG(dag_id="dag_and", schedule=AssetAll(asset_a, asset_b)):
    ...

# OR: 하나라도 갱신되면 실행
with DAG(dag_id="dag_or", schedule=AssetAny(asset_a, asset_b)):
    ...

# 연산자로 표현
with DAG(dag_id="dag_complex", schedule=(asset_a & asset_b) | asset_c):
    ...

5.2 AssetWatcher: 외부 시스템 이벤트 연동

AIP-82에서 도입된 AssetWatcher는 Kafka, SQS, 웹훅 같은 외부 시스템의 이벤트를 Asset 갱신으로 연결한다. Airflow task가 Producer가 될 필요 없이, 외부 이벤트만으로 DAG를 트리거할 수 있다.

from airflow.sdk import Asset
from airflow.assets.watchers import SomeSQSWatcher  # provider 제공

orders_asset = Asset(
    "s3://bucket/orders/",
    watchers=[SomeSQSWatcher(queue_url="https://sqs.../orders-events")],
)

with DAG(dag_id="consumer_dag", schedule=orders_asset):
    ...

6. 패턴 선택 가이드

마무리: 이벤트 기반 오케스트레이션의 방향

세 가지 기법은 진화 방향이 분명하다. Sensor poke 모드는 짧은 대기에만 남기고, 긴 대기는 deferrable로 넘긴다. DAG 간 의존은 ExternalTaskSensor 대신 Dataset/Asset 스케줄링으로 표현해 오케스트레이터가 의존성 그래프를 이해하게 한다. Airflow 3.0의 AssetWatcher는 그 경계를 외부 시스템까지 확장한다.

이 흐름의 핵심은 단 하나다: 파이프라인이 언제 실행될지를 시계가 아닌 데이터 상태가 결정하게 만드는 것.

References

• Apache Airflow Documentation, "Sensors": https://airflow.apache.org/docs/apache-airflow/stable/core-concepts/sensors.html
• Apache Airflow Documentation, "Deferrable Operators & Triggers": https://airflow.apache.org/docs/apache-airflow/stable/authoring-and-scheduling/deferring.html
• Apache Airflow Documentation, "Asset Definitions": https://airflow.apache.org/docs/apache-airflow/stable/authoring-and-scheduling/assets.html
• Apache Airflow Documentation, "Asset-Aware Scheduling": https://airflow.apache.org/docs/apache-airflow/stable/authoring-and-scheduling/asset-scheduling.html
• Apache Airflow Documentation, "Event-driven scheduling": https://airflow.apache.org/docs/apache-airflow/stable/authoring-and-scheduling/event-scheduling.html
• Apache Airflow Blog, "Apache Airflow 3 is Generally Available!": https://airflow.apache.org/blog/airflow-three-point-oh-is-here/
• Apache Airflow Blog, "Airflow 2.4.0: That Data Aware Release": https://airflow.apache.org/blog/airflow-2.4.0/
• AIP-40 Deferrable Operators: https://cwiki.apache.org/confluence/pages/viewpage.action?pageId=177050929
• AIP-48 Data Dependency Management: https://cwiki.apache.org/confluence/display/AIRFLOW/AIP-48+Data+Dependency+Management+and+Data+Driven+Scheduling
• AIP-82 External event driven scheduling: https://cwiki.apache.org/confluence/display/AIRFLOW/AIP-82+External+event+driven+scheduling+in+Airflow
• Astronomer, "Deferrable Operators": https://www.astronomer.io/docs/learn/deferrable-operators
• Astronomer, "Basic asset-based scheduling": https://www.astronomer.io/docs/learn/airflow-datasets