İzleme, Uyarılar ve Geri Çağrılar

Python ile Apache Airflow’a Giriş

Mike Metzger

Data Engineer

DAG Yaşam Döngüsü

Her DAG çalıştırması ve görevi bir dizi durumdan geçer
- Kuyrukta
- Çalışıyor
- Son durum: Başarılı, Hatalı, Atlandı
Airflow durumları ve geçişleri izler

DAG çalıştırma yaşam döngüsünün kuyruktan çalışmaya ve son duruma gidişini gösteren diyagram

Geri Çağrılar

Airflow'un DAG çalıştırması veya Görev üzerinde otomatik çağırdığı fonksiyonlar
Belirli durum geçişleri için geri çağrılar
- on_failure_callback - DAG çalıştırması/görev hatalı
- on_success_callback - DAG çalıştırması/görev başarılı
Göreve özgü geri çağrılar
- on_retry_callback
- on_skipped_callback
- on_execute_callback

Durum geçişlerinde tetiklenen Airflow geri çağrılarını gösteren görsel

Geri çağrı bağlamı

Airflow otomatik olarak bir context sözlüğü geçirir
context DAG çalıştırması/görev hakkında bilgi içerir
- context["dag"].dag_id - DAG adı
- context["task_instance"].task_id - Görev adı
- context["logical_date"] - DAG çalıştırma tarihi
Geri çağrı fonksiyonu context kabul etmelidir

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.")

Geri çağrı örneği

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.

Bildiriciler

Geri çağrılara bağlanabilen fonksiyonlar
Uyarıları harici sistemlere gönderir
Birçok bildirici mevcut:
- SmtpNotifier - E-posta uyarıları gönderir
- SlackNotifier - Bir Slack kanalına mesaj gönderir
- Uygulamaya özel bildiriciler - PagerDuty, OpsGenie

Airflow bildiricilerinin uyarıları harici sistemlere göndermesini gösteren görsel

SmtpNotifier

Geri çağrıda e-posta gönderir
airflow.providers.smtp.notifications.smtp kütüphanesinde
from_email ve to alanları zorunlu

subject, html_content vb. eklenebilir

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

Denetim Günlüğü

Airflow'u izlemek için Denetim Günlüğünü kullanın
Airflow örneğindeki tüm olayların sırasını içerir

Olayları türe göre listeleyen Airflow denetim günlüğü sayfası

Haydi pratik yapalım!

Python ile Apache Airflow’a Giriş