clickstream-data-platform/dags/dds_transform.py

"""Трансформации слоя DDS: сборка модели заменой дневных партиций.

Даг собирает модель поверх принятого ODS. Сущностей у него две — заказ и
сессия, зависимости по данным между ними нет; карта идентичностей встанет
соседней группой задач.

Расписания у дага нет и параметров тоже: объём каждого прогона он выводит сам
из состояния слоёв и позиции мира, поэтому один и тот же запуск одинаково
верен из триггера дневного конвейера, с кнопки менти и после падения
(docs/architecture/etl.md, «Объём прогона»).

Повтор прогона гасит не колонка версии, а замена партиции целиком: день
собирается в таблицу-двойник и подменяет собой партицию цели. Почему слой
устроен так — sql/ddl/45-dds-tables.sql; механика самой подмены —
sql/dds/order_replace.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

    order()
    session()


dds_transform()