ddadmin
39ff0b5bb4
- Зачем: - реализованы требования issue #153 к мосту между кукой и пользователем. - Что: - добавлены физическая карта, точное представление и полный пересчёт через GLOBAL JOIN. - задача identity_map поставлена после сборки заказов в dds_transform. - устройство и живые проверки карты записаны в документации хранилища. - Проверка: - make lint, make smoke, make check-clickhouse, make check-services. - три запуска dds_transform: 4770 физических строк и 1590 уникальных пар в _v.
247 lines
11 KiB
Python
247 lines
11 KiB
Python
"""Трансформации слоя DDS: сборка моделей.
|
|
|
|
Заказ и сессия собираются заменой дневных партиций. Карта идентичностей
|
|
пересчитывается целиком после заказа, потому что читает его текущую модель.
|
|
|
|
Расписания у дага нет и параметров тоже: объём каждого прогона он выводит сам
|
|
из состояния слоёв и позиции мира, поэтому один и тот же запуск одинаково
|
|
верен из триггера дневного конвейера, с кнопки менти и после падения
|
|
(docs/architecture/etl.md, «Объём прогона»).
|
|
|
|
Заказ и сессия гасят повтор заменой партиции целиком: день собирается в
|
|
таблицу-двойник и подменяет собой партицию цели. Карта пишет полный набор пар
|
|
новой версией, а точное чтение даёт FINAL. Замена партиции разобрана в
|
|
sql/dds/order_replace.sql; точное чтение карты описано в
|
|
sql/ddl/50-dds-views.sql.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import datetime
|
|
import logging
|
|
from pathlib import Path
|
|
|
|
from airflow.exceptions import AirflowException
|
|
from airflow.providers.clickhousedb.hooks.clickhouse import ClickHouseHook
|
|
from airflow.providers.common.sql.operators.sql import SQLExecuteQueryOperator
|
|
from airflow.sdk import Variable, dag, task, task_group
|
|
|
|
# Пропуск приезжает из Task SDK: в airflow.exceptions он объявлен устаревшим
|
|
# и говорит об этом в журнал задачи (проверено на 3.3.0).
|
|
from airflow.sdk.exceptions import AirflowSkipException
|
|
|
|
START_DATE = datetime.datetime(2026, 1, 1, tzinfo=datetime.UTC)
|
|
|
|
SQL_ROOT = Path("/opt/airflow/sql")
|
|
CLICKHOUSE_CONNECTION = "clickhouse_default"
|
|
|
|
# Позиция на оси модельного времени: номер первого несыгранного дня. Её ставит
|
|
# пульт мира, и только она отличает прожитый день от живого — зачем это
|
|
# сессиям, разобрано в sql/dds/session_scope.sql.
|
|
WORLD_POSITION = "world_position"
|
|
|
|
|
|
def _world_position() -> 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 < 1:
|
|
raise AirflowException(
|
|
f"мир не готов: world_position={position}; "
|
|
"выполните make up или make rebuild-storage"
|
|
)
|
|
return position
|
|
|
|
|
|
def _days(first_day: datetime.date, last_day: datetime.date) -> list[str]:
|
|
"""Дни отрезка строками — в таком виде их ждёт замена партиции."""
|
|
return [
|
|
(first_day + datetime.timedelta(days=shift)).isoformat()
|
|
for shift in range((last_day - first_day).days + 1)
|
|
]
|
|
|
|
|
|
@task
|
|
def replacements(scope: dict[str, object]) -> list[dict[str, object]]:
|
|
"""Разложить дни в аргументы размноженной задачи замены.
|
|
|
|
Шаг существует ради ограничения Airflow: размножать задачу можно
|
|
только по целому возврату другой задачи — ни по ключу словаря, ни по
|
|
карте поверх него. Отсюда отдельный возврат под один список.
|
|
|
|
День едет в запрос подстановкой Jinja, а не параметром ClickHouse:
|
|
в ALTER … PARTITION сервер параметр не подставляет — замерено,
|
|
отвечает синтаксической ошибкой. Значение при этом не внешнее, его
|
|
вернул сам ClickHouse запросом объёма.
|
|
"""
|
|
return [{"params": {"day": day}} for day in scope["days"]]
|
|
|
|
|
|
@dag(
|
|
dag_id="dds_transform",
|
|
schedule=None,
|
|
start_date=START_DATE,
|
|
is_paused_upon_creation=False,
|
|
# Два прогона разом собирали бы один и тот же двойник: второй затёр бы
|
|
# чужую сборку между вставкой и заменой партиций.
|
|
max_active_runs=1,
|
|
# Потолок на задачи в прогоне. Размноженная замена норовит выстрелить
|
|
# всеми днями сразу, а каждая задача — свой процесс исполнителя, и
|
|
# планировщику отведено 640 МиБ (compose.yaml). Замерено: тринадцать дней
|
|
# двух сущностей без потолка кончаются SIGKILL от контрольной группы.
|
|
max_active_tasks=3,
|
|
template_searchpath=str(SQL_ROOT),
|
|
tags=["dds"],
|
|
)
|
|
def dds_transform():
|
|
"""Пересобрать сущности DDS по текущему состоянию источников."""
|
|
|
|
@task_group(group_id="order")
|
|
def order():
|
|
"""Заказ: дышащее окно плюс дни, которых в слое ещё нет."""
|
|
|
|
@task
|
|
def scope() -> dict[str, object]:
|
|
"""Спросить у хранилища, какие дни собирает этот запуск."""
|
|
query = (SQL_ROOT / "dds" / "order_scope.sql").read_text(encoding="utf-8")
|
|
hook = ClickHouseHook(clickhouse_conn_id=CLICKHOUSE_CONNECTION)
|
|
row = hook.get_first(query)
|
|
# Пустой ответ означает пустой источник: собирать нечего, и
|
|
# зеленеть на этом нельзя — прогон встаёт с внятной причиной.
|
|
if row is None:
|
|
raise AirflowException(
|
|
"в ods.order нет ни одного заказа: собирать нечего; "
|
|
"проверьте, дошёл ли слепок до хранилища"
|
|
)
|
|
first_day, last_day = row
|
|
days = _days(first_day, last_day)
|
|
logging.info(
|
|
"объём пересборки: %s … %s, дней %d",
|
|
first_day,
|
|
last_day,
|
|
len(days),
|
|
)
|
|
return {
|
|
"first_day": first_day.isoformat(),
|
|
"last_day": last_day.isoformat(),
|
|
"days": days,
|
|
}
|
|
|
|
days = scope()
|
|
|
|
rebuild = SQLExecuteQueryOperator(
|
|
task_id="rebuild",
|
|
conn_id=CLICKHOUSE_CONNECTION,
|
|
sql="dds/order_rebuild.sql",
|
|
parameters={
|
|
"load_id": "{{ run_id }}",
|
|
"first_day": days["first_day"],
|
|
"last_day": days["last_day"],
|
|
},
|
|
split_statements=True,
|
|
do_xcom_push=False,
|
|
)
|
|
|
|
# Замена идёт по партиции за раз — другой формы у операции нет. День
|
|
# становится отдельной задачей: в интерфейсе видно, сколько дней несёт
|
|
# прогон, а повтор после отказа стоит одного дня, а не всей пересборки.
|
|
replace = SQLExecuteQueryOperator.partial(
|
|
task_id="replace",
|
|
conn_id=CLICKHOUSE_CONNECTION,
|
|
sql="dds/order_replace.sql",
|
|
do_xcom_push=False,
|
|
).expand_kwargs(replacements(days))
|
|
|
|
rebuild >> replace
|
|
|
|
@task_group(group_id="session")
|
|
def session():
|
|
"""Сессия: дни, которые мир уже прожил, а слой ещё не собрал."""
|
|
|
|
@task
|
|
def scope() -> dict[str, object]:
|
|
"""Спросить у хранилища и у позиции мира, какие дни собирает запуск."""
|
|
query = (SQL_ROOT / "dds" / "session_scope.sql").read_text(encoding="utf-8")
|
|
hook = ClickHouseHook(clickhouse_conn_id=CLICKHOUSE_CONNECTION)
|
|
row = hook.get_first(query, parameters={"position": _world_position()})
|
|
if row is None:
|
|
raise AirflowException(
|
|
"в ods.event нет ни одного события: собирать нечего; "
|
|
"проверьте, дошёл ли поток до хранилища"
|
|
)
|
|
first_day, last_day = row
|
|
# Пропуск, а не отказ: слой полон по последний прожитый день, а
|
|
# сегодняшний ещё идёт. Это обычное состояние повторного прогона,
|
|
# и в интерфейсе оно видно пропущенной группой.
|
|
if first_day is None:
|
|
raise AirflowSkipException(
|
|
f"сессии собраны по {last_day} включительно, "
|
|
"новых прожитых дней нет"
|
|
)
|
|
days = _days(first_day, last_day)
|
|
logging.info(
|
|
"объём сборки: %s … %s, дней %d",
|
|
first_day,
|
|
last_day,
|
|
len(days),
|
|
)
|
|
return {
|
|
"first_day": first_day.isoformat(),
|
|
"last_day": last_day.isoformat(),
|
|
"days": days,
|
|
}
|
|
|
|
days = scope()
|
|
|
|
rebuild = SQLExecuteQueryOperator(
|
|
task_id="rebuild",
|
|
conn_id=CLICKHOUSE_CONNECTION,
|
|
sql="dds/session_rebuild.sql",
|
|
parameters={
|
|
"load_id": "{{ run_id }}",
|
|
"first_day": days["first_day"],
|
|
"last_day": days["last_day"],
|
|
},
|
|
split_statements=True,
|
|
do_xcom_push=False,
|
|
)
|
|
|
|
replace = SQLExecuteQueryOperator.partial(
|
|
task_id="replace",
|
|
conn_id=CLICKHOUSE_CONNECTION,
|
|
sql="dds/session_replace.sql",
|
|
do_xcom_push=False,
|
|
).expand_kwargs(replacements(days))
|
|
|
|
rebuild >> replace
|
|
|
|
orders = order()
|
|
session()
|
|
|
|
identity_map = SQLExecuteQueryOperator(
|
|
task_id="identity_map",
|
|
conn_id=CLICKHOUSE_CONNECTION,
|
|
sql="dds/identity_map_rebuild.sql",
|
|
parameters={"load_id": "{{ run_id }}"},
|
|
do_xcom_push=False,
|
|
)
|
|
|
|
orders >> identity_map
|
|
|
|
|
|
dds_transform()
|