feat(airflow): пульт мира объясняет ось дней в форме и журнале #175

Merged
ddmitry merged 2 commits from 111-world-date into main 2026-08-28 12:16:40 +03:00
3 changed files with 34 additions and 6 deletions
Showing only changes of commit c2c4401a16 - Show all commits

No files matched your search

+6
View File
@@ -262,6 +262,12 @@ uv run --project generator python -m clickstream_generator batch \
`0` — очистка закончена, а начальная генерация ещё не завершена. Работники `0` — очистка закончена, а начальная генерация ещё не завершена. Работники
начинают играть только с позиции `8` и больше. начинают играть только с позиции `8` и больше.
Форма запуска работника спрашивает только своё — сколько дней играть.
Служебные поля Airflow в ней — «Logical date» и «Run ID» за створкой
«Advanced Options» — к оси мира отношения не имеют: какой день пойдёт первым,
смотрите в `world_position`, а какой сыгран — в журнале шага
`remember_played`.
Выключатель создаётся на паузе. Снимите — мир поедет сам; поставите обратно — Выключатель создаётся на паузе. Снимите — мир поедет сам; поставите обратно —
встанет на границе модельных суток, доиграв начатый день. Форма пульта и доводы встанет на границе модельных суток, доиграв начатый день. Форма пульта и доводы
целиком — [ADR 0009](docs/adr/0009-world-control.md). целиком — [ADR 0009](docs/adr/0009-world-control.md).
+25 -3
View File
@@ -28,6 +28,7 @@
from __future__ import annotations from __future__ import annotations
import datetime import datetime
import logging
import os import os
from airflow.exceptions import AirflowException from airflow.exceptions import AirflowException
@@ -141,7 +142,19 @@ def _ingest_orders() -> TriggerDagRunOperator:
tags=TAGS, tags=TAGS,
params={ params={
"days": Param( "days": Param(
1, type="integer", minimum=1, maximum=7, title="Сколько дней прожить" 1,
type="integer",
minimum=1,
maximum=7,
title="Сколько дней прожить",
# description_md рендерится подсказкой прямо в форме запуска
# (params.rst, Airflow 3.3) — объяснение стоит там, где путают.
description_md=(
"Какие именно дни — решает не форма: первый несыгранный "
"день хранит переменная `world_position` (Admin → "
"Variables). Служебные поля Airflow — Logical date, Run "
"ID — к оси мира отношения не имеют."
),
) )
}, },
) )
@@ -149,14 +162,20 @@ def world_next_day():
"""Прожить следующие дни пачкой, без пауз. """Прожить следующие дни пачкой, без пауз.
Запускается руками. День по умолчанию один, но разгон вперёд идёт одним Запускается руками. День по умолчанию один, но разгон вперёд идёт одним
нажимом, а не десятью: сколько дней играть — параметр запуска. нажимом, а не десятью: сколько дней играть — параметр запуска. Какие
именно дни, знает переменная `world_position`, а не форма: логическая
дата прогона к оси мира отношения не имеет.
""" """
@task @task
def remember_played(first_day: int) -> None: def remember_played(first_day: int) -> None:
"""Позиция ставится по сыгранным дням и только по успеху.""" """Позиция ставится по сыгранным дням и только по успеху."""
days = get_current_context()["params"]["days"] days = get_current_context()["params"]["days"]
Variable.set(WORLD_POSITION, str(first_day + days)) position = first_day + days
Variable.set(WORLD_POSITION, str(position))
logging.info(
"сыграны дни %s–%s, позиция теперь %s", first_day, position - 1, position
)
first_day = first_unplayed_day() first_day = first_unplayed_day()
played = _generator( played = _generator(
@@ -196,12 +215,15 @@ def world_live_day():
Ускорение ×60: модельные сутки укладываются примерно в двадцать четыре Ускорение ×60: модельные сутки укладываются примерно в двадцать четыре
реальные минуты, и суточная волна разворачивается на глазах. Запускается реальные минуты, и суточная волна разворачивается на глазах. Запускается
руками; чтобы мир жил так день за днём сам, есть выключатель `world_live`. руками; чтобы мир жил так день за днём сам, есть выключатель `world_live`.
Какой день пойдёт, знает переменная `world_position`: логическая дата
прогона к оси мира отношения не имеет.
""" """
@task @task
def remember_played(first_day: int) -> None: def remember_played(first_day: int) -> None:
"""Позиция ставится по сыгранному дню и только по успеху.""" """Позиция ставится по сыгранному дню и только по успеху."""
Variable.set(WORLD_POSITION, str(first_day + 1)) Variable.set(WORLD_POSITION, str(first_day + 1))
logging.info("сыгран день %s, позиция теперь %s", first_day, first_day + 1)
first_day = first_unplayed_day() first_day = first_unplayed_day()
played = _generator("play_day", ["live", "--day", PLAYED_DAY]) played = _generator("play_day", ["live", "--day", PLAYED_DAY])
+3 -3
View File
@@ -137,9 +137,9 @@ ODS и ещё не собранный в DDS, затем при необходи
## Не входит ## Не входит
- Партиционная переобработка — переиграть произвольный прошлый день или - Партиционная переобработка — переиграть произвольный прошлый день или
диапазон, включая форму необязательных параметров того же дага, — этап 5 диапазон, включая форму необязательных параметров того же дага.
(#7); туда же стык #111 про дату модельного мира и `logical_date`. - Инкрементальный ETL вместо пересборки — отклонён (#2, wontfix); учебная
- Инкрементальный ETL вместо пересборки — техдолг #2. сторона темы — идея лабы #149.
- Состав витрин `dm_transform` и проверок `dq_check` — - Состав витрин `dm_transform` и проверок `dq_check` —
[дока слоя DM](dm.md). [дока слоя DM](dm.md).