Airflow DAG 작성은(는) 무엇을 하나요?
사용 시점
- Airflow DAG 및 태스크 생성 또는 리팩토링 시
- 스케줄, 캐치업/백필, 재시도 및 SLA 구성 시
- 태스크 간 데이터 전달(XCom) 또는 연결/변수 사용 시
- 실행 중인 손상된 DAG 진단에는 사용하지 마십시오. (대신
debugging-airflow-pipelines사용)
워크플로우
- [ ] 각 태스크를 멱등적이게 만들고 데이터 간격에 의해 매개변수화
- [ ] 비용이 많이 들거나 임포트가 많은 코드는 모듈 최상위 레벨이 아닌 태스크 내부에 유지
- [ ] 스케줄 + 캐치업을 의도적으로 설정
- [ ] 재시도, retry_delay 및 SLA 구성
- [ ] TaskFlow 반환 값 또는 >> 연산자를 통해 의존성 연결
- 멱등성 태스크 —
2026-01-15간격의 태스크는 한 번 실행되든 다시 실행되든 동일한 결과를 생성해야 합니다.datetime.now()대신 데이터 간격을 사용하십시오. - 무거운 최상위 코드 없음 — 스케줄러는 모든 DAG 파일을 자주 파싱합니다. 모듈 레벨에서의 데이터베이스 호출, API 호출 또는 큰 임포트는 스케줄링 속도를 늦추고 파싱을 중단시킬 수 있습니다. 이를 태스크 내부에 배치하십시오.
- 의도적인 스케줄 및 캐치업 —
catchup=True는start_date부터 모든 누락된 간격을 백필합니다. 원하지 않으면 기본값False로 설정하십시오. - 재시도 및 SLA — 일시적인 실패는 정상입니다.
retries및retry_delay를 설정하고, 지연에 대해 SLA/알림을 사용하십시오.
패턴
TaskFlow DAG, 멱등적이고 깔끔하게 연결됨:
from airflow.decorators import dag, task
import pendulum
@dag(
schedule="@daily",
start_date=pendulum.datetime(2026, 1, 1, tz="UTC"),
catchup=False,
default_args={"retries": 3, "retry_delay": pendulum.duration(minutes=5)},
tags=["orders"],
)
def orders_pipeline():
@task
def extract(data_interval_start=None):
# now()가 아닌 간격을 사용하므로 재실행이 결정적입니다.
return fetch_orders(day=data_interval_start.date())
@task
def load(rows):
# 파티션을 삭제-삽입 -> 재시도 시 멱등성 보장.
overwrite_partition("fct_orders", rows)
load(extract())
orders_pipeline()
작은 데이터는 XCom(반환 값)을 통해 전달하고, 큰 데이터는 스토리지를 통해 전달 — S3/GCS/웨어하우스에 쓰고 경로/키를 전달하며, XCom으로 메가바이트를 절대 전달하지 마십시오.
연결/변수 사용 — 비밀 및 구성을 위해 (BaseHook.get_connection, Variable.get) 사용하고, 절대 하드코딩된 자격 증명을 사용하지 마십시오.
일반적인 함정
- 최상위 API/DB 호출 또는 무거운 임포트 — 스케줄러 속도를 늦추고 전체 배포의 DAG 파싱을 실패시킬 수 있습니다.
- 태스크 내부의
datetime.now()— 멱등성과 백필을 깨뜨립니다.data_interval_start/_end를 사용하십시오. - 의도하지 않은
catchup=True— 첫 배포 시 클러스터를 과거 실행으로 가득 채웁니다. - XCom을 통한 대용량 페이로드 — 메타데이터 DB를 부풀립니다. 참조를 전달하십시오.
- 동적
start_date(예:days_ago) — 스케줄을 비결정적으로 만듭니다. 고정 타임스탬프를 사용하십시오. - 하나의 거대한 태스크 — 추출/변환/로드를 분할하여 재시도 단위를 세분화하십시오.