"""Даги модельного времени: ими двигают ось мира. Их три, и они двух родов. **Шаговые даги** играют день и запускаются руками: `world_next_day` — пачкой, без пауз, сколько дней попросили; `world_live_day` — один день в темпе модельного времени. **Выключатель** `world_live` своей работы не делает: он тикает по расписанию и дёргает шаговый даг живого дня, дожидаясь конца. Снят с паузы — мир едет день за днём; поставлен на паузу — встал на границе модельных суток. Сыграть день — половина работы. Вторая половина: отправить слепок заказов и дождаться, пока хранилище его заберёт. Поэтому у обоих шаговых дагов за проигрышем идут отправка слепка тем же контейнером генератора и ждущий триггер дага приёма `orders_ingest`. Живой день после установки позиции дожидается еще и `etl_pipeline`. Пачка только разгоняет источники: слои после нее менти догоняет ручным запуском. Разделение не косметическое. Расписание на самом шаговом даге заставило бы кнопку паузы значить две вещи разом — «мир не едет сам» и «даг выключен», — а шаговый даг при этом выглядел бы в списке выключенным, хотя нажимают его каждый день. Календарь Airflow к оси мира отношения не имеет: какой день играть, шаговый даг спрашивает у переменной, а не у логической даты прогона. Решения и доводы целиком — ADR 0009. """ from __future__ import annotations import datetime import logging import os from airflow.exceptions import AirflowException from airflow.providers.docker.operators.docker import DockerOperator from airflow.providers.standard.operators.trigger_dagrun import TriggerDagRunOperator from airflow.sdk import Param, Variable, dag, get_current_context, task # Факты стенда — образ генератора, сеть, адрес брокера, топик и размер # стартового мира — приходят окружением, и называет их compose: тот же, что # называет их разовой службе генератора. GENERATOR_IMAGE = os.environ["GENERATOR_IMAGE"] STAND_NETWORK = os.environ["STAND_NETWORK"] GENERATOR_ENVIRONMENT = { "KAFKA_BOOTSTRAP_SERVERS": os.environ["KAFKA_BOOTSTRAP_SERVERS"], "KAFKA_TOPIC": os.environ["KAFKA_TOPIC"], } # Топик слепка называется аргументом, а не окружением: KAFKA_TOPIC выше — топик # событий, и промолчи мы, заказы уехали бы к ним. ORDERS_TOPIC = os.environ["KAFKA_ORDERS_TOPIC"] STARTING_DAYS = int(os.environ["WORLD_STARTING_DAYS"]) # Позиция на оси: номер первого несыгранного дня. До конца инициализации # шаговые даги не имеют права угадывать её — состояние объяснено в ADR 0013. # # Позиция ставится, а не увеличивается. Наложись один прогон на другой, худшее # при таком правиле — сыгранный дважды день, а повтор схлопнет # ReplacingMergeTree. Увеличение молча съело бы день, и в мире осталась бы # дыра, которой никто не заметит. WORLD_POSITION = "world_position" # Тик выключателя. Каденцию задаёт не он, а сама длина живого дня — около # двадцати четырёх минут: тик только спрашивает «не пора ли снова». LIVE_TICK = datetime.timedelta(minutes=25) # Какой день играть, шаговый даг узнаёт у первой задачи. Спрашивают её все # генераторные шаги: слепок снимается с той же позиции, что и проигрыш. PLAYED_DAY = "{{ ti.xcom_pull(task_ids='first_unplayed_day') }}" START_DATE = datetime.datetime(2026, 1, 1, tzinfo=datetime.UTC) TAGS = ["модельное время"] @task def first_unplayed_day() -> int: """Номер дня, с которого играть.""" raw_position = Variable.get(WORLD_POSITION, default=None) if raw_position is None: raise AirflowException( "мир ещё не создан: выполните make up и дождитесь world_initialize" ) try: position = int(raw_position) except (TypeError, ValueError) as error: raise AirflowException( f"world_position={raw_position!r} не является номером дня; " "выполните make rebuild-storage" ) from error if position < STARTING_DAYS: action = ( "выполните make up" if position == 0 else "выполните make rebuild-storage" ) raise AirflowException(f"мир не готов: world_position={position}; {action}") return position def _generator(task_id: str, command: list[str]) -> DockerOperator: """Задача, зовущая генератор в его каноническом контейнере. Внутрь образа Airflow генератор не поставить: он требует Python 3.14, а образ несёт 3.13. Да и обещание побайтовой воспроизводимости дано для зафиксированного образа генератора — держится оно только там. """ return DockerOperator( task_id=task_id, image=GENERATOR_IMAGE, command=command, network_mode=STAND_NETWORK, environment=GENERATOR_ENVIRONMENT, # Контейнер убирается в любом исходе: вывод генератора оператор уже # перелил в журнал задачи, а мёртвые контейнеры копить незачем. auto_remove="force", # По умолчанию оператор монтирует контейнеру временный каталог. Здесь # это ловушка: путь он заводит внутри Airflow, а монтирует демон с # хоста, где такого пути нет. Генератору временный каталог не нужен. mount_tmp_dir=False, ) def _ingest_orders() -> TriggerDagRunOperator: """Задача забора: дёрнуть даг приёма и дождаться, чем он кончился. Ожидание здесь несущее. Без него шаговый даг позеленел бы, не узнав, доехал ли слепок, и позиция мира ушла бы вперёд хранилища — а зелёный конец графа не должен переживать отказ выше (ADR 0003). """ return TriggerDagRunOperator( task_id="trigger_orders_ingest", trigger_dag_id="orders_ingest", wait_for_completion=True, # Умолчание — минута опроса, а весь прогон шагового дага пачкой короче. poke_interval=10, ) @dag( dag_id="world_next_day", schedule=None, start_date=START_DATE, is_paused_upon_creation=False, max_active_runs=1, tags=TAGS, params={ "days": Param( 1, type="integer", minimum=1, maximum=7, title="Число модельных дней для генерации", # description_md рендерится подсказкой прямо в форме запуска # (params.rst, Airflow 3.3), чтобы правило было видно сразу. description_md=( "Создать данные за следующие 1–7 модельных дней; по умолчанию " "за один день. События отправляются без пауз. Номер первого дня " "берется из `world_position` (Admin → Variables). Поля Logical " "date и Run ID не меняют выбор дня. После завершения запустите " "`etl_pipeline`, чтобы обновить DDS и витрины DM." ), ) }, ) def world_next_day(): """Создать данные за следующие модельные дни без пауз между событиями. Параметр `days` задает число дней: от 1 до 7, по умолчанию 1. Первый день берется из `world_position`. После загрузки событий и слепков заказов запустите `etl_pipeline` вручную, чтобы обновить DDS и витрины DM. """ @task def remember_played(first_day: int) -> None: """Позиция ставится по сыгранным дням и только по успеху.""" days = get_current_context()["params"]["days"] position = first_day + days Variable.set(WORLD_POSITION, str(position)) logging.info( "данные за модельные дни %s–%s отправлены; world_position=%s", first_day, position - 1, position, ) first_day = first_unplayed_day() played = _generator( "play_days", ["batch", "--day", PLAYED_DAY, "--days", "{{ params.days }}"], ) # Разгон на N дней отправляет N слепков — по одному за сыгранный день, и # каждый со своим сдвигом: прогон дня D везёт слепок дня D−1. Забор при # этом остаётся один. sent = _generator( "send_snapshots", [ "snapshot", "--day", PLAYED_DAY, "--days", "{{ params.days }}", "--topic", ORDERS_TOPIC, ], ) first_day >> played >> sent >> _ingest_orders() >> remember_played(first_day) @dag( dag_id="world_live_day", schedule=None, start_date=START_DATE, is_paused_upon_creation=False, max_active_runs=1, tags=TAGS, ) def world_live_day(): """Постепенно отправить события за следующий модельный день. При ускорении ×60 отправка занимает около 24 реальных минут. Номер дня берется из `world_position`; Logical date его не меняет. Затем даг отправляет слепок заказов и ждет его загрузки и завершения `etl_pipeline`. Для автоматического запуска следующих дней снимите с паузы `world_live`. """ @task def remember_played(first_day: int) -> None: """Позиция ставится по сыгранному дню и только по успеху.""" Variable.set(WORLD_POSITION, str(first_day + 1)) logging.info( "данные за модельный день %s отправлены; world_position=%s", first_day, first_day + 1, ) first_day = first_unplayed_day() played = _generator("play_day", ["live", "--day", PLAYED_DAY]) # Слепок уезжает пачкой и после дня, а не в темпе: у выгрузки бэкенда темпа # нет вовсе — она снимается на границе суток целиком. sent = _generator( "send_snapshot", ["snapshot", "--day", PLAYED_DAY, "--topic", ORDERS_TOPIC], ) position = remember_played(first_day) etl = TriggerDagRunOperator( task_id="trigger_etl_pipeline", trigger_dag_id="etl_pipeline", wait_for_completion=True, poke_interval=10, ) first_day >> played >> sent >> _ingest_orders() >> position >> etl @dag( dag_id="world_live", schedule=LIVE_TICK, start_date=START_DATE, catchup=False, is_paused_upon_creation=True, max_active_runs=1, tags=TAGS, ) def world_live(): """Автоматически запускать модельные дни, пока даг снят с паузы. Запускает `world_live_day` и ждет его завершения. Пауза останавливает запуск следующих дней; уже запущенный день завершится. Если он завершился с ошибкой, этот даг тоже завершится с ошибкой: откройте дочерний запуск и журнал неуспешной задачи. """ TriggerDagRunOperator( task_id="trigger_live_day", trigger_dag_id="world_live_day", wait_for_completion=True, ) world_next_day() world_live_day() world_live()