Airflow DAG 오케스트레이터은(는) 무엇을 하나요?
Apache Airflow DAG를 설계하고 운영하여 적절한 의존성 관리, SLA, 모니터링을 갖춘 안정적인 데이터 파이프라인 오케스트레이션을 구현합니다.
활성화 트리거
다음에서 활성화: "Airflow", "DAG", "operator", "sensor", "scheduler", "task dependency", "SLA", "backfill", "XCom", "TaskFlow API", "MWAA", "Cloud Composer"
다음에는 사용하지 않음: dbt 모델 실행 → dbt-analytics-engineer (Airflow가 dbt를 트리거할 수는 있음) | 스트림 처리 → streaming-pipeline-architect | 워크플로 엔진(Temporal) → distributed-transaction-manager
빠른 시작
- DAG 정의 — TaskFlow API(
@dag,@task데코레이터)를 사용하여 Python 네이티브 DAG 작성 - 스케줄 설정 — cron 또는 timetable,
catchup=False(백필 의도가 없는 경우) - 재시도 구성 — 모든 태스크에
retries=2,retry_delay=timedelta(minutes=5)설정 - SLA 추가 — 중요 경로 태스크에
sla=timedelta(hours=2)설정 - 로컬 테스트 — 배포 전
airflow dags test my_dag 2026-01-01실행
핵심 기능
| 도메인 | 기술 |
|---|---|
| Airflow | Apache Airflow 2.10+, MWAA, Cloud Composer 3 |
| 오퍼레이터 | BashOperator, PythonOperator, KubernetesPodOperator |
| 프로바이더 | apache-airflow-providers-{snowflake, google, aws, dbt-cloud} |
| 실행기 | CeleryExecutor, KubernetesExecutor, LocalExecutor |
| 모니터링 | SLA 미달, 태스크 지속 시간, Airflow 메트릭 → Prometheus |
아키텍처 패턴
TaskFlow API DAG
from airflow.decorators import dag, task
from datetime import datetime, timedelta
@dag(
schedule="0 6 * * *", # 매일 06:00 UTC
start_date=datetime(2026, 1, 1),
catchup=False,
default_args={
"retries": 2,
"retry_delay": timedelta(minutes=5),
"sla": timedelta(hours=2),
},
tags=["finance", "daily"],
)
def daily_revenue_pipeline():
@task()
def extract_payments() -> dict:
"""Stripe API에서 추출"""
data = stripe_client.list_payments(date=today())
return {"count": len(data), "path": "s3://raw/payments/"}
@task()
def extract_orders() -> dict:
"""Shopify API에서 추출"""
data = shopify_client.list_orders(date=today())
return {"count": len(data), "path": "s3://raw/orders/"}
@task()
def transform(payments: dict, orders: dict) -> str:
"""DuckDB에서 조인 및 변환"""
result_path = run_duckdb_transform(payments["path"], orders["path"])
return result_path
@task()
def load(path: str):
"""Snowflake에 로드"""
snowflake_copy_into("fct_revenue", path)
# 함수 호출을 통해 의존성 정의
payments = extract_payments()
orders = extract_orders()
transformed = transform(payments, orders)
load(transformed)
daily_revenue_pipeline()
동적 태스크 매핑 (팬아웃/팬인)
@task()
def get_partitions() -> list[str]:
return ["2026-01-01", "2026-01-02", "2026-01-03"]
@task()
def process_partition(partition_date: str) -> dict:
"""각 파티션에 대해 병렬 실행"""
return {"date": partition_date, "rows": process(partition_date)}
@task()
def aggregate(results: list[dict]):
"""팬인: 모든 파티션 결과 수신"""
total = sum(r["rows"] for r in results)
log.info(f"총 {total} 행 처리됨")
# process_partition을 모든 파티션에 동적으로 매핑
partitions = get_partitions()
results = process_partition.expand(partition_date=partitions)
aggregate(results)
dbt + Airflow 통합
from airflow.operators.bash import BashOperator
from cosmos import DbtDag, ProjectConfig, ProfileConfig
# 옵션 1: cosmos (권장)
dbt_dag = DbtDag(
project_config=ProjectConfig("/opt/airflow/dbt/"),
profile_config=ProfileConfig(
profile_name="default",
target_name="prod",
),
schedule="@daily",
dag_id="dbt_daily",
)
# 옵션 2: BashOperator (간단)
dbt_run = BashOperator(
task_id="dbt_run",
bash_command="cd /opt/airflow/dbt && dbt build --select tag:daily",
)
안티패턴
- Fat 태스크 — 태스크는 원자적인 단위여야 함; 하나의 태스크에 추출+변환+로드를 넣지 말 것
- 대용량 데이터에 XCom 사용 — XCom은 메타데이터(경로, 개수)용; 데이터는 S3/GCS를 통해 전달하고 XCom 피클은 사용 금지
- 실수로 catchup=True 설정 — 백필이 필요하지 않으면
catchup=False로 설정; 그렇지 않으면 Airflow가 모든 누락된 간격을 실행함 - 재시도 없음 — 일시적 실패는 흔함; 항상 지연 시간과 함께
retries >= 1설정 - DAG 파일의 최상위 레벨 코드 — DAG 파일은 30초마다 파싱됨; 모듈 레벨에서 무거운 import나 API 호출은 스케줄러를 느리게 만듦
품질 체크리스트
- Python 네이티브 DAG에 TaskFlow API 사용 (레거시 연산자 체이닝 대신)
- 백필 의도가 없으면
catchup=False설정 - 모든 태스크에 재시도 구성 (최소 1회 재시도)
- 중요 경로 태스크에 SLA 설정
- XCom이 데이터 페이로드 대신 참조(S3 경로)를 전달
- 팬아웃 병렬 처리를 위해 동적 태스크 매핑 사용
- 배포 전
airflow dags test로 DAG 로컬 테스트 완료 - 태스크 멱등성: 태스크 재실행 시 동일한 결과 생성
- 태스크 실패 및 SLA 미달에 대한 알림 구성
- Airflow UI에서 필터링을 위해 DAG 태그 적용