การมอนิเตอร์ การแจ้งเตือน และ Callbacks

Apache Airflow เบื้องต้นด้วย Python

Mike Metzger

Data Engineer

วงจรชีวิตของ Dag

 

  • ทุก Dag run และ task จะผ่านลำดับสถานะต่างๆ
    • Queued
    • Running
    • สถานะสุดท้าย: Success, Failed, Skipped
  • Airflow ติดตามสถานะและการเปลี่ยนแปลง

ไดอะแกรมวงจรชีวิตของ Dag run ตั้งแต่ queued ถึง running จนถึงสถานะสุดท้าย

Apache Airflow เบื้องต้นด้วย Python

Callbacks

  • ฟังก์ชันที่ Airflow เรียกโดยอัตโนมัติเมื่อ Dag run หรือ task เปลี่ยนสถานะ
  • Callbacks สำหรับการเปลี่ยนสถานะต่างๆ
    • on_failure_callback - Dag run / task ล้มเหลว
    • on_success_callback - Dag run / task สำเร็จ
  • Callbacks เฉพาะ task
    • on_retry_callback
    • on_skipped_callback
    • on_execute_callback

ภาพประกอบ callbacks ของ Airflow ที่ถูกเรียกเมื่อสถานะเปลี่ยน

Apache Airflow เบื้องต้นด้วย Python

Context ของ Callback

  • Airflow ส่ง dictionary context ให้โดยอัตโนมัติ
  • context มีข้อมูลเกี่ยวกับ Dag run / task
    • context["dag"].dag_id - ชื่อของ Dag
    • context["task_instance"].task_id - ชื่อของ task
    • context["logical_date"] - วันที่ของ Dag run
  • ฟังก์ชัน callback ต้องรับพารามิเตอร์ 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.")
Apache Airflow เบื้องต้นด้วย Python

ตัวอย่าง Callback

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.
Apache Airflow เบื้องต้นด้วย Python

Notifiers

  • ฟังก์ชันที่เชื่อมต่อกับ callbacks ได้
  • ส่งการแจ้งเตือนไปยังระบบภายนอก
  • มี notifiers ให้เลือกใช้หลายรูปแบบ:
    • SmtpNotifier - ส่งการแจ้งเตือนทางอีเมล
    • SlackNotifier - โพสต์ข้อความไปยัง Slack channel
    • Notifiers เฉพาะแอป - PagerDuty, OpsGenie

ภาพประกอบ notifiers ของ Airflow ที่ส่งการแจ้งเตือนไปยังระบบภายนอก

Apache Airflow เบื้องต้นด้วย Python

SmtpNotifier

  • ส่งอีเมลเมื่อมี callback
  • อยู่ในไลบรารี airflow.providers.smtp.notifications.smtp
  • ต้องระบุ attribute 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'
   )
)
Apache Airflow เบื้องต้นด้วย Python

Audit Log

  • ใช้ Audit Log เพื่อมอนิเตอร์ Airflow
  • บันทึกลำดับเหตุการณ์ทั้งหมดบน Airflow instance

หน้า Audit Log ของ Airflow แสดงรายการเหตุการณ์ของระบบตามประเภท

Apache Airflow เบื้องต้นด้วย Python

มาฝึกกันเถอะ!

Apache Airflow เบื้องต้นด้วย Python

Preparing Video For Download...