Airflow Slack Webhook 사용법 정리

BAO.DE·2025년 10월 13일

airflow

Apache Airflow

목록 보기

19/20

Slack Webhook 이란?

Airflow에서 DAG 성공/실패 시 Slack으로 알림

사전준비

Slack 앱 설정 페이지로 이동:
👉 https://api.slack.com/apps

"Create New App" → "From Scratch" 선택

Incoming Webhooks 기능 활성화 ( On )
-> Add New Webhooks

"Add New Webhook to Workspace" 클릭 → 원하는 채널 선택

생성된 Webhook URL 복사
(예: https://hooks.slack.com/services/T000/B000/XXXXX

Airflow 연결 설정

Airflow UI → Admin → Connections → + (Add a new record)

Extra
{
"webhook_token": "https://hooks.slack.com/services/xxxxx
}

dag성공했는데 메시지 전송이 안돼서 애먹음 ㅠㅠ

pip list 확인

없으면 설치필요

Slack Alert Test

def send_slack():
    # Connection에서 webhook_token을 읽도록 설정
    hook = SlackWebhookHook(slack_webhook_conn_id="slack_webhook")

    # Slack 메시지 보내기
    response = hook.send_dict({
        "text": "Airflow Slack Test 🚀@@@@@@@"
    })

    print("Slack response:", response)

Callback 종류

call back 함수 성공,실패외에 트리거 조건이 여러개 존재한다.

Callback Test (T/F)

성공 실패에 따른 callback 함수 작성
-> xcom 이용해서 선행 task를 통해 True/False 결과값을 전달받고
context[] 메세지 전송

좀 더 양질의 정보와 , 가독성 확보를 위해서 context 수정 필요해보임

context 사용 가능 객체

chat gpt 짱편하다 ^^..

구분	Key 이름	설명	예시 값 / 형태
DAG 관련	`dag`	현재 실행 중인 DAG 객체	`<DAG: example_dag>`
	`dag_run`	현재 실행 중인 DAG Run 객체	`<DagRun example_dag @ 2025-10-16T00:00:00+00:00>`
	`task`	현재 Task 객체	`<Task(PythonOperator): sample_task>`
	`task_instance` (`ti`)	현재 TaskInstance 객체 (XCom 등에서 자주 사용)	`<TaskInstance: example_dag.sample_task 2025-10-16 00:00:00+00:00 [running]>`
	`task_map_index`	동적 태스크 매핑 시 인덱스 값	`0`, `1`, `2`, ...
스케줄 관련	`ds`	실행 날짜 (YYYY-MM-DD, 문자열)	`'2025-10-16'`
	`ds_nodash`	하이픈 없는 날짜	`'20251016'`
	`ts`	Task 실행 타임스탬프 (ISO 8601 형식)	`'2025-10-16T00:00:00+00:00'`
	`ts_nodash`	하이픈 없는 타임스탬프	`'20251016T000000'`
	`execution_date`	실행 기준 시각 (datetime 객체)	`datetime(2025, 10, 16, 0, 0)`
	`logical_date`	DAG의 논리적 실행 시간 (Airflow 2.4+ 기준 `execution_date`와 동일)	`datetime(2025, 10, 16, 0, 0)`
	`next_ds`	다음 스케줄의 실행 날짜	`'2025-10-17'`
	`prev_ds`	이전 스케줄의 실행 날짜	`'2025-10-15'`
	`data_interval_start`	데이터 처리 구간의 시작 시각 (`datetime`)	`datetime(2025, 10, 15, 0, 0)`
	`data_interval_end`	데이터 처리 구간의 종료 시각 (`datetime`)	`datetime(2025, 10, 16, 0, 0)`
경로 관련	`run_id`	DAG Run 고유 ID	`'scheduled__2025-10-16T00:00:00+00:00'`
	`run_type`	DAG Run의 유형 (`manual`, `scheduled`, `backfill`)	`'scheduled'`
파일/경로	`conf`	DAG Run 시 전달된 config (trigger 시 전달 가능)	`{"key": "value"}`
	`params`	DAG 내에서 정의된 파라미터	`{"my_param": "test"}`
템플릿 관련	`macros`	Airflow에서 제공하는 매크로 모음	`macros.datetime, macros.ds_add` 등
기타 유용한 키	`try_number`	현재 TaskInstance의 재시도 횟수	`1`
	`prev_start_date_success`	이전 성공 실행의 시작 시간	`datetime(2025, 10, 15, 0, 0)`
	`inlets`, `outlets`	Data Lineage에서 입출력 객체	`[Dataset("s3://...")]`
	`templates_dict`	`PythonOperator`의 `templates_dict` 인자에 전달된 값	`{'table': 'users'}`

수정버전

msg = (
        f"{status_emoji} *Airflow DAG 실행 결과 알림*\n"
        f"> 실행 시각: *{exec_date}* (Airflow Logical Date)\n"
        f"> DAG ID: `{dag_id}`\n"
        f"> Task ID: `{task_id}`\n"
        f"> Run ID: `{run_id}`\n"
        f"> 결과: *{status_text}* (랜덤 결과={result})\n"
        f"> 알림 전송 시각: {now_time}"
    )

좀 더 가독성이 좋아보인다.

Slack Block Kit

Slack Block Kit은 슬랙 메시지를 구조적이고 예쁘게 표현하기 위한 UI 프레임워크이다.(JSON 기반 레이아웃)

{
  "blocks": [
    {
      "type": "section",
      "text": {
        "type": "mrkdwn",
        "text": "Hello World!"
      }
    },
    {
      "type": "actions",
      "elements": [
        {
          "type": "button",
          "text": {
            "type": "plain_text",
            "text": "Click me"
          },
          "value": "button_click"
        }
      ]
    }
  ]
}

https://app.slack.com/block-kit-builder
여기 들어가서 만들어보고 테스트 가능