Airflow 해부 (3) DAG 작성 기초: TaskFlow API와 의존성

Airflow 해부 시리즈의 3편입니다. 전체 목차는 0편에 있습니다.

TaskFlow API 최소 DAG

현재 표준 작성법은 데코레이터 기반의 TaskFlow API입니다. ~/airflow/dags/demo_pipeline.py로 저장합니다.

from datetime import datetime
from airflow.decorators import dag, task

@dag(
    schedule="@daily",
    start_date=datetime(2026, 7, 1),
    catchup=False,
)
def demo_pipeline():
    @task
    def extract() -> dict:
        return {"rows": 1200, "path": "/data/raw/2026-07-21.parquet"}

    @task
    def transform(meta: dict) -> str:
        return meta["path"].replace("raw", "features")

    @task
    def load(feature_path: str) -> None:
        print(f"features at {feature_path}")

    load(transform(extract()))

demo_pipeline()

• @dag 함수 하나가 DAG 하나입니다. 마지막 줄의 호출(demo_pipeline())이 있어야 scheduler가 인식합니다
• @task 함수 하나가 task 하나입니다
• schedule, start_date, catchup은 시간 동작을 결정하는 파라미터로, 4편 전체가 이 세 개의 이야기입니다

의존성은 함수 호출로 만들어진다

load(transform(extract())) 한 줄이 extract, transform, load의 실행 순서를 만듭니다. TaskFlow API에서 task 함수의 반환값을 다른 task 함수에 넘기는 것이 곧 의존성 선언입니다.

주의할 점은 이 함수들이 DAG 파일이 파싱될 때 실제로 실행되는 것이 아니라는 것입니다. 호출은 “의존성 그래프를 그리는” 단계이고, 실제 실행은 스케줄이 됐을 때 executor가 합니다. task 함수 밖에 쓴 코드는 파싱 때마다 실행되므로, 무거운 코드를 두면 안 됩니다.

병렬 분기도 같은 방식입니다.

raw = extract()
stats = compute_stats(raw)      # 두 task가 raw를 받아
features = build_features(raw)  # 병렬로 실행된다
train(stats, features)          # 둘 다 끝나야 실행된다

XCom: task 사이의 값 전달

반환값이 task 사이를 이동하는 통로가 XCom입니다. 반환값은 metadata DB에 저장되고, 받는 task가 DB에서 꺼내 갑니다.

여기서 한계가 나옵니다. XCom은 DB를 거치므로 작은 메타데이터 전용입니다. 경로, 행 수, 지표 같은 것은 XCom으로 넘기고, DataFrame이나 모델 파일 같은 큰 데이터는 스토리지(디스크, S3)에 쓴 뒤 경로만 넘깁니다. 위 예제가 parquet 경로를 주고받는 것이 그 패턴입니다. DataFrame을 직접 반환하는 DAG는 처음에는 돌아가다가 데이터가 커지면 DB와 함께 무너집니다.

실패에 대비하는 기본 설정

운영 DAG에는 재시도와 알림을 기본으로 붙입니다.

from datetime import timedelta

@dag(
    schedule="@daily",
    start_date=datetime(2026, 7, 1),
    catchup=False,
    default_args={
        "retries": 2,
        "retry_delay": timedelta(minutes=5),
        "owner": "hoseung",
    },
)

• retries: task instance가 실패하면 지정 횟수만큼 자동 재시도합니다. 일시적 네트워크 오류가 새벽에 사람을 깨우지 않게 하는 첫 번째 방어선입니다
• retry_delay: 재시도 간격입니다. 외부 API 장애라면 바로 재시도해도 또 실패하므로 간격을 둡니다
• 재시도가 의미 있으려면 task가 멱등해야 합니다. 같은 task instance를 두 번 돌려도 결과가 같아야 한다는 조건인데, 이것은 6편에서 설계 원칙으로 다룹니다

실패 알림은 on_failure_callback에 함수를 걸어 Slack이나 메일로 보냅니다. 구체적 연동은 5편의 Connection과 함께 다룹니다.

다음 편은 Airflow에서 가장 많이 데이는 시간 개념입니다.

다음 글: Airflow 해부 (4) 스케줄과 시간: logical date, catchup, backfill