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

Иногда стандартных сенсоров недостаточно, и тогда требуется написать свой. 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-модель с триггерами.