Airflow 로 일정 돌리기

아파트 관리사무소에는 "화요일 분리수거, 매달 1일 소독" 같은 일정표가 있습니다. 관리인은 일정표를 보고 때가 되면 일을 나눠 맡기고, 끝나면 장부에 적습니다. 비가 와서 못 했으면 다음 날 다시 합니다. Airflow 가 이 관리사무소입니다.

1번 스케줄러는 일정표(DAG 의 schedule)를 보다가 때가 되면 2번 DAG 실행 하나를 만듭니다. 3번 작업자(worker)가 작업을 차례로 돌리고, 실패하면 정한 횟수만큼 다시 시도합니다. 4번 웹 화면에는 날짜마다 작업의 성공과 실패가 칸으로 남습니다. 일정표가 DAG 파일, 관리인이 스케줄러, 일하는 사람이 작업자, 장부가 웹 화면입니다.

Airflow 의 낱말

DAG 파일의 꼴

아래는 이 PC 에서 실행하지 않은 안내 코드입니다. Airflow 3 의 TaskFlow 꼴로 이 장의 색인 갱신을 적었습니다. Airflow 는 Windows 에서 바로 돌지 않아 Linux, WSL2, Docker 에서 띄웁니다.

from datetime import timedelta
import pendulum
from airflow.sdk import dag, task


@dag(schedule="0 2 * * *",                                   # 매일 새벽 2시
     start_date=pendulum.datetime(2026, 11, 1, tz="Asia/Seoul"),
     catchup=False,                                         # 지난날을 거슬러 돌리지 않는다
     default_args={"retries": 2, "retry_delay": timedelta(minutes=5)})
def library_refresh():
    @task
    def read_export(ds=None):                               # ds: 데이터 구간의 날짜(2026-11-03 꼴)
        return f"/exports/library/{ds}"

    @task.short_circuit                                     # False 를 돌려주면 뒤 작업을 건너뛴다
    def check(path):
        return validate(path)["통과"]

    @task
    def build_index(path):
        return refresh(path)                                # 실습 3의 증분 색인

    path = read_export()
    check(path) >> build_index(path)


library_refresh()

하루치를 시험으로 돌리는 명령과 예상 출력(실행하지 않고 유추한 예시)입니다.

airflow dags test library_refresh 2026-11-03
... Running task read_export ... Marking task as SUCCESS
... Running task check ... Marking task as SUCCESS
... Running task build_index ... Marking task as SUCCESS
... DagRun Finished: state=success

데이터 구간 셈하기

매일 2시에 도는 DAG 의 실행 하나는 "어제 2시부터 오늘 2시까지"를 맡습니다. catchup=True 로 두면 시작일부터 오늘까지 빠진 구간을 모두 채웁니다.

from datetime import datetime, timedelta

start = datetime(2026, 11, 1, 2, 0)          # DAG 를 켠 날의 첫 시각
now = datetime(2026, 11, 4, 9, 30)           # 지금
# 끝이 지금보다 앞선 하루 구간만 돌릴 수 있다
intervals = []
t = start
while t + timedelta(days=1) <= now:
    intervals.append((t.strftime("%m-%d %H:%M"), (t + timedelta(days=1)).strftime("%m-%d %H:%M")))
    t += timedelta(days=1)
# 결과: catchup=True 면 세 구간을 모두, False 면 마지막 구간만 돌린다
print("catchup=True :", intervals)
print("catchup=False:", intervals[-1:])

코드에 나오는 함수

이 코드가 뜻하는 것

아파트 관리비 고지서를 떠올려 봅니다. 고지서 한 장은 한 달 동안 쓴 몫을 맡고, 그 달이 다 끝나야 나옵니다. 오래 집을 비웠다 돌아오면 밀린 고지서가 여러 장 쌓여 있습니다. 여기서 고지서 한 장이 데이터 구간 하나, 쓴 기간이 구간의 시작과 끝, 밀린 고지서를 모두 처리하는 것이 catchup=True 입니다.

결과에 11월 4일 2시부터 5일 2시까지의 구간이 없는 것은, 그 구간이 지금(4일 9시 30분)까지 끝나지 않았기 때문입니다. 비유와 다른 점은, 관리비는 밀린 것을 건너뛸 수 없지만 DAG 는 catchup=False 로 두면 지난 구간을 정말로 돌리지 않는다는 것입니다.

주의할 점

• 작업 함수는 날짜(ds)를 받아 그 날짜의 데이터만 처리하게 씁니다. 그래야 지난날을 다시 돌려도 같은 결과가 나옵니다.
• 작업 사이에 큰 데이터를 넘기지 않습니다. 파일 경로처럼 작은 값만 넘기고, 데이터는 파일이나 저장소에 둡니다.