Как реализуется собственный сенсор в Airflow?

Question

Собственный сенсор создаётся путём наследования от BaseSensorOperator. В нём реализуется метод poke, который проверяет условие. Airflow вызывает этот метод с заданным интервалом. Для асинхронной версии используется deferrable-модель и триггеры. Такой подход позволяет реализовать ожидание любых внешних условий.

YeaHub · Accepted Answer

Иногда стандартных сенсоров недостаточно, и тогда требуется написать свой. Airflow предоставляет для этого понятный и расширяемый механизм.ОпределениеCustom sensor — это пользовательский сенсор, реализованный через наследование от базового класса Airflow.1. Классический (blocking) сенсорСамый простой вариант — синхронный сенсор.1.1. Базовая структураНужно:унаследоваться от BaseSensorOperator;реализовать метод poke.Пример упрощённой реализации:from airflow.sensors.base import BaseSensorOperator

class MySensor(BaseSensorOperator):
    def poke(self, context):
        # проверить условие
        return check_condition()
Если poke вернул:True → сенсор завершён;False → Airflow подождёт poke_interval и попробует снова.2. Deferrable-сенсорДля долгих ожиданий лучше использовать асинхронную модель.2.1. Основная идеяDeferrable-сенсор:проверяет условие;если не готово — вызывает defer;передаёт управление trigger-у.2.2. Упрощённая логикаdef execute(self, context):
    if not condition():
        self.defer(trigger=MyTrigger())
Trigger:асинхронно ждёт событие;возвращает управление сенсору.3. Когда писать свой сенсорИмеет смысл, если:нет готового сенсора;логика проверки специфична;важно оптимизировать ресурсы.4. Практические рекомендациивсегда задавайте timeout;логируйте проверки;тестируйте сенсор отдельно от DAG.5. Краткий выводСобственные сенсоры позволяют адаптировать Airflow под любые внешние условия. Для коротких ожиданий подойдут blocking-сенсоры, для долгих — deferrable-модель с триггерами.