docs(warehouse): объяснены расчет строк и замена дневных разделов

Зачем: выражения «собираются в донор» и «публикуются дни» требовали
знать устройство конвейера до чтения объяснения.

Что: описаны назначение промежуточных таблиц и последовательность
расчета, записи и замены разделов. Перечитаны остальные объяснения:
выбор дней, разворачивание позиций заказа, проверка сессий и повтор запуска.

Проверка: make lint, git diff --check, локальные ссылки и якоря.
SQL без комментариев и AST обоих дагов без строк документации не изменились.

Refs: #249
This commit is contained in:
ddadmin committed 2026-09-13 15:16:26 -04:00
1 parent 5f317ba8d2
commit 086007fbd4
12 files changed
+216 -166

No files matched your search

+19 -22
View File
@@ -1,8 +1,12 @@
"""Трансформации DDS: заказы, сессии и карта идентичностей.
Заказы и сессии собираются в донор, затем публикуются заменой дневных партиций.
Карта идентичностей пересчитывается целиком после заказов и записывается новой
версией; публичное представление выбирает текущие строки через FINAL.
Запросы рассчитывают заказы и сессии за выбранные дни и записывают результат
в промежуточные таблицы. Затем готовые дневные разделы этих таблиц заменяют
разделы основных таблиц. Пока идет расчет, читатели видят прежние данные.
Карта идентичностей пересчитывается целиком после заказов. Новые строки
добавляются с меткой времени сборки; представление с FINAL выбирает последнюю
версию каждой строки.
Даг запускается без расписания и параметров. Даты он выбирает по состоянию
таблиц и позиции мира. Новый запуск заново определяет диапазон: наличие даты
@@ -252,16 +256,11 @@ def _days(first_day: datetime.date, last_day: datetime.date) -> list[str]:
@task
def replacements(scope: dict[str, object]) -> list[dict[str, object]]:
"""Разложить дни в аргументы размноженной задачи замены.
"""Подготовить параметры отдельной задачи replace для каждого дня.
Шаг существует ради ограничения Airflow: размножать задачу можно
только по целому возврату другой задачи — ни по ключу словаря, ни по
карте поверх него. Отсюда отдельный возврат под один список.
День едет в запрос подстановкой Jinja, а не параметром ClickHouse:
в ALTER … PARTITION сервер параметр не подставляет — замерено,
отвечает синтаксической ошибкой. Значение при этом не внешнее, его
вернул сам ClickHouse запросом объёма.
expand_kwargs получает список словарей и создает по одному экземпляру
задачи на каждый словарь. Дата подставляется в SQL шаблоном Jinja:
параметр ClickHouse в ALTER ... PARTITION здесь не поддерживается.
"""
return [{"params": {"day": day}} for day in scope["days"]]
@@ -271,8 +270,8 @@ def replacements(scope: dict[str, object]) -> list[dict[str, object]]:
schedule=None,
start_date=START_DATE,
is_paused_upon_creation=False,
# Два прогона разом собирали бы один и тот же двойник: второй затёр бы
# чужую сборку между вставкой и заменой партиций.
# Одновременные запуски используют одни промежуточные таблицы.
# Второй мог бы очистить результат первого до замены партиций.
max_active_runs=1,
template_searchpath=str(SQL_ROOT),
tags=["dds"],
@@ -301,8 +300,7 @@ def dds_transform():
query = (SQL_ROOT / "dds" / "order_scope.sql").read_text(encoding="utf-8")
hook = ClickHouseHook(clickhouse_conn_id=CLICKHOUSE_CONNECTION)
row = hook.get_first(query)
# Пустой ответ означает пустой источник: собирать нечего, и
# зеленеть на этом нельзя — прогон встаёт с внятной причиной.
# Без заказов ODS расчет невозможен: задача сообщает об ошибке.
if row is None:
raise AirflowException(
"в ods.order нет ни одного заказа: собирать нечего; "
@@ -337,9 +335,9 @@ def dds_transform():
do_xcom_push=False,
)
# Замена идёт по партиции за раз — другой формы у операции нет. День
# становится отдельной задачей: в интерфейсе видно, сколько дней несёт
# прогон, а повтор после отказа стоит одного дня, а не всей пересборки.
# На каждый день создается отдельная задача замены партиции.
# Неудачную замену можно повторить, пока промежуточная таблица
# сохраняет рассчитанные строки этого запуска.
replace = SQLExecuteQueryOperator.partial(
task_id="replace",
conn_id=CLICKHOUSE_CONNECTION,
@@ -365,9 +363,8 @@ def dds_transform():
"проверьте, дошёл ли поток до хранилища"
)
first_day, last_day = row
# Пропуск, а не отказ: слой полон по последний прожитый день, а
# сегодняшний ещё идёт. Это обычное состояние повторного прогона,
# и в интерфейсе оно видно пропущенной группой.
# Для каждого прожитого дня уже есть строки сессий.
# Новых дней для расчета нет: группа задач будет пропущена.
if first_day is None:
raise AirflowSkipException(
f"сессии собраны по {last_day} включительно, "