Timetable — DAG 스케줄의 내부 엔진 이해하기

크론 표현식이나 timedelta로 스케줄을 잡으면 충분한 경우가 대부분이지만, "매월 음력 첫날", "일주일치 데이터를 매일 묶어서 처리", "경기 일정에 맞춰 실행"처럼 일반적인 스케줄로는 표현하기 어려운 요구가 생겨요. 이럴 때 Airflow는 스케줄을 결정짓는 내부 엔진인 timetable을 직접 구현해서 schedule 인자에 넘길 수 있어요. timetable이 언제 DAG을 실행할지뿐 아니라 각 실행의 데이터 간격(data interval)과 logical date까지 정해 줘요.

출처: Airflow 공식 문서 — Timetables

본문

시간 기반 스케줄로 도는 DAG이라면 그 내부의 timetable이 스케줄링을 담당해요. 크론 표현식이나 timedelta로 스케줄한 DAG도 내부적으로는 항상 timetable로 변환돼요. 그래서 그 정도 요구만 있다면 커스텀 timetable에 신경 쓸 필요 없이 기본 timetable이 알아서 처리해 줘요. 더 복잡한 요구가 생겼을 때만 직접 timetable 클래스를 만들어 schedule 인자에 넘기면 돼요.

커스텀 timetable이 필요한 전형적인 사례가 몇 가지 있어요.

음력처럼 불규칙한 달력 기반 일정 (예: 음력 매월 특정일)
롤링 윈도우, 즉 겹치는 데이터 간격. 매일 실행하되 이전 7일을 커버하는 형태를 크론으로 흉내 내기 어려울 때 커스텀 데이터 간격이 더 자연스러워요.
사이에 "구멍"이 있는 데이터 간격. 크론과 timedelta는 연속 구간만 표현하거든요.

커스텀 timetable은 플러그인으로 작성해 DAG에 사용할 수 있어요. 일반적인 규칙 하나만 기억하세요. Variables나 Connections처럼 데이터베이스 접근이 필요한 것은 코드에서 가능한 한 늦게 접근하는 게 좋아요.

Airflow가 기본 제공하는 Timetable

자주 쓰이는 사례는 Airflow에 이미 내장되어 있어요. 추가 timetable은 플러그인으로도 공급될 수 있어요.

DeltaTriggerTimetable — datetime.timedelta나 dateutil.relativedelta.relativedelta를 받아 그 시간 간격이 지날 때마다 DAG을 한 번 실행해요.

from datetime import timedelta

from airflow.timetables.trigger import DeltaTriggerTimetable


@dag(schedule=DeltaTriggerTimetable(timedelta(days=7)), ...)  # Once every week.
def example_dag():
    pass

선택 인자인 interval에 정적 데이터 간격을 줄 수도 있어요. 이 인자도 timedelta 또는 relativedelta여야 하고, 이렇게 쓰면 트리거된 실행의 데이터 간격이 지정한 길이만큼이고 트리거 시점에 끝나요.

from datetime import UTC, datetime, timedelta

from dateutil.relativedelta import relativedelta, FR

from airflow.timetables.trigger import DeltaTriggerTimetable


@dag(
    # Runs every Friday at 18:00 to cover the work week.
    schedule=DeltaTriggerTimetable(
        relativedelta(weekday=FR(), hour=18),
        interval=timedelta(days=4, hours=9),
    ),
    start_date=datetime(2025, 1, 3, 18, tzinfo=UTC),
    ...,
)
def example_dag():
    pass

CronTriggerTimetable — 크론 표현식을 받아 그에 맞춰 DAG 실행을 트리거해요.

from airflow.timetables.trigger import CronTriggerTimetable


@dag(schedule=CronTriggerTimetable("0 1 * * 3", timezone="UTC"), ...)  # At 01:00 on Wednesday
def example_dag():
    pass

MultipleCronTriggerTimetable — 여러 크론 표현식을 받는다는 점만 CronTriggerTimetable과 달라요. 어느 표현식이든 시간에 맞으면 DAG 실행이 예약돼요. 하나의 크론 표현식으로는 표현할 수 없는 일정에 특히 유용하죠.

from airflow.timetables.trigger import MultipleCronTriggerTimetable


# At 1:10 and 2:40 each day.
@dag(schedule=MultipleCronTriggerTimetable("10 1 * * *", "40 2 * * *", timezone="UTC"), ...)
def example_dag():
    pass

DeltaDataIntervalTimetable — 시간 간격으로 데이터 간격을 예약해요. DAG의 schedule 파라미터에 DeltaDataIntervalTimetable을 넘겨 선택할 수 있어요. 이 timetable은 데이터 간격 값에 집중해서, 실행 시점을 하루의 시작이나 시간의 시작 같은 임의 경계에 맞추지 않아요.

from datetime import timedelta

from airflow.sdk import dag, DeltaDataIntervalTimetable


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

CronDataIntervalTimetable — 크론 표현식을 받아 각 크론 트리거 지점 사이의 간격대로 데이터 간격을 만들고, 각 데이터 간격이 끝나는 시점에 DAG 실행을 트리거해요.

from airflow.sdk import dag, CronDataIntervalTimetable


@dag(schedule=CronDataIntervalTimetable("0 1 * * 3"))  # At 01:00 on Wednesday.
def example_dag():
    pass

