BaseOperator anatomy

В Airflow всё, что выполняется как task, — это инстанс BaseOperator или его наследник. PythonOperator, BashOperator, S3CopyObjectOperator, SparkSubmitOperator, ваш собственный MyCustomOperator — все они проходят через один и тот же lifecycle, описанный в корневом классе. Понимание этой механики отделяет тех, кто использует Airflow как чёрный ящик, от тех, кто пишет надёжные production operators и эффективно диагностирует странное поведение.

Этот урок препарирует BaseOperator до уровня методов: что происходит между моментом, когда scheduler решает «пора запускать task», и моментом, когда статус TaskInstance переходит в success.

Где BaseOperator в иерархии классов

Корневая иерархия Airflow 2.x:

Важный момент: MappedOperator (это task после .expand(...)) — отдельная ветка от AbstractOperator. Он не наследник BaseOperator, а альтернативная реализация, которая на runtime разворачивается в обычные task instances.

Минимальный custom operator

Минимально необходимый код для собственного operator:

from airflow.models import BaseOperator
from airflow.utils.context import Context

class GreetOperator(BaseOperator):
    template_fields = ("name",)
    ui_color = "#a8e6cf"
    ui_fgcolor = "#000000"

    def __init__(self, *, name: str, **kwargs):
        super().__init__(**kwargs)
        self.name = name

    def execute(self, context: Context):
        msg = f"Hello, {self.name}!"
        self.log.info(msg)
        return msg

Что здесь критично:

• super().__init__(**kwargs) — обязательный вызов. Без него scheduler не подхватит task_id, retries, dag и десятки других kwargs.
• template_fields — tuple строк, имена атрибутов, которые scheduler должен прогнать через Jinja перед execute().
• execute(context) — единственная обязательная сигнатура. Возвращаемое значение пушится в XCom (return_value) автоматически.
• ui_color / ui_fgcolor — цвет node в Graph view UI.

Полный lifecycle execute() — пошагово

Когда scheduler переводит TI из queued в running, executor запускает worker process, который вызывает TaskInstance._run_raw_task(). Внутри происходит цепочка из 11 шагов:

Если на шаге 5 или 6 возникает Exception:

1. state ставится up_for_retry (если есть retries) или failed.
2. Вызывается on_failure_callback (если задан).
3. Если это последний retry и есть email_on_failure — отправляется email через SmtpHook.
4. Если задан on_retry_callback и есть retries — он вызывается перед следующим try.

Template fields — Jinja-рендеринг атрибутов

template_fields — самый недооценённый механизм BaseOperator. Это объявление: «эти атрибуты на runtime прогоняются через Jinja с контекстом».

class S3DownloadOperator(BaseOperator):
    template_fields = ("bucket_name", "key", "local_path")
    template_ext = (".sql", ".json")  # Если значение заканчивается на эти суффиксы — это путь к файлу, читается и render-ится контент
    template_fields_renderers = {"key": "python"}  # Подсказка UI как highlight-ить в Rendered Templates tab

    def __init__(self, *, bucket_name: str, key: str, local_path: str, **kwargs):
        super().__init__(**kwargs)
        self.bucket_name = bucket_name
        self.key = key
        self.local_path = local_path

    def execute(self, context):
        self.log.info(f"Downloading s3://{self.bucket_name}/{self.key} → {self.local_path}")

Использование:

S3DownloadOperator(
    task_id="download_orders"
    bucket_name="data-prod"
    key="orders/{{ ds }}.parquet",          # ← Jinja
    local_path="/tmp/{{ ts_nodash }}_orders.parquet",
)

На runtime до pre_execute():

• self.key = "orders/2026-05-12.parquet"
• self.local_path = "/tmp/20260512T000000_orders.parquet"

template_fields — это tuple на классе, не на инстансе. Это значит, что наследники должны явно объявить свой набор:

