오류 처리와 재시도로 견고한 DAG 만들기

Airflow로 데이터 파이프라인 구축하기

Volker Janz

Senior Developer Advocate at Astronomer

태스크 실패 원인

태스크 연결 문제

API 타임아웃 및 속도 제한
일시적인 네트워크 중단 및 DNS 조회 지연
여러 태스크가 동일한 리소스를 동시에 요청하는 경우

재시도

@task(
    retries=3,
    retry_delay=timedelta(minutes=2),
)
def fetch_weather():
    response = requests.get("https://api.weather.com/forecast")
    response.raise_for_status()
    return response.json()

실패 시 최대 3번 재시도
재시도 사이에 2분 대기

지수 백오프

@task(
    retries=3,
    retry_delay=timedelta(minutes=2),
    retry_exponential_backoff=2.0,
)
def fetch_weather():
    ...

지수 백오프

재시도마다 대기 시간이 2배로 증가
외부 서비스가 복구할 시간을 확보

on_failure_callback

def alert_on_failure(context):
    dag_id = context["dag"].dag_id
    task_id = context["ti"].task_id
    print(f"ALERT: {task_id} in {dag_id} failed!")

@task(on_failure_callback=alert_on_failure)
def fetch_weather():
    ...

context 딕셔너리: dag, ti, exception, 로그 URL
Slack, PagerDuty 등 알림 도구로 전달 가능

DAG 수준 vs 태스크 수준 콜백

DAG 수준 (전체 적용)

@dag(
    on_failure_callback=alert_team,
)
def my_pipeline():
    ...

태스크 수준 (개별 적용)

@task(
    on_failure_callback=page_oncall,
)
def critical_step():
    ...

태스크 수준이 DAG 수준을 재정의
두 수준을 함께 사용해 계층적 알림 구성 가능

max_consecutive_failed_dag_runs

@dag(max_consecutive_failed_dag_runs=3)
def monitoring_pipeline():
    ...

연속 실패 후 자동 일시 중지된 DAG 실행

지속적인 실패로 인한 알림 피로 방지
수동으로 재개할 때까지 DAG 일시 중지 유지

전체 정리

@dag(
    max_consecutive_failed_dag_runs=3,
    on_failure_callback=alert_team,
)
def weather_pipeline():

    @task(
        retries=3,
        retry_delay=timedelta(minutes=2),
        retry_exponential_backoff=True,
        on_failure_callback=page_oncall,
    )
    def fetch_weather():
        ...

재시도 + 백오프로 일시적 오류 처리
콜백으로 담당자에게 즉시 알림
자동 일시 중지로 불필요한 알림 차단

연습해 봅시다!

Airflow로 데이터 파이프라인 구축하기