Начнем с того, что в Airflow есть несколько состояний для таски:
⭐️none - пока отдыхает
⭐️scheduled - должна быть запущена, все зависимости выполнены
⭐️queued - ждет свободный воркер
⭐️running - работает
⭐️success - успешно завершилась
⭐️restarting - перезапустили
⭐️failed - упала
⭐️skipped - пропущена
⭐️upstream_failed - упала предыдущая таска, которая нам нужна
⭐️up_for_retry - упала, но будет перезапущена
⭐️up_for_reschedule - сенсор будет перезапущен
⭐️deferred - отложена и ждет триггер
⭐️removed - удалена из дага после запуска
Подвисшие сенсоры уходят в статус deferred. У нас они имеют такой нейминг - mytask_awaiting_somedag. Я написала себе запрос, который выводит:
⭕️название дага, на который смотрит сенсор
⭕️количество сенсоров, которые ждут этот даг
⭕️общее количество подвисших сенсоров
И так можно сразу понять, на какой даг смотрит наибольшее количество сенсоров, и посмотреть причину
with sensored as (
SELECT
substr(
task_id,
strpos(task_id, 'awaiting_') + length('awaiting_')
) as sensor,
dag_id
FROM airflow.public.task_instance
WHERE state = 'deferred'
)
select
sensor,
count(1) over(partition by sensor) as sensor_cnt,
count(1) over() as total_cnt,
dag_id
from sensored
order by 2 desc, sensor, dag_id;