class S3DownloadWithRegionOperator(S3DownloadOperator):
    template_fields = (*S3DownloadOperator.template_fields, "region")

    def __init__(self, *, region: str, **kwargs):
        super().__init__(**kwargs)
        self.region = region

Что доступно в Jinja context

Стандартные variables, доступные в template fields:

on_kill — что вызывается при tаsk timeout / clear running

on_kill() — критически важный, часто игнорируемый метод. Вызывается, когда:

1. Task превышает execution_timeout — scheduler шлёт SIGTERM в worker process.
2. Пользователь делает clear на running task через UI/CLI.
3. Worker pod evicted в Kubernetes.
4. Scheduler видит TI с state=running, но heartbeat прокис — adopt_or_reset.

Default on_kill — pass. Но если ваш operator запускает внешний процесс (Spark job, Dataflow pipeline, K8s pod, BigQuery query), отсутствие cleanup означает leak external resources даже после того, как Airflow считает task failed.

class SparkSubmitOperator(BaseOperator):
    def __init__(self, *, application: str, **kwargs):
        super().__init__(**kwargs)
        self.application = application
        self._spark_app_id = None  # запомним, чтобы убить при kill

    def execute(self, context):
        self._spark_app_id = self._submit_spark_application()
        self._wait_for_completion(self._spark_app_id)

    def on_kill(self):
        if self._spark_app_id:
            self.log.info(f"Killing Spark application {self._spark_app_id}")
            self._cancel_spark_application(self._spark_app_id)

Полезные attributes BaseOperator

Атрибуты, которые часто используются в production:

Production gotchas

1. template_fields забыли на наследнике. Subclass добавил новый атрибут, не указал в template_fields, и Jinja-выражения в нём не работают. Симптом: в UI Rendered Templates атрибут отсутствует, в логах видно raw {{ ds }} строку. Fix — template_fields = (*Parent.template_fields, "new_field").

2. super().__init__ забыт или передан с *args вместо **kwargs. Если вы пишете __init__(self, name, **kwargs) без super().__init__(**kwargs), то task_id, retries, dag и прочее не попадут в parent. DAG не запарсится с TypeError: BaseOperator.__init__() got an unexpected keyword argument 'task_id' или, что хуже, парсится, но retries молча игнорируются.

3. Mutable default в __init__. Classic Python pitfall в Airflow operators: def __init__(self, items: list = []): — этот list расшарен между всеми инстансами. На втором парсе DAG в нём окажутся данные первого. Используйте items: list | None = None и внутри self.items = items or [].

4. on_kill делает blocking call без timeout. Воркер начинает shutdown, on_kill ждёт ответа Spark master 30 минут, в итоге SIGKILL приходит, всё равно leak. Всегда добавляйте timeout=10 к любым network calls в on_kill.

5. State хранится на self между retries. Между try_number=1 и try_number=2 worker — отдельный процесс. Любое состояние, которое вы храните в self.X после execute(), теряется. Сохраняйте в XCom или внешнее хранилище.

6. do_xcom_push=False забыли на large-result tasks. По default любой return value пушится в XCom — это запись в metadata DB. Если ваш operator возвращает 50MB DataFrame as JSON, вы засоряете xcom table до тех пор, пока DB не начнёт деградировать. Для больших данных — do_xcom_push=False, а результат складывайте в S3/object store.

Краткий контракт собственного operator

Если пишете custom operator для production, чек-лист:

1. Наследник BaseOperator (или BaseSensorOperator для sensors).
2. template_fields объявлен явно (даже если пустой).
3. __init__ принимает **kwargs и вызывает super().__init__(**kwargs).
4. execute(self, context) — единственный обязательный метод.
5. on_kill(self) реализован, если operator управляет внешним ресурсом.
6. do_xcom_push=False если return value большой.
7. Все network calls используют hook (наследник BaseHook), не raw connections.
8. Логирование через self.log (это logging.Logger), не print и не свой logger.