diff --git a/compose.yaml b/compose.yaml index 56671e3..60a07de 100644 --- a/compose.yaml +++ b/compose.yaml @@ -310,7 +310,10 @@ services: - ./infra/airflow/init.sh:/opt/airflow/init.sh:ro - airflow_logs:/opt/airflow/logs - airflow_auth:/opt/airflow/auth - mem_limit: 640m + # LocalExecutor запускает здесь до четырех процессов задач одновременно. + # 640 МиБ дали OOM, а 1 ГиБ - давление на memory.max при штатном dq_check; + # 1,5 ГиБ оставляет запас при AIRFLOW__CORE__PARALLELISM=4. + mem_limit: 1536m healthcheck: test: ["CMD-SHELL", "curl -sf http://127.0.0.1:8974/health >/dev/null"] interval: 10s diff --git a/dags/dds_transform.py b/dags/dds_transform.py index 46ef624..59c4793 100644 --- a/dags/dds_transform.py +++ b/dags/dds_transform.py @@ -279,11 +279,6 @@ def replacements(scope: dict[str, object]) -> list[dict[str, object]]: # Два прогона разом собирали бы один и тот же двойник: второй затёр бы # чужую сборку между вставкой и заменой партиций. max_active_runs=1, - # Потолок на задачи в прогоне. Размноженная замена норовит выстрелить - # всеми днями сразу, а каждая задача — свой процесс исполнителя, и - # планировщику отведено 640 МиБ (compose.yaml). Замерено: тринадцать дней - # двух сущностей без потолка кончаются SIGKILL от контрольной группы. - max_active_tasks=3, template_searchpath=str(SQL_ROOT), tags=["dds"], ) diff --git a/dags/dm_transform.py b/dags/dm_transform.py index d2b38a4..8e7210c 100644 --- a/dags/dm_transform.py +++ b/dags/dm_transform.py @@ -76,9 +76,6 @@ def replacements(scope: dict[str, object]) -> list[dict[str, object]]: start_date=START_DATE, is_paused_upon_creation=False, max_active_runs=1, - # Каждая задача LocalExecutor - отдельный процесс. Четыре процесса уже - # дали SIGKILL при лимите scheduler 640 МиБ; три - измеренный потолок DDS. - max_active_tasks=3, template_searchpath=str(SQL_ROOT), tags=["dm"], ) @@ -136,11 +133,6 @@ def dm_transform(): conn_id=CLICKHOUSE_CONNECTION, sql="dm/revenue_daily_replace.sql", do_xcom_push=False, - # Airflow 3 применяет этот лимит и к размноженным экземплярам - # (сверено через Context7). Каждая замена - отдельный процесс - # LocalExecutor; три разом исчерпали память контейнера scheduler, - # а самим дням параллельность ничего не дает. - max_active_tis_per_dag=1, ).expand_kwargs(replacements(days)) rebuild >> replace @@ -198,8 +190,6 @@ def dm_transform(): conn_id=CLICKHOUSE_CONNECTION, sql="dm/purchase_vs_orders_replace.sql", do_xcom_push=False, - # Причина единицы разобрана у revenue_daily.replace выше. - max_active_tis_per_dag=1, ).expand_kwargs(replacements(days)) rebuild >> replace diff --git a/dags/dq_check.py b/dags/dq_check.py new file mode 100644 index 0000000..accefcf --- /dev/null +++ b/dags/dq_check.py @@ -0,0 +1,313 @@ +"""Проверки качества стендовых моделей и витрин. + +Пять проверок сначала собирают общий донор. После его полной сборки дневные +партиции публикуются одной серией замен, и только затем именованные задачи +утверждают результат. Расхождение поэтому остаётся в ``dm.dq_summary_v`` до +красного завершения дага; ошибка сборки прежнюю сводку не меняет. +""" + +from __future__ import annotations + +import datetime +import json +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, get_current_context, task, task_group + +START_DATE = datetime.datetime(2026, 1, 1, tzinfo=datetime.UTC) + +SQL_ROOT = Path("/opt/airflow/sql") +INVENTORY_PATH = Path("/opt/airflow/world/data/world-inventory.json") +CLICKHOUSE_CONNECTION = "clickhouse_default" +WORLD_POSITION = "world_position" +DETAIL_LIMIT = 20 + +SUMMARY_COLUMNS = [ + "data_date", + "check_name", + "reference_rows", + "actual_rows", + "failed_rows", + "status", + "_load_id", + "_load_ts", +] + + +def _clickhouse_hook() -> ClickHouseHook: + """Подключение, которое дожидается записи в Distributed-таблицу.""" + return ClickHouseHook( + clickhouse_conn_id=CLICKHOUSE_CONNECTION, + session_settings={"distributed_foreground_insert": 1}, + ) + + +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 _publish_summary( + check_name: str, + summaries: list[tuple[object, ...]], +) -> list[str]: + """Записать дневную сводку и вернуть малый срез готовой диагностики.""" + if not summaries: + raise AirflowException(f"{check_name}: область проверки пуста") + + diagnostics: list[str] = [] + context = get_current_context() + load_id = context["run_id"] + load_ts = datetime.datetime.now(datetime.UTC) + rows = [] + for ( + data_date, + reference_rows, + actual_rows, + failed_rows, + details, + ) in summaries: + rows.append( + ( + data_date, + check_name, + reference_rows, + actual_rows, + failed_rows, + "pass" if failed_rows == 0 else "fail", + load_id, + load_ts, + ) + ) + diagnostics.extend(details[: DETAIL_LIMIT - len(diagnostics)]) + + _clickhouse_hook().bulk_insert_rows( + "dm.dq_summary_stage_dist", + rows, + column_names=SUMMARY_COLUMNS, + ) + return diagnostics + + +def _build_sql_check( + check_name: str, + sql_name: str, + parameters: dict[str, object] | None = None, +) -> list[str]: + """Выполнить SQL-сверку и записать ее дневные итоги в донор.""" + query = (SQL_ROOT / "dq" / sql_name).read_text(encoding="utf-8") + summaries = _clickhouse_hook().get_records(query, parameters=parameters) + return _publish_summary(check_name, summaries) + + +@task +def build_sql_check(check_name: str, sql_name: str) -> list[str]: + """Собрать проверку, которой не нужны координаты модельного мира.""" + return _build_sql_check(check_name, sql_name) + + +@task +def build_sessions_vs_reference() -> list[str]: + """Сверить сессии только в пределах прожитых дней мира.""" + return _build_sql_check( + "sessions_vs_reference", + "sessions_vs_reference.sql", + parameters={"position": _world_position()}, + ) + + +@task +def build_classes_vs_inventory() -> list[str]: + """Сопоставить классы закрытых дней с техническим оракулом генератора.""" + inventory = json.loads(INVENTORY_PATH.read_text(encoding="utf-8")) + expected = { + (datetime.date.fromisoformat(day["date"]), mismatch_class): count + for day in inventory["days"] + if "orders" in day + for mismatch_class, count in day["orders"].items() + } + if not expected: + raise AirflowException("classes_vs_inventory: в описи нет закрытых дней") + + query = (SQL_ROOT / "dq" / "classes_vs_inventory.sql").read_text(encoding="utf-8") + actual = { + (data_date, mismatch_class): count + for data_date, mismatch_class, count in _clickhouse_hook().get_records(query) + if (data_date, mismatch_class) in expected + } + summaries = [] + for data_date in sorted({day for day, _ in expected}): + details = [] + reference_rows = 0 + actual_rows = 0 + failed_rows = 0 + for (day, mismatch_class), reference_count in sorted(expected.items()): + if day != data_date: + continue + actual_key = (day, mismatch_class) + actual_count = actual.get(actual_key) + reference_rows += 1 + actual_rows += int(actual_key in actual) + if reference_count != actual_count: + failed_rows += 1 + actual_text = actual_count if actual_count is not None else "нет ключа" + details.append( + f"{day} key={mismatch_class}: " + f"эталон={reference_count}, " + f"объект={actual_text}" + ) + summaries.append((data_date, reference_rows, actual_rows, failed_rows, details)) + return _publish_summary("classes_vs_inventory", summaries) + + +@task +def publication_scope() -> list[dict[str, dict[str, str]]]: + """Вернуть дни полностью собранного донора для размноженной замены.""" + rows = _clickhouse_hook().get_records( + "SELECT DISTINCT data_date FROM dm.dq_summary_stage_dist ORDER BY data_date" + ) + if not rows: + raise AirflowException("донор dm.dq_summary пуст: публиковать нечего") + return [{"params": {"day": data_date.isoformat()}} for (data_date,) in rows] + + +@task +def drop_stale_partitions() -> None: + """Убрать дни, исчезнувшие после починки с обеих сторон сверки.""" + hook = _clickhouse_hook() + stale_days = hook.get_records( + """ + SELECT DISTINCT data_date + FROM dm.dq_summary_dist + WHERE data_date GLOBAL NOT IN + (SELECT data_date FROM dm.dq_summary_stage_dist) + ORDER BY data_date + """ + ) + for (data_date,) in stale_days: + # ALTER PARTITION не принимает параметр ClickHouse. Значение пришло + # из колонки Date самого хранилища и форматируется без внешнего ввода. + hook.run( + "ALTER TABLE dm.dq_summary_rep ON CLUSTER clickstream_cluster " + f"DROP PARTITION '{data_date.isoformat()}'" + ) + + +@task +def assert_check(check_name: str, diagnostics: list[str]) -> None: + """Покрасить именованную задачу по уже опубликованной сводке.""" + failed_days = _clickhouse_hook().get_records( + """ + SELECT data_date, reference_rows, actual_rows, failed_rows + FROM dm.dq_summary_v + WHERE check_name = {check_name:String} AND failed_rows > 0 + ORDER BY data_date + """, + parameters={"check_name": check_name}, + ) + if not failed_days: + return + + for detail in diagnostics: + logging.error("%s: %s", check_name, detail) + summary = "; ".join( + f"{data_date}: эталонных ключей {reference_rows}, " + f"фактических {actual_rows}, нарушено {failed_rows}" + for data_date, reference_rows, actual_rows, failed_rows in failed_days + ) + raise AirflowException(f"{check_name}: {summary}") + + +@dag( + dag_id="dq_check", + schedule=None, + start_date=START_DATE, + is_paused_upon_creation=False, + max_active_runs=1, + template_searchpath=str(SQL_ROOT), + tags=["dq"], +) +def dq_check(): + """Собрать, опубликовать и утвердить текущее качество данных.""" + prepare = SQLExecuteQueryOperator( + task_id="prepare", + conn_id=CLICKHOUSE_CONNECTION, + sql="dq/prepare.sql", + do_xcom_push=False, + ) + + sessions = build_sessions_vs_reference() + classes = build_classes_vs_inventory.override( + task_id="build_classes_vs_inventory" + )() + revenue = build_sql_check.override(task_id="build_revenue_daily_vs_dds")( + "revenue_daily_vs_dds", "revenue_daily_vs_dds.sql" + ) + purchases = build_sql_check.override(task_id="build_purchase_vs_orders_vs_dds")( + "purchase_vs_orders_vs_dds", "purchase_vs_orders_vs_dds.sql" + ) + traffic = build_sql_check.override(task_id="build_daily_traffic_vs_dds")( + "daily_traffic_vs_dds", "daily_traffic_vs_dds.sql" + ) + builders = [sessions, classes, revenue, purchases, traffic] + prepare >> builders + + days = publication_scope() + for builder in builders: + builder >> days + + replace = SQLExecuteQueryOperator.partial( + task_id="publish", + conn_id=CLICKHOUSE_CONNECTION, + sql="dq/replace.sql", + do_xcom_push=False, + ).expand_kwargs(days) + + sessions_assertion = assert_check.override(task_id="sessions_vs_reference")( + "sessions_vs_reference", sessions + ) + classes_assertion = assert_check.override(task_id="classes_vs_inventory")( + "classes_vs_inventory", classes + ) + + @task_group(group_id="marts_vs_dds") + def assert_marts(): + return [ + assert_check.override(task_id="revenue_daily_vs_dds")( + "revenue_daily_vs_dds", revenue + ), + assert_check.override(task_id="purchase_vs_orders_vs_dds")( + "purchase_vs_orders_vs_dds", purchases + ), + assert_check.override(task_id="daily_traffic_vs_dds")( + "daily_traffic_vs_dds", traffic + ), + ] + + assertions = [sessions_assertion, classes_assertion, *assert_marts()] + remove_stale = drop_stale_partitions() + replace >> remove_stale >> assertions + + +dq_check() diff --git a/dags/etl_pipeline.py b/dags/etl_pipeline.py index 918dc2d..0771948 100644 --- a/dags/etl_pipeline.py +++ b/dags/etl_pipeline.py @@ -1,4 +1,4 @@ -"""Дневной конвейер: сначала DDS, затем DM. +"""Дневной конвейер: DDS, DM и проверка качества. Своих преобразований у дага нет. Ждущие триггеры связывают состояния: если слой краснеет, конвейер не запускает следующий слой и краснеет сам. @@ -36,8 +36,18 @@ def etl_pipeline(): wait_for_completion=True, poke_interval=10, ) + # Airflow 3.3: оператор провайдера standard с wait_for_completion=True + # поднимает AirflowException, если запущенный даг завершился красным. + # Поведение и путь импорта повторно сверены через Context7 27 августа + # 2026 года. + dq = TriggerDagRunOperator( + task_id="trigger_dq", + trigger_dag_id="dq_check", + wait_for_completion=True, + poke_interval=10, + ) - dds >> dm + dds >> dm >> dq etl_pipeline() diff --git a/docs/adr/0004-resource-limits.md b/docs/adr/0004-resource-limits.md index 648ec58..74c5433 100644 --- a/docs/adr/0004-resource-limits.md +++ b/docs/adr/0004-resource-limits.md @@ -19,16 +19,23 @@ Нодам ClickHouse — по 4 ГиБ вместо гигабайта. Дальше в этом тексте предел памяти контейнера (`mem_limit`) называется коробкой: сервер живёт внутри неё и -о самой машине ничего не знает. Остальные `mem_limit` не меняются. +о самой машине ничего не знает. На 1 августа остальные `mem_limit` не менялись. Числа, выставленные ради прежнего бюджета, остаются в силе, но лишаются статуса решения: `KAFKA_HEAP_OPTS`, `shared_buffers`, `max_connections`, -`AIRFLOW__CORE__PARALLELISM` и девять неизменённых `mem_limit`. Основания, +`AIRFLOW__CORE__PARALLELISM` и восемь оставшихся `mem_limit`. Основания, кроме отменённого, у них нет. Кто в них упрётся — меняет по первому свидетельству и этот ADR не пересматривает. Срок хранения метрик Prometheus сюда не входит: неделя истории — осознанный учебный выбор, и цена у него дисковая. +27 августа таким пределом впервые стала коробка `airflow-scheduler`: +LocalExecutor исполняет в ней до четырех процессов задач одновременно, а +640 МиБ этого не вместили. Коробка поднята до 1,5 ГиБ. Ресурсные ограничения +из отдельных дагов удалены. Общий предел параллелизма уже задает Airflow. +`max_active_runs=1` остается там, где два запуска делили бы один донор и могли +повредить публикацию. + Ресурсный довод ADR 0001 отозван. ## Почему @@ -94,8 +101,8 @@ ClickHouse сам понижает каждый до 512 МиБ, а на сте заранее, и в покое стенд занимает около 3,2 ГБ. Восемь гигабайт нужны не сумме пределов, а работе с запасом. -Остальные девять лимитов не меняются, потому что ни один из них ни разу не -сработал. Снять их скопом — то же изменение без свидетельств, каким они были +На 1 августа остальные девять лимитов не менялись, потому что ни один из них +не сработал. Снять их скопом — то же изменение без свидетельств, каким они были выставлены, только в обратную сторону. Лечение здесь не в том, чтобы стереть числа, а в том, чтобы лишить их статуса закона: число без основания меняют, когда оно мешает, и не защищают как принятое решение. @@ -151,3 +158,13 @@ to 2.00 GiB because the system has limited RAM` и такие же строки Значение `max_server_memory_usage_to_ram_ratio` = 0,9 и то, что в cgroup предел считается от коробки, а не от машины, сверены по документации ClickHouse через MCP Context7. + +`airflow-scheduler` проверен отдельно 27 августа 2026 года. На свежем +контейнере две попытки запустить четыре DQ-сверки параллельно в коробке +640 МиБ дошли ровно до `memory.max` и дали `oom_kill=1`: сначала при выгрузке +57 567 ключей сессий в Python, затем и после переноса дневной агрегации в +ClickHouse. В коробке 1 ГиБ сам `dq_check` прошел, но снова коснулся верхней +границы (`memory.events.max=57`). Полный `etl_pipeline` без локальных +ограничений задач в коробке 1,5 ГиБ завершился успешно: пик - 1 013 760 000 +байт, `memory.events.max=0`, `oom_kill=0`. Docker не резервирует разницу между +пиком и пределом, поэтому этот запас не увеличивает расход памяти в покое. diff --git a/docs/architecture/dm.md b/docs/architecture/dm.md index c84e4cf..8c0f702 100644 --- a/docs/architecture/dm.md +++ b/docs/architecture/dm.md @@ -149,31 +149,66 @@ Distributed-таблица вернула бы каждый непарный з ## `dm.dq_summary` -Зерно — «день данных × проверка»: строка отвечает, каково качество дня -сейчас, а не сколько раз его проверяли, — история прогонов и так живёт в -Airflow. Партиция — день данных; прогон звена DQ заменяет партиции -проверенных дней через `REPLACE PARTITION`. Обещанный спекой перевод с -`TRUNCATE+INSERT` выполняется той же механикой замены, что и у витрин, — -ничего отдельного учить не надо. +Зерно публичного `dm.dq_summary_v` — «день данных × атомарная проверка». +Строка отвечает, каково качество дня сейчас; историю запусков хранит Airflow. +Партиция совпадает с днём данных и заменяется через `REPLACE PARTITION`. -## Будущее звено DQ конвейера +Деловые колонки сводки: -Встанет в `etl_pipeline` после DM ([конвейер](etl.md)) и проверит только -объекты стенда: +- `data_date` — проверяемый день; +- `check_name` — устойчивое имя инварианта; +- `reference_rows` — число деловых ключей на эталонной стороне; +- `actual_rows` — число деловых ключей в проверяемом объекте; +- `failed_rows` — число отсутствующих или различающихся ключей; +- `status` — `pass` при нуле нарушений, иначе `fail`. -- счётная сверка сессий с эталоном `VisitID` — кука × день, эталон читается - из `ods.event_v` ([модель сессии](dds/session.md)); -- счётчики классов расхождений против описи мира — по дням с закрытым - окном. Проверка техническая, не урок DQ: её оракул — опись генератора, - а у настоящего магазина оракула нет. Обещание узкое и честное: до - витрин доехал тот мир, который родился, — одна растяжка через весь - путь вместо сторожей на каждом шаге. Верно ли сам мир рождён, она - не видит: опись считает тот же генератор; -- счётные сверки витрин стенда с их DDS-источниками. +Служебные `_load_id` и `_load_ts` называют собравший запуск. Представление +перечисляет колонки явно. Порогов и промежуточного состояния нет. -Результаты ложатся в `dm.dq_summary`. Звено — не только сторож конвейера, -но и экспонат: менти видит, как контроль качества устроен в бою — по -расписанию, с историей, рядом с трансформациями. +## Звено DQ конвейера + +Даг `dq_check` стоит в `etl_pipeline` после `dm_transform` и проверяет пять +атомарных договоров: + +- `sessions_vs_reference`: ключ — `client_id` внутри дня, эталон — + `uniqExact(VisitID)` из `ods.event_v`, объект — число строк + `dds.session_v`; +- `classes_vs_inventory`: ключ — `mismatch_class`, эталон — счётчик из + `orders` в описи мира, объект — число заказов класса в + `dm.purchase_vs_orders_v`. Проверяются `match`, `cancelled` и + `amount_delta` только для покрытых описью закрытых дней; +- `revenue_daily_vs_dds`: ключ — `product_category` внутри дня. Сравниваются + `orders`, `units`, `revenue` и `aov` с прямым агрегатом `dds.order_v`; +- `purchase_vs_orders_vs_dds`: ключ — `order_id` внутри дня. Все деловые + поля сверки сравниваются с независимой сборкой из `dds.event_v` и + `dds.order_v`; +- `daily_traffic_vs_dds`: ключ — день. `visitors` и `known_users` + сравниваются с прямыми агрегатами `dds.session_v` и + `dds.identity_map_v`. + +Область каждой проверки выводится из объединения сторон. Полностью потерянный +день или ключ поэтому не исчезает из результата. Разные ошибки тоже не могут +скомпенсировать друг друга общим итогом: `failed_rows` считается после +сопоставления деловых ключей. + +`classes_vs_inventory` — техническая растяжка пути от генератора до DM. Она +подтверждает, что до витрины доехал записанный в описи мир, но не проверяет +сам генератор: у настоящего магазина такого оракула нет. + +Четыре SQL-сверки считают дневные итоги в ClickHouse и возвращают Airflow +только сводку и малый срез диагностики. Техническая сверка с описью читает +небольшой файл JSON в Python. Все пять сборок пишут в общий донор. После их +успеха даг заменяет проверенные дневные партиции. Затем именованные задачи +читают опубликованную сводку и падают при `failed_rows > 0`. Расхождение успевает +сохраниться до красного завершения, а ошибка сборки не затирает прежний +результат. Красная задача печатает не больше двадцати нарушенных ключей за +проверку со значениями сторон. Для полного списка итоговую агрегацию SQL-сверки +заменяют на `SELECT * FROM comparison WHERE failed ORDER BY data_date, +business_key`. + +Если после починки день исчез с обеих сторон, его больше нет и в новом доноре. +Такой день удаляется из сводки отдельно: старая красная строка не должна +превращаться в историю попыток. ## Даг самопроверки @@ -188,12 +223,10 @@ Airflow. Партиция — день данных; прогон звена DQ валидационный даг с группами проверок по слоям давал менти обратную связь по каждому шагу домашки, не трогая рабочий конвейер. -## Остаётся нарезке +## Остается нарезке -- Имена звена DQ и дага самопроверки. -- Ключи шардирования остальных материализованных витрин — вместе с их - DDL; правило партиции-донора — в [доке хранилища](storage.md). -- Тексты учебных заданий и точный список проверок обоих контуров. +- Имя и состав дага самопроверки домашних заданий. +- Тексты учебных заданий и точный список проверок учебного контура. ## Отклонённые варианты diff --git a/docs/architecture/etl.md b/docs/architecture/etl.md index 05ea681..2d755e7 100644 --- a/docs/architecture/etl.md +++ b/docs/architecture/etl.md @@ -13,8 +13,8 @@ Дневной конвейер на стенде один — даг `etl_pipeline`. Своей работы у него нет: он дёргает слойные даги последовательно и ждёт каждый — -`dds_transform`, затем `dm_transform`. Будущее звено DQ встанет после DM. -Его состав описан в [доке слоя DM](dm.md). +`dds_transform`, затем `dm_transform` и `dq_check`. Состав проверок описан в +[доке слоя DM](dm.md). Ждущие вызовы сделаны `TriggerDagRunOperator`. Проверка через Context7 по ближайшей доступной документации Airflow 3.x (версия 3.1.6) подтвердила, что @@ -28,7 +28,7 @@ ```text world_initialize ─┐ -world_live_day ───┼─> etl_pipeline ─> dds_transform ─> dm_transform +world_live_day ───┼─> etl_pipeline ─> dds_transform ─> dm_transform ─> dq_check запуск рукой ─────┘ world_next_day ─────> источники без ETL @@ -140,8 +140,8 @@ ODS и ещё не собранный в DDS, затем при необходи диапазон, включая форму необязательных параметров того же дага, — этап 5 (#7); туда же стык #111 про дату модельного мира и `logical_date`. - Инкрементальный ETL вместо пересборки — техдолг #2. -- Состав витрин `dm_transform` и звена DQ — [дока слоя DM](dm.md); - имена задач — нарезка этапа 4. +- Состав витрин `dm_transform` и проверок `dq_check` — + [дока слоя DM](dm.md). ## Отклонённые варианты diff --git a/docs/architecture/storage.md b/docs/architecture/storage.md index 7eacb5e..bd3d97c 100644 --- a/docs/architecture/storage.md +++ b/docs/architecture/storage.md @@ -606,6 +606,9 @@ SQL-файлы исполняет штатный `SQLExecuteQueryOperator`. Пр | `dm` | `dm.purchase_vs_orders_v` | публичный договор сверки с `NULL` для отсутствующей стороны | | `dm` | `dm.daily_traffic_rep` / `_dist` | версии дневных счетчиков трафика | | `dm` | `dm.daily_traffic_v` | точные текущие счетчики трафика с `FINAL` | +| `dm` | `dm.dq_summary_rep` / `_dist` | текущее качество дня по атомарным проверкам | +| `dm` | `dm.dq_summary_stage_rep` / `_dist` | общий донор дневных партиций DQ-сводки | +| `dm` | `dm.dq_summary_v` | публичная сводка качества дней | | `dic` | `dic.products_file` | чтец CSV-каталога, общего с генератором | | `dic` | `dic.products` | словарь товаров поверх подложки | diff --git a/docs/specs/2026-07-30-stand-v2-realism.md b/docs/specs/2026-07-30-stand-v2-realism.md index a8400ef..24690ce 100644 --- a/docs/specs/2026-07-30-stand-v2-realism.md +++ b/docs/specs/2026-07-30-stand-v2-realism.md @@ -514,8 +514,8 @@ Kafka день переигрывается генератором заново: - `events_enriched_v`, `top_pages_daily_v`, `session_overview_v`, `dq_errors_daily_v` — переезжают на новую модель без смены роли: источник — `dds.event_v`, не `ods.event`. -- `dm.dq_summary` переводится с TRUNCATE+INSERT на партиционную замену - (политика «без TRUNCATE»). +- `dm.dq_summary` хранит текущее качество дня и обновляется партиционной + заменой (политика «без TRUNCATE»). Дашборд Superset получает три новых сюжета: выручка по дням и категориям, таблица сверки с классами расхождений, пара посетители/известные. diff --git a/sql/ddl/55-dm-tables.sql b/sql/ddl/55-dm-tables.sql index de4ea26..3dc0dec 100644 --- a/sql/ddl/55-dm-tables.sql +++ b/sql/ddl/55-dm-tables.sql @@ -121,3 +121,50 @@ ENGINE = Distributed( 'daily_traffic_rep', cityHash64(report_date) ); + +-- Сводка качества хранит текущее состояние атомарной проверки одного дня. +-- История запусков остаётся в Airflow, поэтому версия строки не нужна: +-- следующий успешный прогон заменяет дневную партицию целиком. +CREATE TABLE IF NOT EXISTS dm.dq_summary_rep ON CLUSTER clickstream_cluster +( + data_date Date, + check_name LowCardinality(String), + reference_rows UInt64, + actual_rows UInt64, + failed_rows UInt64, + status LowCardinality(String), + _load_id String, + _load_ts DateTime64(3, 'UTC') +) +ENGINE = ReplicatedMergeTree('/clickhouse/tables/{shard}/{database}/{table}', '{replica}') +PARTITION BY data_date +ORDER BY (data_date, check_name); + +-- Все строки одной проверки лежат на одном шарде. Таблица мала, а ключ +-- check_name делает раскладку донора и цели одинаковой без нового измерения. +CREATE TABLE IF NOT EXISTS dm.dq_summary_dist ON CLUSTER clickstream_cluster +AS dm.dq_summary_rep +ENGINE = Distributed( + 'clickstream_cluster', + 'dm', + 'dq_summary_rep', + cityHash64(check_name) +); + +-- Источник и цель REPLACE PARTITION обязаны совпадать по структуре, ключам и +-- политике хранения. Требование повторно сверено по документации ClickHouse +-- через Context7 27 августа 2026 года; AS не даёт формам разойтись. +CREATE TABLE IF NOT EXISTS dm.dq_summary_stage_rep ON CLUSTER clickstream_cluster +AS dm.dq_summary_rep +ENGINE = ReplicatedMergeTree('/clickhouse/tables/{shard}/{database}/{table}', '{replica}') +PARTITION BY data_date +ORDER BY (data_date, check_name); + +CREATE TABLE IF NOT EXISTS dm.dq_summary_stage_dist ON CLUSTER clickstream_cluster +AS dm.dq_summary_stage_rep +ENGINE = Distributed( + 'clickstream_cluster', + 'dm', + 'dq_summary_stage_rep', + cityHash64(check_name) +); diff --git a/sql/ddl/60-dm-views.sql b/sql/ddl/60-dm-views.sql index f47a130..bce76ba 100644 --- a/sql/ddl/60-dm-views.sql +++ b/sql/ddl/60-dm-views.sql @@ -42,3 +42,18 @@ SELECT _load_id, _load_ts FROM dm.daily_traffic_dist FINAL; + +-- Сводка отвечает, какие атомарные договоры дня выполняются сейчас. +-- Явный список не пропускает в публичный договор будущие физические колонки. +CREATE VIEW IF NOT EXISTS dm.dq_summary_v ON CLUSTER clickstream_cluster +AS +SELECT + data_date, + check_name, + reference_rows, + actual_rows, + failed_rows, + status, + _load_id, + _load_ts +FROM dm.dq_summary_dist; diff --git a/sql/dq/classes_vs_inventory.sql b/sql/dq/classes_vs_inventory.sql new file mode 100644 index 0000000..d686290 --- /dev/null +++ b/sql/dq/classes_vs_inventory.sql @@ -0,0 +1,17 @@ +-- DQ: стендовая сторона технической растяжки от описи мира. +-- +-- Эталонные счётчики читает Airflow из world-inventory.json: у настоящего +-- магазина такого оракула нет. Запрос возвращает только наблюдаемую сторону; +-- Python-задача сопоставляет ее с тремя счетчиками закрытого дня из описи. +SELECT + order_day AS data_date, + mismatch_class, + count() AS orders +FROM dm.purchase_vs_orders_v +WHERE mismatch_class IN ('match', 'cancelled', 'amount_delta') +GROUP BY + data_date, + mismatch_class +ORDER BY + data_date, + mismatch_class diff --git a/sql/dq/daily_traffic_vs_dds.sql b/sql/dq/daily_traffic_vs_dds.sql new file mode 100644 index 0000000..9861076 --- /dev/null +++ b/sql/dq/daily_traffic_vs_dds.sql @@ -0,0 +1,71 @@ +-- DQ: дневной трафик против прямого агрегата сессий и карты идентичностей. +-- +-- Публичные представления скрывают физическую ко-локацию, поэтому GLOBAL +-- назван явно, как в самой витрине. Полное объединение не прячет день, +-- исчезнувший с любой стороны. +WITH + reference AS + ( + SELECT + sessions.session_date AS report_date, + uniqExact(sessions.client_id) AS visitors, + uniqExactIf(identities.user_id, identities.user_id != 0) AS known_users + FROM dds.session_v AS sessions + GLOBAL LEFT JOIN dds.identity_map_v AS identities USING (client_id) + GROUP BY report_date + ), + actual AS + ( + SELECT + report_date, + count() AS physical_rows, + any(visitors) AS visitors, + any(known_users) AS known_users + FROM dm.daily_traffic_v + GROUP BY report_date + ), + comparison AS + ( + SELECT + coalesce(reference.report_date, actual.report_date) AS data_date, + toString(coalesce(reference.report_date, actual.report_date)) + AS business_key, + isNotNull(reference.report_date) AS reference_present, + isNotNull(actual.report_date) AS actual_present, + toString(tuple(reference.visitors, reference.known_users)) + AS reference_value, + concat( + toString(tuple(actual.visitors, actual.known_users)), + ', rows=', + toString(actual.physical_rows) + ) AS actual_value, + isNull(reference.report_date) + OR isNull(actual.report_date) + OR actual.physical_rows != 1 + OR NOT isNotDistinctFrom(reference.visitors, actual.visitors) + OR NOT isNotDistinctFrom(reference.known_users, actual.known_users) + AS failed + FROM reference + GLOBAL FULL OUTER JOIN actual USING (report_date) + ) +SELECT + data_date, + countIf(reference_present) AS reference_rows, + countIf(actual_present) AS actual_rows, + countIf(failed) AS failed_rows, + groupArrayIf(20)( + concat( + toString(data_date), + ' key=', + business_key, + ': эталон=', + if(reference_present, reference_value, 'нет ключа'), + ', объект=', + if(actual_present, actual_value, 'нет ключа') + ), + failed + ) AS diagnostics +FROM comparison +GROUP BY data_date +ORDER BY data_date +SETTINGS join_use_nulls = 1 diff --git a/sql/dq/prepare.sql b/sql/dq/prepare.sql new file mode 100644 index 0000000..a68f35d --- /dev/null +++ b/sql/dq/prepare.sql @@ -0,0 +1,7 @@ +-- DQ: общий донор очищается перед сборкой всех проверок. +-- +-- Если одна сборка упадет, публикация не начнется. Следующий запуск удалит +-- неполный результат здесь и соберет донор заново. + +ALTER TABLE dm.dq_summary_stage_rep ON CLUSTER clickstream_cluster +DROP PARTITION ALL; diff --git a/sql/dq/purchase_vs_orders_vs_dds.sql b/sql/dq/purchase_vs_orders_vs_dds.sql new file mode 100644 index 0000000..cc03c78 --- /dev/null +++ b/sql/dq/purchase_vs_orders_vs_dds.sql @@ -0,0 +1,120 @@ +-- DQ: строка сверки покупок и заказов против независимой сборки из DDS. +-- +-- Проверяются все деловые поля. Счёт физических строк отдельно ловит дубль +-- ключа, даже если обе копии несут одинаковые значения. +WITH + purchases AS + ( + SELECT + purchase_id[1] AS purchase_order_id, + any(event_date) AS purchase_day, + any(purchase_revenue[1]) AS declared_revenue, + count() AS purchase_count + FROM dds.event_v + WHERE event_type = 'purchase' + GROUP BY purchase_order_id + ), + orders AS + ( + SELECT + order_id AS backend_order_id, + order_date, + items_total, + status + FROM dds.order_v + ), + reference AS + ( + SELECT + coalesce(order_date, purchase_day) AS order_day, + coalesce(purchase_order_id, backend_order_id) AS order_id, + declared_revenue, + items_total, + status, + multiIf( + backend_order_id IS NULL, 'awaiting_order', + status = 'cancelled', 'cancelled', + purchase_order_id IS NULL, 'lost_event', + purchase_count > 1, 'duplicate_event', + declared_revenue != items_total, 'amount_delta', + 'match' + ) AS mismatch_class + FROM purchases + GLOBAL FULL OUTER JOIN orders + ON purchase_order_id = backend_order_id + ), + actual AS + ( + SELECT + order_day, + order_id, + count() AS physical_rows, + any(declared_revenue) AS declared_revenue, + any(items_total) AS items_total, + any(status) AS status, + any(mismatch_class) AS mismatch_class + FROM dm.purchase_vs_orders_v + GROUP BY + order_day, + order_id + ), + comparison AS + ( + SELECT + coalesce(reference.order_day, actual.order_day) AS data_date, + coalesce(reference.order_id, actual.order_id) AS business_key, + isNotNull(reference.order_id) AS reference_present, + isNotNull(actual.order_id) AS actual_present, + toString(tuple( + reference.declared_revenue, + reference.items_total, + reference.status, + reference.mismatch_class + )) AS reference_value, + concat( + toString(tuple( + actual.declared_revenue, + actual.items_total, + actual.status, + actual.mismatch_class + )), + ', rows=', + toString(actual.physical_rows) + ) AS actual_value, + isNull(reference.order_id) + OR isNull(actual.order_id) + OR actual.physical_rows != 1 + OR NOT isNotDistinctFrom( + reference.declared_revenue, + actual.declared_revenue + ) + OR NOT isNotDistinctFrom(reference.items_total, actual.items_total) + OR NOT isNotDistinctFrom(reference.status, actual.status) + OR NOT isNotDistinctFrom( + reference.mismatch_class, + actual.mismatch_class + ) AS failed + FROM reference + GLOBAL FULL OUTER JOIN actual USING (order_day, order_id) + ) +SELECT + data_date, + countIf(reference_present) AS reference_rows, + countIf(actual_present) AS actual_rows, + countIf(failed) AS failed_rows, + groupArrayIf(20)( + concat( + toString(data_date), + ' key=', + business_key, + ': эталон=', + if(reference_present, reference_value, 'нет ключа'), + ', объект=', + if(actual_present, actual_value, 'нет ключа') + ), + failed + ) AS diagnostics +FROM comparison +GROUP BY data_date +ORDER BY data_date +SETTINGS join_use_nulls = 1 diff --git a/sql/dq/replace.sql b/sql/dq/replace.sql new file mode 100644 index 0000000..6c755b7 --- /dev/null +++ b/sql/dq/replace.sql @@ -0,0 +1,7 @@ +-- DQ: публикация одного полностью собранного дня общей сводки. +-- +-- День приходит из самого донора после успеха всех пяти сборок. REPLACE +-- PARTITION работает с локальными таблицами одинаковой структуры; это +-- ограничение сверено по документации ClickHouse 27 августа 2026 года. +ALTER TABLE dm.dq_summary_rep ON CLUSTER clickstream_cluster +REPLACE PARTITION '{{ params.day }}' FROM dm.dq_summary_stage_rep; diff --git a/sql/dq/revenue_daily_vs_dds.sql b/sql/dq/revenue_daily_vs_dds.sql new file mode 100644 index 0000000..1a9bae5 --- /dev/null +++ b/sql/dq/revenue_daily_vs_dds.sql @@ -0,0 +1,107 @@ +-- DQ: дневная выручка против прямого агрегата заказов DDS. +-- +-- День берётся из объединения сторон. Поэтому исчезнувший целиком день или +-- категория остаются видимы. Денежный расчет повторяет публичный договор +-- витрины и сохраняет Decimal до округления AOV. +WITH + reference AS + ( + SELECT + report_date, + product_category, + orders, + units, + CAST(revenue, 'Decimal(18, 2)') AS revenue, + CAST( + round(divideDecimal(revenue, toDecimal128(orders, 0), 6), 2), + 'Decimal(18, 2)' + ) AS aov + FROM + ( + SELECT + order_date AS report_date, + dictGet('dic.products', 'category', item_sku) AS product_category, + uniqExact(order_id) AS orders, + sum(item_quantity) AS units, + sum(item_price * item_quantity) AS revenue + FROM dds.order_v + ARRAY JOIN + item_sku, + item_quantity, + item_price + WHERE status = 'paid' + GROUP BY + report_date, + product_category + ) + ), + actual AS + ( + SELECT + report_date, + product_category, + count() AS physical_rows, + any(orders) AS orders, + any(units) AS units, + any(revenue) AS revenue, + any(aov) AS aov + FROM dm.revenue_daily_v + GROUP BY + report_date, + product_category + ), + comparison AS + ( + SELECT + coalesce(reference.report_date, actual.report_date) AS data_date, + coalesce(reference.product_category, actual.product_category) + AS business_key, + isNotNull(reference.product_category) AS reference_present, + isNotNull(actual.product_category) AS actual_present, + toString(tuple( + reference.orders, + reference.units, + reference.revenue, + reference.aov + )) AS reference_value, + concat( + toString(tuple( + actual.orders, + actual.units, + actual.revenue, + actual.aov + )), + ', rows=', + toString(actual.physical_rows) + ) AS actual_value, + isNull(reference.product_category) + OR isNull(actual.product_category) + OR actual.physical_rows != 1 + OR NOT isNotDistinctFrom(reference.orders, actual.orders) + OR NOT isNotDistinctFrom(reference.units, actual.units) + OR NOT isNotDistinctFrom(reference.revenue, actual.revenue) + OR NOT isNotDistinctFrom(reference.aov, actual.aov) AS failed + FROM reference + GLOBAL FULL OUTER JOIN actual USING (report_date, product_category) + ) +SELECT + data_date, + countIf(reference_present) AS reference_rows, + countIf(actual_present) AS actual_rows, + countIf(failed) AS failed_rows, + groupArrayIf(20)( + concat( + toString(data_date), + ' key=', + business_key, + ': эталон=', + if(reference_present, reference_value, 'нет ключа'), + ', объект=', + if(actual_present, actual_value, 'нет ключа') + ), + failed + ) AS diagnostics +FROM comparison +GROUP BY data_date +ORDER BY data_date +SETTINGS join_use_nulls = 1 diff --git a/sql/dq/sessions_vs_reference.sql b/sql/dq/sessions_vs_reference.sql new file mode 100644 index 0000000..cb03c2a --- /dev/null +++ b/sql/dq/sessions_vs_reference.sql @@ -0,0 +1,65 @@ +-- DQ: эталонная сессия Метрики против сессии, нарезанной слоем DDS. +-- +-- Сопоставление идет по куке внутри завершенного дня: лишняя сессия одной +-- куки не скрывает потерянную сессию другой. Результат содержит итог дня. +WITH + (SELECT min(EventDate) FROM ods.event_v) AS first_source_day, + first_source_day + ({position:UInt16} - 1) AS last_lived_day, + reference AS + ( + SELECT + EventDate AS data_date, + ClientID AS client_id, + uniqExact(VisitID) AS sessions + FROM ods.event_v + WHERE EventDate <= last_lived_day + GROUP BY + data_date, + client_id + ), + actual AS + ( + SELECT + session_date AS data_date, + client_id, + count() AS sessions + FROM dds.session_v + WHERE session_date <= last_lived_day + GROUP BY + data_date, + client_id + ), + comparison AS + ( + SELECT + coalesce(reference.data_date, actual.data_date) AS data_date, + toString(coalesce(reference.client_id, actual.client_id)) AS business_key, + isNotNull(reference.client_id) AS reference_present, + isNotNull(actual.client_id) AS actual_present, + toString(reference.sessions) AS reference_value, + toString(actual.sessions) AS actual_value, + NOT isNotDistinctFrom(reference.sessions, actual.sessions) AS failed + FROM reference + GLOBAL FULL OUTER JOIN actual USING (data_date, client_id) + ) +SELECT + data_date, + countIf(reference_present) AS reference_rows, + countIf(actual_present) AS actual_rows, + countIf(failed) AS failed_rows, + groupArrayIf(20)( + concat( + toString(data_date), + ' key=', + business_key, + ': эталон=', + if(reference_present, reference_value, 'нет ключа'), + ', объект=', + if(actual_present, actual_value, 'нет ключа') + ), + failed + ) AS diagnostics +FROM comparison +GROUP BY data_date +ORDER BY data_date +SETTINGS join_use_nulls = 1