스케줄링 (Scheduling)

Airflow에서 가장 자주 마주치는 질문 중 하나가 "이 DAG는 언제 실행되나요?"예요. 답은 DAG를 정의할 때 넘기는 schedule 인자 하나로 결정됩니다. 이 값에 따라 에어플로우가 언제, 어떤 순서로 DAG Run을 만들어 실행할지가 정해지죠.

schedule 인자로 받을 수 있는 값은 크게 네 종류예요.

cron 표현식 — "0 0 * * *"처럼 분/시/일/월/요일을 지정하는 방식
cron 프리셋 — @daily, @hourly처럼 미리 정의된 약어
datetime.timedelta 객체 — datetime.timedelta(days=1)처럼 일정 간격마다 실행
Timetable — 위 방식들로는 표현하기 어려운 복잡한 일정을 위한 사용자 정의 클래스

from airflow.sdk import DAG

import datetime

dag = DAG("regular_interval_cron_example", schedule="0 0 * * *", ...)

dag = DAG("regular_interval_cron_preset_example", schedule="@daily", ...)

dag = DAG("regular_interval_timedelta_example", schedule=datetime.timedelta(days=1), ...)

cron 프리셋 알아두기

가장 간단하게 시작하는 방법은 cron 프리셋을 쓰는 거예요. 아래 표는 에어플로우가 기본으로 제공하는 프리셋과 그 의미, 그리고 실제로 변환되는 cron 표현식이에요.

프리셋	의미	cron
`None`	스케줄하지 않음, "외부 트리거" 전용 DAG에 사용
`@once`	정확히 한 번만 실행
`@continuous`	이전 실행이 끝나는 즉시 실행
`@hourly`	매시 정각(시의 끝)에 한 번 실행	`0 * * * *`
`@daily`	매일 자정(24:00)에 한 번 실행	`0 0 * * *`
`@weekly`	매주 일요일 자정(24:00)에 한 번 실행	`0 0 * * 0`
`@monthly`	매월 1일 자정(24:00)에 한 번 실행	`0 0 1 * *`
`@quarterly`	분기마다 첫날 자정(24:00)에 한 번 실행	`0 0 1 /3 `
`@yearly`	매년 1월 1일 자정(24:00)에 한 번 실행	`0 0 1 1 *`

스케줄을 정하면 에어플로우는 그 일정에 맞춰 DAG를 인스턴스화하고, 데이터베이스에 대응하는 DAG Run 항목을 만듭니다.

cron 표현식 깊게 보기

더 세밀하게 조정해야 한다면 cron 표현식을 직접 쓰면 돼요. 이때 에어플로우는 cron 표현식을 croniter 라이브러리로 파싱하는데, 이 라이브러리는 기존 cron 문법에 확장 표기법을 추가 지원합니다. 예를 들어 매월 첫 번째 월요일 자정에 실행하고 싶다면 0 0 * * MON#1처럼 쓸 수 있어요.

기본 cron 문법이 헷갈리면 Crontab guru 같은 온라인 편집기로 확인하면서 작성하면 편합니다.

데이터 구간(Data Interval) — 언제 실행되는가

스케줄을 이해할 때 가장 중요한 개념이 **데이터 구간(data interval)**이에요. 각 DAG Run에는 자신이 처리할 시간 범위가 할당되는데, 이것이 바로 데이터 구간입니다.

예를 들어 @daily로 스케줄된 DAG는 각 실행의 데이터 구간이 매일 00:00에 시작해 24:00(다음날 자정)에 끝나요.

여기서 중요한 포인트가 하나 있어요. DAG Run은 보통 자신의 데이터 구간이 끝난 뒤에 실행되도록 예약됩니다. 구간이 끝나야 그 시간대의 데이터를 빠짐없이 수집할 수 있기 때문이에요.

즉, 2020-01-01 하루치를 다루는 실행은 2020-01-01이 끝난 다음인 2020-01-02 00:00:00 이후에 시작합니다. 같은 논리로, 데이터 구간의 "시작"을 가리키는 **논리 날짜(logical date)**는 실제로 DAG가 실행된 시각이 아니라 데이터 구간의 시작 시각을 뜻해요. (Airflow 2.2 이전 버전에서는 이를 execution_date라고 불렀습니다.)

start_date도 마찬가지예요. DAG와 태스크의 start_date는 첫 데이터 구간의 시작을 표시할 뿐, 태스크가 실제로 실행되기 시작하는 시각이 아닙니다. 다시 말해, DAG Run은 start_date 이후 한 구간이 지난 뒤에야 처음 예약됩니다.

Catchup — 지나간 구간을 되짚어 실행하기

start_date(선택적으로 end_date)가 있고 cron/프리셋 기반 일정을 가진 DAG는 여러 개의 구간을 정의합니다. 스케줄러는 이 구간들을 개별 DAG Run으로 바꿔 실행하죠.

기본적으로 에어플로우는 DAG가 활성화될 때 아직 실행되지 않은 과거 구간을 자동으로 생성하지 않아요. 설정 scheduler.catchup_by_default는 기본값이 False라서, 스케줄러는 가장 최신 구간에 대한 DAG Run 하나만 만듭니다.

만약 DAG에서 catchup=True로 설정하면, 아직 실행되지 않았거나(또는 클리어된) 모든 데이터 구간에 대해 DAG Run을 하나씩 만들어 실행합니다. 이 개념을 Catchup이라고 불러요.

