В этой статье для обучения дата-инженеров рассмотрим типы оповещений в Apache AirFlow и их отслеживание в сервисе мониторинга cron-заданий Healthchecks.io.
Оповещения Apache AirFlow: какие они бывают и зачем их отслеживать
Apache AirFlow позволяет создавать сложные конвейеры обработки данных, которые могут выполняться по расписанию, по событию или запускаться вручную. Для повышения надежности работы этих конвейеров в AirFlow есть механизм оповещений, которые предупреждают дата-инженера о сбоях на различных уровнях, от уровня отдельной задачи до целого DAG или проблем с инфраструктурой и внешними зависимостями. Настроив эти оповещения, можно оперативно узнавать о проблемах по мере их возникновения, чтобы быстро устранять их, сведя к минимуму время простоя.
При работе с Apache AirFlow наиболее важно настроить следующие оповещения:
- о сбоях DAG, если он не выполняется или при попытке выполнения возникают ошибки;
- о сбое планирования DAG, если он не запускается в течение определенного периода времени из-за отказа экземпляра AirFlow или его неправильной конфигурации;
- о сбоях задач, когда конкретная задача в DAG не выполняется или при попытке выполнения возникла ошибка;
- нарушения SLA (Service Level Agreement, соглашение об уровне обслуживания), когда рабочий процесс занимает больше времени, чем ожидалось. Часто это свидетельствует о проблемах с производительностью системы.
- о качестве данных, когда в рабочих процессах обнаруживаются отсутствующие или неверные данные, что может привести к некорректным результатам.
- об использовании ресурсов, когда кластер AirFlow потребляет слишком много ЦП или памяти. Эти оповещения позволят вовремя обнаружит проблемы с инфраструктурой и обеспечить бесперебойную работу конвейеров обработки данных.
Наиболее значимыми для дата-инженера при работе с Apache AirFlow можно назвать оповещения о сбоях DAG, о сбое планирования DAG, о сбое задачи и SLA. Целесообразно отслеживать их в сторонней системе, чтобы обеспечить избыточность предупреждений и гарантировать отсутствие пропуска этих важных сигналов, если сервер AirFlow недоступен. Настроив эти типы оповещений, можно обеспечить упреждающий мониторинг рабочих процессов и быстрое решение возникающих проблем, повышая общую надежность и производительность конвейеров обработки данных. Далее рассмотрим, как это сделать с помощью сервиса мониторинга HealthChecks.io.
Мониторинг оповещений с помощью HealthChecks.io
В идеале для работы с Apache AirFlow дата-инженеру нужна внешняя панель мониторинга, которая позволяет отслеживать запланированные задачи, просматривать историю их выполнения и предупреждать в случае возникновения проблемы или если запланированная задача не завершена в свое время. Этим требованиям отвечает HealthChecks.io – сервис мониторинга cron-заданий и подобных периодических пакетных процессов. Healthchecks.io прослушивает HTTP-запросы от cron-заданий и запланированных задач. Если сигнал ping не проходит вовремя, Healthchecks выдает предупреждение.
Этот сервис работает как аварийный выключатель для процессов, которые должны выполняться непрерывно или по регулярному известному расписанию. Например, создание резервных копий, очищение очереди задач, выполнение скриптов мониторинга состояния базы данных или формирования отчетов, периодический импорт данных и задания синхронизации, антивирусное сканирование, обновления DDNS и SSL. Однако, Healthchecks не подходит для мониторинга времени работы веб-сайта путем проверки его с помощью HTTP-запросов, сбора метрик производительности приложения и агрегации журналов.
Проверка в Healthchecks.io представляет один сервис, который надо отслеживать. Например, при отслеживании cron-заданий надо создать отдельную проверку для каждого отслеживаемого задания. Каждая проверка имеет уникальный URL-адрес для проверки связи, установленное расписание и соответствующие интеграции. Клиенты (задания cron, фоновые рабочие процессы, пакетные сценарии, запланированные задачи, веб-сервисы и пр.) отправляют HTTP-запросы к URL-адресу ping, чтобы сообщить о начале выполнения, об успешном или неудачном выполнении.
Healthchecks.io поддерживает два формата URL-адресов для проверки связи:
- https://hc-ping.com/<uuid> — по уникальному идентификатору UUID;
- https://hc-ping.com/<project-ping-key>/<name-slug> — по ключу проекта и именем проверки.
Можно добавить /start, /fail или /<exitcode> к базовому URL-адресу ping для отправки сигналов start и failure, если это необходимо. UUID проверки и ключи Ping проекта являются конфиденциальной информацией. Если они будут общедоступными, несанкционированный пользователь может отправить сигналы телеметрии на проверки и нарушить мониторинг.
Для каждой проверки можно установить параметр конфигурации Grace Time — дополнительное время отсрочки перед отправкой оповещения в случае задержки проверки. Это позволяет учитывать небольшие ожидаемые отклонений во времени выполнения задания, связанные, например, с некритическими помехами в сети. При использовании сигнала start для измерения времени выполнения задания, время отсрочки также устанавливает максимально допустимый временной интервал между сигналами start и success. Если задание отправляет сигнал start, но затем не отправляет сигнал success в течение периода Grace Time, Healthchecks предполагает, что задание не выполнено, и генерирует предупреждения.
Сервис Healthchecks имеет открытый исходный код и может быть развернут на собственном сервере, а также доступен как продукт SAAS. Он предоставляет API, который упрощает его интеграцию с AirFlow, чтобы отправлять сигналы ping из AirFlow в HealthChecks после завершения запуска DAG. Это реализуется с помощью функции обратного вызова AirFlow, такой как on_failure_callback или on_success_callback, о чем мы писали здесь и здесь.
Для настройки отслеживания оповещений AirFlow в сервисе Healthchecks надо сначала создать в каталоге DAGS и папке utils новый файл с именем healthchecksio.py, содержащий следующий Python-код:
import requests
# Ping key can be found in the HealthChecks.io dashboard
PING_KEY: str = 'REPLACE-WITH-YOUR-HEALTHCHECKS.IO-PINGKEY'
def fail_alert(context) -> None:
task_instance = context.get('task_instance')
dag_id = task_instance.dag_id.lower()
url = 'https://hc-ping.com/' + PING_KEY + '/' + dag_id + '/fail'
try:
requests.get (url, timeout=50.0)
except:
print("HealthChecks.io exception occurred when sending fail alert for " + dag_id)
def success_alert(context):
task_instance = context.get('task_instance')
dag_id = task_instance.dag_id.lower()
url = 'https://hc-ping.com/' + PING_KEY + '/' + dag_id
try:
requests.get (url, timeout=50.0)
except:
print("HealthChecks.io exception occurred when sending success alert for " + dag_id)
Далее надо создать еще 1 новый файл в папке utils с именем alerts.py и следующим Python-кодом:
from utils import healthchecksio
def failure(context) -> None:
healthchecksio.fail_alert(context)
def success(context) -> None:
healthchecksio.success_alert(context)
Наконец, можно перейти к DAG, для которого надо включить оповещения, и добавить следующую строчку в начало описания DAG:
from utils import alerts
а также 2 строки в раздел определения DAG:
on_failure_callback = alerts.failure, on_success_callback = alerts.success
Теперь для каждого успеха и неудачи DAG с названием, указанном в проверяемом URL-адресе, будет отправляться сигнал ping под контролем сервиса мониторинга HealthChecks.
Далее следует настроить оповещения для DAG, которые надо отслеживать в HealthChecks, задав в GUI этого сервиса имя проверки, соответствующее имени DAG, расписание запуска cron-заданий для него, время отсрочки перед отправкой уведомлений, теги для логической группировки нескольких DAG, а также каналы уведомлений (Slack, email и пр.). Читайте в нашей новой статье про отладку DAG с помощью GUI AirFlow.
Узнайте больше про использование Apache AirFlow для дата-инженерии и аналитики больших данных на специализированных курсах в нашем лицензированном учебном центре обучения и повышения квалификации для разработчиков, менеджеров, архитекторов, инженеров, администраторов, Data Scientist’ов и аналитиков Big Data в Москве:
Источники
