Airflow

DeltaDataIntervalTimetable과 CronDataIntervalTimetable

``DeltaDataIntervalTimetable``은 ``timedelta``를 받아 시간 간격으로 data interval을 스케줄한다. 이 타임테이블은 data interval '값' 자체에 집중하며, 하루의 시작이나 시(hour)의 시작 같은 임의의 경계에 실행 시각을 반드시 맞추지는 않는다.

@dag(schedule=DeltaDataIntervalTimetable(timedelta(minutes=30)))
def example_dag(): ...

``CronDataIntervalTimetable``은 cron 표현식을 받아, 연속된 cron 트리거 시점 사이의 간격을 data interval로 만들고 '각 data interval이 끝나는 시점'에 Dag run을 트리거한다.

@dag(schedule=CronDataIntervalTimetable("0 1 * * 3"))  # 매주 수요일 01:00에 그 인터벌의 끝으로 트리거
def example_dag(): ...

두 타임테이블 모두 이름이 비슷한 Trigger 계열 상대(``DeltaTriggerTimetable``, ``CronTriggerTimetable``)와 data interval을 다루는 방식이 근본적으로 다르다는 점이 핵심이며, 이 차이는 다음 모듈(Timetable 비교)에서 자세히 다룬다.

EventsTimetable

``EventsTimetable``은 Dag가 실행되어야 할 ``datetime`` 목록을 그대로 받는다. 스포츠 경기 일정, 계획된 커뮤니케이션 캠페인처럼 불규칙하지만 예측 가능한 일정에 유용하다. 이벤트 목록은 유한하고 적당한 크기여야 하는데, Dag가 파싱될 때마다 이 목록 전체를 로드하기 때문이다.

``restrict_to_events`` 플래그를 켜면 수동 실행(manual run)도 가장 최근 또는 최초 이벤트의 시각을 data interval로 강제로 사용하게 할 수 있다. 이 플래그가 꺼져 있으면(기본값) 수동 실행은 실행을 시작한 시각을 ``data_interval_start``/``data_interval_end``로 그대로 쓴다. ``description`` 파라미터로 이벤트 집합에 이름을 붙이면 Airflow UI에 표시된다.

@dag(
    schedule=EventsTimetable(
        event_dates=[
            pendulum.datetime(2022, 4, 5, 8, 27, tz="America/Chicago"),
            pendulum.datetime(2022, 4, 17, 8, 27, tz="America/Chicago"),
            pendulum.datetime(2022, 4, 22, 20, 50, tz="America/Chicago"),
        ],
        description="My Team's Baseball Games",
        restrict_to_events=False,
    ),
    ...,
)
def example_dag(): ...

AssetOrTimeSchedule

``AssetOrTimeSchedule``은 조건식 Asset 표현식과 시간 기반 스케줄을 결합하는 특수한 타임테이블이다. 전통적인 타임테이블에 따른 정기 실행과, Asset 갱신에 의한 트리거 실행을 각각 독립적으로 만들어낸다. Dag가 Asset 갱신에도 반응하면서 동시에 주기적인 점검이나 갱신도 계속 수행해야 하는 시나리오에 특히 유용하다.

from airflow.timetables.assets import AssetOrTimeSchedule
from airflow.timetables.trigger import CronTriggerTimetable


@dag(
    schedule=AssetOrTimeSchedule(
        timetable=CronTriggerTimetable("0 1 * * 3", timezone="UTC"), assets=(dag1_asset & dag2_asset)
    ),
    ...,
)
def example_dag(): ...

이 예시는 매주 수요일 01:00에 정기 실행되는 것과 별개로, ``dag1_asset``과 ``dag2_asset``이 모두 갱신될 때도 독립적으로 실행된다.