from airflow.sdk import DAG
from airflow.providers.standard.operators.bash import BashOperator

import datetime
import pendulum

dag = DAG(
    "tutorial",
    default_args={
        "depends_on_past": True,
        "retries": 1,
        "retry_delay": datetime.timedelta(minutes=3),
    },
    start_date=pendulum.datetime(2015, 12, 1, tz="UTC"),
    description="A simple tutorial Dag",
    schedule="@daily",
)

위 예시에서 스케줄러 데몬이 2016-01-02 오전 6시에 이 DAG를 발견했다면(또는 커맨드라인에서 실행했다면) 하나의 DAG Run이 만들어져요. 그 구간은 2016-01-01에서 2016-01-02 사이에 해당하고, 다음 실행은 2016-01-03 아침 자정 직후에 만들어지며 2016-01-02~2016-01-03 구간을 다룹니다.

만약 dag.catchup이 True였다면, 스케줄러는 2015-12-01부터 2016-01-02 사이의 완료된 각 구간에 대해 DAG Run을 만들고(2016-01-02 구간은 아직 끝나지 않았으므로 제외) 순차적으로 실행합니다.

Catchup은 DAG를 일정 기간 껐다가 다시 켜는 경우에도 동작해요.

DAG가 구간으로 쉽게 나눌 수 있는 원자적인(atomic) 작업이라면 catchup을 켜는 게 좋아요.
반대로 DAG가 내부적으로 스스로 지난 데이터를 처리하도록 짜여 있다면(예: 구간이 아니라 "지금(Now)" 기준으로 동작한다면) catchup을 꺼두는 게 맞아요. 기본값이 꺼짐이지만, 환경 설정이 바뀌어 있다면 DAG 정의에서 catchup=False로 명시하면 확실합니다.

주의할 점이 하나 있어요. schedule에 datetime.timedelta 객체를 쓰면 동작이 달라질 수 있어요. 이 경우 만들어진 DAG Run 하나는 2016-01-01 06:00부터 2016-01-02 06:00까지(지금에 끝나는 한 구간)를 다룹니다. cron 기반 스케줄과 delta 기반 스케줄의 차이에 대한 자세한 설명은 티머테이블(timetables) 비교 문서에서 확인할 수 있어요.

Timetable — 복잡한 일정을 위한 확장

cron 표현식이나 timedelta로는 표현할 수 없는 더 복잡한 일정이 필요하다면 Timetable을 쓰면 됩니다. 사실 에어플로우는 cron이나 timedelta로 스케줄된 DAG도 내부적으로 항상 timetable로 변환해서 처리해요. 즉, 위 방식들만으로 충분하다면 사용자 정의 timetable을 신경 쓸 필요가 없어요. 다만 복잡한 요구가 생기면 직접 timetable 클래스를 만들어서 DAG의 schedule 인자에 넘기면 됩니다.

몇 가지 대표적인 timetable을 살펴볼게요.

CronTriggerTimetable — cron 표현식을 받아서 그에 따라 DAG Run을 트리거하는 timetable이에요.

from airflow.timetables.trigger import CronTriggerTimetable

@dag(schedule=CronTriggerTimetable("0 1 * * 3", timezone="UTC"), ...)
# 매주 수요일 01:00에 실행
def example_dag():
    pass

여기에 선택적으로 interval 인자를 주면 데이터 구간을 고정할 수도 있어요. 구간은 지정된 시간만큼 길어지고 트리거 시각에 끝납니다.

from datetime import timedelta
from airflow.timetables.trigger import CronTriggerTimetable

@dag(
    # 매주 금요일 18:00에 실행, 근무 주(월요일 9:00 ~ 금요일 18:00)를 다룸
    schedule=CronTriggerTimetable("0 18 * * 5", timezone="UTC", interval=timedelta(days=4, hours=9)),
    ...
)
def example_dag():
    pass

MultipleCronTriggerTimetable — 여러 cron 표현식을 지정해 하루 중 여러 시각에 실행할 수 있는 timetable이에요.

from airflow.timetables.trigger import MultipleCronTriggerTimetable

# 매일 01:10와 02:40에 실행
@dag(schedule=MultipleCronTriggerTimetable("10 1 * * *", "40 2 * * *", timezone="UTC"), ...)
def example_dag():
    pass

여기에도 CronTriggerTimetable과 같은 선택적 interval 인자를 쓸 수 있어요.

from datetime import timedelta
from airflow.timetables.trigger import MultipleCronTriggerTimetable

@dag(
    schedule=MultipleCronTriggerTimetable("10 1 * * *", "40 2 * * *", timezone="UTC", interval=timedelta(hours=1)),
    ...
)
def example_dag():
    pass

정리

Airflow 스케줄링의 핵심을 한 번에 요약하면 이래요.

실행 시점은 schedule 인자로 정하고, cron 표현식 / 프리셋 / timedelta / timetable 중에 고를 수 있어요.
DAG Run은 자신이 처리할 데이터 구간을 가지며, 보통 구간이 끝난 뒤에 실행돼요.
logical_date(구버전 execution_date)는 실행 시각이 아니라 데이터 구간의 시작을 가리킵니다.
catchup=True로 설정하면 지나간 구간의 실행을 자동으로 채워 넣을 수 있어요.
cron이나 timedelta로 부족한 복잡한 일정은 timetable로 직접 만들 수 있어요.