모니터링, 알림 및 콜백

Python으로 배우는 Apache Airflow 입문

Mike Metzger

Data Engineer

DAG 생명 주기

모든 DAG 실행 및 태스크는 일련의 상태를 거침
- 대기 중
- 실행 중
- 최종 상태: 성공, 실패, 건너뜀
Airflow가 상태 및 전환 추적

대기 중에서 실행 중, 최종 상태로의 DAG 실행 생명 주기 다이어그램

콜백

Airflow가 DAG 실행 또는 태스크에서 자동으로 호출하는 함수
특정 상태 전환에 대한 콜백
- on_failure_callback - DAG 실행 / 태스크 실패 시
- on_success_callback - DAG 실행 / 태스크 성공 시
태스크 전용 콜백
- on_retry_callback
- on_skipped_callback
- on_execute_callback

상태 전환 시 트리거되는 Airflow 콜백 그림

콜백 컨텍스트

Airflow가 context 딕셔너리를 자동으로 전달
context에는 DAG 실행 / 태스크 정보 포함
- context["dag"].dag_id - DAG 이름
- context["task_instance"].task_id - 태스크 이름
- context["logical_date"] - DAG 실행 날짜
콜백 함수는 반드시 context를 인수로 받아야 함

def alert_on_failure(context):

    dag_id = context["dag"].dag_id
    task_id = context["task_instance"].task_id
    print(f"Task {task_id} in DAG {dag_id} has failed.")

콜백 예시

def alert_on_failure(context):
    dag_id = context["dag"].dag_id
    task_id = context["task_instance"].task_id
    print(f"Task {task_id} in Dag {dag_id} has failed.")


@dag(on_failure_callback=alert_on_failure)
def sales_etl_dag():
    @task()
    def data_import_task():
        raise ValueError("Simulated failure")


# Task data_import_task in Dag sales_etl_dag has failed.

노티파이어

콜백에 연결할 수 있는 함수
외부 시스템에 알림 전송
사용 가능한 노티파이어:
- SmtpNotifier - 이메일 알림 전송
- SlackNotifier - Slack 채널에 메시지 게시
- 앱별 노티파이어 - PagerDuty, OpsGenie

Airflow 노티파이어가 외부 시스템에 알림을 전송하는 그림

SmtpNotifier

콜백 시 이메일 전송
airflow.providers.smtp.notifications.smtp 라이브러리 사용
from_email 및 to 속성 필수

subject, html_content 등 포함 가능

@dag(dag_id=`sales_etl_dag`,
   on_failure_callback=SmtpNotifier(
       to='[email protected]',
       from_email='[email protected]',
       subject='Dag sales_etl_dag has failed'
   )
)

감사 로그

감사 로그를 사용하여 Airflow 모니터링
Airflow 인스턴스의 모든 이벤트 순서 포함

유형별 시스템 이벤트가 나열된 Airflow 감사 로그 페이지

연습해 봅시다!

Python으로 배우는 Apache Airflow 입문