前言

通知告警在完善 data pipeline 當中是非常重要的，slack 只是其中一個方式，就像 AWS 很多服務都會設定條件用 email 通知，在設計 data pipeline 時也需要常常思考，什麼情況需要做檢查，有什麼問題需要通知，最簡單好懂的就是程式 error 要通知，其他情況會比較像是部署、跑測試、資料檢查時，都會需要在特定節點通知，今天的內容是從最基礎的如何用 slack api 結合 airflow providers 提供的 slack 通知 function 來實作。

先決條件

1. 有 slack 帳號
2. 有加入任一個 workspace

開通 Slack Token 步驟

1. 進到 slack api 網站

https://api.slack.com/

2.點擊右上方的「Your apps」，並登入 slack account

3.點擊 create an app

4. 選擇 from scratch

5. 給 App 取名，選擇要在哪個 Workspace

• 名稱後面都可以調整，但工作站設定就固定了

6. Enable Socket Mode

7. 產生 slack token

8. 進到 OAuth 並 新增 scope

9. 新增完之後上方的 Oauth Tokens 就能夠載入到 Workspace 中

10. 允許之後就能看到token已經可以使用

• 取得 xoxb- 開頭的 token 後續連結就沒問題了

Airflow 設定 Slack API

設定 Connections

和前幾天設定 s3 一樣，但是這次的 type 是 slack api ，接著只需要將 xoxb- 開頭的 token 填入，並設定一個連接的名稱，我這邊是設定 slack_conn。

示範程式碼

from airflow import DAG
from airflow.operators.empty import EmptyOperator
from airflow.providers.slack.notifications.slack import send_slack_notification
from pendulum import datetime

with DAG(
    dag_id='slack_notification_dag',
    start_date=datetime(2024, 9, 19),
    schedule=None,
    catchup=False
    ) as dag:

    start_task = EmptyOperator(
        task_id='start_task',
        on_success_callback=send_slack_notification(
            slack_conn_id="slack_conn",
            text="DAG run successful! ",
            channel="#alerts"),
        dag=dag
    )

    start_task

• 用 EmptyOperator 模擬平常 task 執行，成功的話就會執行 on_success_callback 後的內容，在這裡就直接設定 send_slack_notification 就可以在成功時通知了