Scheduling

상위: Airflow

요약

Airflow 스케줄링은 DAG를 자동으로 실행하는 시간을 제어합니다. Cron 표현식, 예약어(@daily), timedelta로 실행 주기를 설정할 수 있습니다. start_date, execution_date, schedule_interval의 관계를 이해하는 것이 중요하며, catchup과 backfill로 과거 데이터를 처리할 수 있습니다.

DAG 스케줄 개념

Airflow에서 DAG 실행을 이해하려면 아래 시간 개념을 명확히 이해해야 합니다:

핵심 개념

• start_date: DAG 시작 기준
• execution_date: 처리할 데이터 기준 시간
• schedule_interval: 실행 간격
• data_interval: 처리 간격

중요: DAG는 항상 이전 주기의 데이터를 처리함

Start Date

정의

• DAG의 최초 실행 시간을 지정
• 예: datetime(2025, 3, 1, 0, 0, 0) → 2025년 3월 1일부터 실행 시작

주의사항

• ❗ 미래 시점으로 설정하면 실행되지 않을 수 있음
• 현재 시점을 기준으로 실행해야 할 과거 스케줄이 있으면 실행됨
• 과거 시점 설정 시 catchup 설정 고려 필요

예제

from datetime import datetime

dag = DAG(
    dag_id="example_dag",
    start_date=datetime(2025, 1, 1),  # 과거 시점
    schedule_interval='@daily',
    catchup=False
)

Execution Date

정의

• DAG가 실제 실행되는 시간이 아닌, 처리 대상 데이터 기준 시간
• 예: execution_date = 2025-03-01 → DAG는 2025-03-02에 실행됨

실행 시간 관계

Schedule_interval = "@daily"인 경우:

항상 Execution Date보다 늦게 실행됨

활용

• Backfill 작업에 활용
• 데이터 파티션 기준으로 사용
• 멱등성 보장에 활용

Schedule Interval

정의

• DAG 실행 주기를 결정하는 값
• Cron 표현식 또는 예약어 사용 가능
• DAG 실행 간격 = Execution Date 결정

예약어

예제

# 예약어 사용
dag = DAG(
    dag_id="daily_dag",
    start_date=datetime(2025, 1, 1),
    schedule_interval='@daily'
)

# 수동 실행만
dag = DAG(
    dag_id="manual_dag",
    start_date=datetime(2025, 1, 1),
    schedule_interval=None
)

Cron 표현식

구조

{분} {시} {일} {월} {요일}

예제

30 5 * * *    → 매일 05:30
0 * * * *     → 매시 정각
*/5 * * * *   → 5분마다
0 9 * * 1     → 매주 월요일 09:00
0 0 1 * *     → 매월 1일 00:00
30 23 L * *   → 매월 마지막 날 23:30

Cron 필드 설명

특수 문자

• *: 모든 값
• ,: 값 목록 (예: 1,3,5)
• -: 값 범위 (예: 1-5)
• /: 증분 (예: */5 = 5마다)
• L: 마지막 (Last)
• W: 평일 (Weekday)

Cron DAG 예제

# 매일 12시 실행
dag = DAG(
    dag_id="noon_dag",
    start_date=datetime(2025, 1, 1),
    schedule_interval="0 12 * * *"
)

# 평일 오전 9시 실행
dag = DAG(
    dag_id="weekday_morning",
    start_date=datetime(2025, 1, 1),
    schedule_interval="0 9 * * 1-5"
)

timedelta 사용

정의

• Cron 표현식으로는 "3일에 한 번" 같은 표현 불가
• schedule_interval=timedelta(days=3)로 설정

예제

from datetime import timedelta

# 3일마다 실행
dag = DAG(
    dag_id="run_every_3_days",
    start_date=datetime(2025, 1, 1),
    schedule_interval=timedelta(days=3),
    catchup=False
)

# 6시간마다 실행
dag = DAG(
    dag_id="run_every_6_hours",
    start_date=datetime(2025, 1, 1),
    schedule_interval=timedelta(hours=6)
)

Catchup

정의

• DAG 시작 이후 누락된 실행 주기를 보완하는 기능

설정

예제

# Catchup 활성화 (과거 데이터 처리)
dag = DAG(
    dag_id="historical_dag",
    start_date=datetime(2024, 1, 1),  # 과거 시점
    schedule_interval='@daily',
    catchup=True  # 2024-01-01부터 현재까지 모든 날짜 실행
)

# Catchup 비활성화 (최신 데이터만)
dag = DAG(
    dag_id="latest_only_dag",
    start_date=datetime(2024, 1, 1),
    schedule_interval='@daily',
    catchup=False  # 오늘부터만 실행
)

주의사항

• catchup=True일 때 과거 기간이 길면 많은 DAG Run 생성
• 리소스 고려하여 설정 필요
• 테스트 시 catchup=False 권장

Backfill

정의

• 특정 과거의 execution_date에 대해 DAG를 수동으로 실행
• 누락된 데이터를 복구하는 작업

사용 시기

• DAG 실행이 누락된 경우
• 데이터 오류가 발생했을 경우
• 과거 데이터에 대한 처리 또는 재처리가 필요한 경우

CLI 사용

# 특정 기간 Backfill
airflow dags backfill \
    --start-date 2025-01-01 \
    --end-date 2025-01-31 \
    my_dag_id

# 특정 날짜만 Backfill
airflow dags backfill \
    --start-date 2025-01-15 \
    --end-date 2025-01-15 \
    my_dag_id

Catchup vs Backfill

스케줄링 Best Practices

권장 사항

• start_date는 과거로 설정: 테스트 시 최근 과거 시점
• catchup=False 기본 사용: 의도적인 경우만 True
• 명확한 schedule_interval: 예약어 또는 명확한 cron 표현식
• 타임존 명시: pendulum 라이브러리 사용 권장

예제

import pendulum

dag = DAG(
    dag_id="best_practice_dag",
    start_date=pendulum.datetime(2025, 1, 1, tz="Asia/Seoul"),
    schedule_interval='@daily',
    catchup=False,
    max_active_runs=1,  # 동시 실행 제한
)

주의사항

Execution Date 혼동

• Execution Date ≠ 실행 시간
• 데이터 기준 시간으로 이해

Catchup 활성화 주의

• 과거 기간이 길면 많은 실행 발생
• 리소스 소진 가능

타임존 문제

• 서버 타임존과 설정 타임존 확인
• pendulum 라이브러리로 명확히 지정

다음 단계

• DAG - DAG 기본 개념
• Task - Task 실행 이해
• Introduction - Web UI에서 스케줄 확인

참고 자료

• 스케줄링 공식 문서
• Cron 표현식 생성기
• Backfill 가이드