EventsTimetable — datetime 목록을 넘겨 그 시각들을 지나면 DAG을 실행해요. 스포츠 경기 일정, 계획된 캠페인 등 불규칙하지만 예측 가능한 일정에 어울려요. 이벤트 목록은 DAG이 파싱될 때마다 로드되므로 유한하고 합리적인 크기여야 해요. 선택 인자 restrict_to_events를 쓰면 수동 실행에서도 데이터 간격을 가장 최근(또는 가장 처음) 이벤트의 시각으로 강제할 수 있어요. 그렇지 않으면 수동 실행은 시작 시각과 끝 시각이 실행 시작 시점과 같은 데이터 간격으로 시작돼요. description으로 이벤트 묶음에 이름을 붙일 수 있고 UI에 표시돼요.

from airflow.timetables.events import EventsTimetable


@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():
    pass

시간 스케줄에 에셋 이벤트 결합하기

조건부 에셋 표현식과 시간 기반 스케줄을 결합하면 스케줄링 유연성이 크게 올라가요. AssetOrTimeSchedule은 시간 스케줄과 에셋 이벤트를 함께 기반으로 DAG을 예약하는 특화 timetable이에요. 전통적 timetable처럼 예약 실행도 만들고, 에셋 트리거 실행도 독립적으로 만들어요. 에셋이 갱신될 때도 돌아야 하고 주기적인 확인·업데이트도 필요한 워크플로에 딱이에요.

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)
    )
    # Additional arguments here, replace this comment with actual arguments
)
def example_dag():
    # Dag tasks go here
    pass

트리거 timetable과 데이터 간격 timetable의 차이

Airflow는 cron과 delta 일정마다 각각 두 개의 timetable을 제공해요.

CronTriggerTimetable과 CronDataIntervalTimetable은 둘 다 크론 표현식을 받아요.
DeltaTriggerTimetable과 DeltaDataIntervalTimetable은 둘 다 timedelta 또는 relativedelta를 받아요.
트리거 timetable(...TriggerTimetable)은 데이터 간격 개념을 다루지 않는 반면, 데이터 간격 timetable(...DataIntervalTimetable)은 그 개념을 다뤄요.
두 종류의 run_id에 들어가는 타임스탬프와 logical_date는 데이터 간격을 어떻게 처리하는지에 따라 다르게 정의돼요.

데이터 간격을 다루는지의 차이: 트리거 timetable은 데이터 간격 개념이 없어서 data_interval_start와 data_interval_end 값이 같아요. 둘 다 DAG 실행이 트리거된 시각이죠. 데이터 간격 timetable에서는 두 값이 달라요. data_interval_end는 실행이 트리거된 시각이고, data_interval_start는 구간의 시작이에요.

Catchup 동작: 기본적으로 catchup은 False예요. 이 설정이 다음 시나리오에서 불필요한 실행을 막아 줍니다.

시작일이 과거인 새 DAG을 만들었는데 과거의 실행을 원하지 않을 때. catchup이 True면 그 구간 동안 실행됐을 모든 DAG을 돌려요.
실행하던 DAG을 일시 중지했다가 나중에 다시 켰을 때. False라면 중지 기간 동안 실행됐을 DAG을 돌리지 않아요.

이런 시나리오에서 run_id의 logical_date는 timetable이 데이터 간격을 다루는 방식에 따라 결정돼요. 기본 catchup 동작은 [scheduler] catchup_by_default 설정으로 바꿀 수 있어요.

DAG 실행이 트리거되는 시각: 트리거 timetable과 데이터 간격 timetable 모두 같은 시각에 실행을 트리거해요. 다만 run_id의 타임스탬프는 서로 달라요. 그 이유는 run_id가 logical_date에 기반하기 때문이에요. 트리거 timetable은 logical_date를 트리거 시각으로, 데이터 간격 timetable은 data_interval_start로 앵커해요.

예를 들어 DAG을 다시 켠 직후 2월 2일자 실행 하나가 바로 트리거돼요. 이 예시들에서 트리거 timetable이 사람이 워크플로에 기대하는 직관적인 실행을 만들고, 데이터 간격 timetable은 자기가 처리하는 데이터 간격에 맞춰 설계되어 워크플로의 속성을 그대로 반영하지 않는다는 걸 볼 수 있어요.

기존 DAG에서 timetable 종류 전환하기: 두 종류는 logical_date를 다르게 고정하잖아요. 이미 실행 이력이 있는 DAG을 트리거 timetable에서 데이터 간격 timetable로 전환하면 예약 실행 하나가 건너뛰어져요. 다음 실행이 이전 실행의 logical_date와 충돌하지 않도록 한 주기 앞으로 밀리기 때문이에요. 반대 방향(데이터 간격 → 트리거)은 실행을 건너뛰지 않아요.

참고로 [scheduler] create_cron_data_intervals 설정을 뒤집으면 schedule=에 크론 문자열만 쓴 모든 DAG의 timetable 해석이 달라져요. 기본값이 달라지는 버전 경계를 넘을 때 특히 주의하세요. Airflow 3는 기본이 False, Airflow 2.x는 기본이 True예요.

cron과 delta 데이터 간격 timetable의 차이

DeltaDataIntervalTimetable과 CronDataIntervalTimetable 중 무엇을 쓸지는 사용 사례에 달려 있어요. 예를 들어 2월 1일 01:05에 DAG을 켠다면 만들어진 DAG 실행과 데이터 간격이 어떻게 달라지는지 표로 정리돼 있어요. 실행 주기를 데이터 간격과 자연스럽게 맞추고 싶으면 cron 계열, 임의 경계와 무관하게 일정 간격 실행이면 delta 계열을 고르면 돼요.

Timetable — DAG 스케줄의 내부 엔진 이해하기