Monitoring, Peringatan, dan Callback

Pengantar Apache Airflow dengan Python

Mike Metzger

Data Engineer

Siklus Hidup Dag

Setiap run Dag dan task melalui urutan status
- Queued
- Running
- Status akhir: Success, Failed, Skipped
Airflow melacak status dan transisi

Diagram siklus hidup Dag run dari queued ke running hingga status akhir

Callback

Fungsi pada run Dag atau Task yang otomatis dipanggil Airflow
Callback untuk transisi status tertentu
- on_failure_callback - Dag run / task gagal
- on_success_callback- Dag run / task berhasil
Callback khusus task
- on_retry_callback
- on_skipped_callback
- on_execute_callback

Ilustrasi callback Airflow yang dipicu pada transisi status

Konteks callback

Airflow otomatis meneruskan dictionary context
context berisi info tentang run Dag / task
- context["dag"].dag_id - Nama Dag
- context["task_instance"].task_id - Nama task
- context["logical_date"] - Tanggal run Dag
Fungsi callback harus menerima 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.")

Contoh 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.

Notifier

Fungsi yang dapat diikat ke callback
Kirim peringatan ke sistem eksternal
Tersedia beberapa notifier:
- SmtpNotifier - Mengirim peringatan email
- SlackNotifier - Mengirim pesan ke channel Slack
- Notifier khusus aplikasi - PagerDuty, OpsGenie

Ilustrasi notifiers Airflow yang mengirim peringatan ke sistem eksternal

SmtpNotifier

Mengirim email saat callback dipicu
Ada di pustaka airflow.providers.smtp.notifications.smtp
Memerlukan atribut from_email dan to

Dapat menyertakan subject, html_content, dll.

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

Audit Log

Gunakan Audit Log untuk memantau Airflow
Berisi urutan semua peristiwa pada instance Airflow

Halaman audit log Airflow yang mencantumkan peristiwa sistem menurut jenis

Ayo berlatih!

Pengantar Apache Airflow dengan Python