Triggerer и Deferrable Operators — обзор модуля

Deferrable operators (AIP-40, 2.2+) — фундаментальное улучшение для долгоживущих sensors. Раньше sensor в poke mode блокировал worker slot на часы, или в reschedule mode попадал в queue каждые N секунд. Deferrable — async event loop держит тысячи triggers без блокировки worker.

Уроки модуля

Ключевые концепты

Lifecycle deferrable task

1. Task в worker вызывает self.defer(trigger=MyTrigger(...), method_name="execute_complete").
2. Worker сериализует trigger → пишет в таблицу trigger → TI state=deferred → освобождает slot.
3. Triggerer берёт trigger (SELECT ... FOR UPDATE SKIP LOCKED — multiple triggerer-ов работают параллельно).
4. asyncio loop запускает trigger.run() — async generator.
5. Trigger yields TriggerEvent → triggerer пишет event в БД, TI → scheduled.
6. Worker подбирает task заново, вызывает execute_complete(event).

Custom Trigger пример

class DateTimeTrigger(BaseTrigger):
    def __init__(self, moment):
        self.moment = moment

    def serialize(self):
        return ("path.to.DateTimeTrigger", {"moment": self.moment})

    async def run(self):
        while self.moment > utcnow():
            await asyncio.sleep(1)   # NB: asyncio, не time.sleep
        yield TriggerEvent(self.moment)

Scaling сравнение

Killer takeaway

Любой sensor с timeout > 1 час должен быть deferrable=True. Это фундаментальное правило production Airflow. Triggerer asyncio scale handles 5000+ одновременных triggers на одном процессе.