Моніторинг, сповіщення та зворотні виклики

Вступ до Apache Airflow на Python

Mike Metzger

Data Engineer

Життєвий цикл Dag

Кожен запуск Dag і завдання проходить послідовність станів
- Queued
- Running
- Кінцевий стан: Success, Failed, Skipped
Airflow відстежує стани та переходи

Схема життєвого циклу запуску Dag: від queued і running до фінального стану

Зворотні виклики

Функції на запуску Dag або завданні, які Airflow викликає автоматично
Зворотні виклики на специфічні переходи станів
- on_failure_callback — збій запуску Dag / завдання
- on_success_callback — успішний запуск Dag / завдання
Специфічні для завдань зворотні виклики
- on_retry_callback
- on_skipped_callback
- on_execute_callback

Ілюстрація зворотних викликів Airflow, що спрацьовують на переходах станів

Контекст зворотного виклику

Airflow автоматично передає словник context
context містить відомості про запуск Dag / завдання
- context["dag"].dag_id — назва Dag
- context["task_instance"].task_id — назва завдання
- context["logical_date"] — дата запуску Dag
Функція зворотного виклику має приймати 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.")

Приклад зворотного виклику

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-и

Функції, які можна прив'язати до зворотних викликів
Надсилайте сповіщення у зовнішні системи
Доступні кілька notifier-ів:
- SmtpNotifier — надсилає email-сповіщення
- SlackNotifier — публікує повідомлення в каналі Slack
- Спеціалізовані — PagerDuty, OpsGenie

Ілюстрація notifier-ів Airflow, що надсилають сповіщення у зовнішні системи

SmtpNotifier

Надсилає email під час зворотного виклику
У бібліотеці 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

Використовуйте Audit Log для моніторингу Airflow
Містить послідовність усіх подій інстансу Airflow

Сторінка журналу аудиту Airflow зі списком системних подій за типами

Давайте потренуємось!

Вступ до Apache Airflow на Python