Monitoring, Alerting, and Callbacks

Python में Apache Airflow परिचय

Mike Metzger

Data Engineer

Dag Lifecycle

हर Dag रन और टास्क कुछ स्टेट्स के क्रम से गुजरता है
- Queued
- Running
- Final state: Success, Failed, Skipped
Airflow स्टेट्स और उनके ट्रांज़िशन ट्रैक करता है

Queued से Running होते हुए अंतिम स्टेट तक Dag रन लाइफसाइकल का डायग्राम

Callbacks

वे फंक्शन जो Dag रन या टास्क पर Airflow अपने-आप कॉल करता है
खास स्टेट ट्रांज़िशन के लिए कॉलबैक
- on_failure_callback - Dag रन/टास्क फेल होने पर
- on_success_callback - Dag रन/टास्क सफल होने पर
टास्क-विशिष्ट कॉलबैक
- on_retry_callback
- on_skipped_callback
- on_execute_callback

स्टेट ट्रांज़िशन पर ट्रिगर होने वाले Airflow कॉलबैक का चित्रण

Callback context

Airflow अपने-आप एक context डिक्शनरी पास करता है
context में Dag रन/टास्क की जानकारी होती है
- context["dag"].dag_id - Dag का नाम
- context["task_instance"].task_id - टास्क का नाम
- context["logical_date"] - Dag रन की तारीख
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.")

Callback example

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.

Notifiers

वे फंक्शन जिन्हें कॉलबैक से जोड़ा जा सकता है
अलर्ट बाहरी सिस्टम तक भेजें
कई नोटिफायर उपलब्ध हैं:
- SmtpNotifier - ईमेल अलर्ट भेजता है
- SlackNotifier - Slack चैनल पर मैसेज पोस्ट करता है
- ऐप-विशिष्ट नोटिफायर - PagerDuty, OpsGenie

बाहरी सिस्टम को अलर्ट भेजते Airflow नोटिफायर का चित्रण

SmtpNotifier

कॉलबैक पर ईमेल भेजता है
airflow.providers.smtp.notifications.smtp लाइब्रेरी में है
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'
   )
)

Audit Log

Airflow की मॉनिटरिंग के लिए Audit Log का उपयोग करें
इसमें Airflow इंस्टेंस की सभी घटनाओं का क्रम होता है

इवेंट टाइप के अनुसार सिस्टम इवेंट्स दिखाता Airflow ऑडिट लॉग पेज

अभ्यास करते हैं!

Python में Apache Airflow परिचय