From 26b2e5fea11837cc491726276b7339e64db883d1 Mon Sep 17 00:00:00 2001 From: Dmitry Dementev Date: Thu, 27 Aug 2026 07:55:28 -0400 Subject: [PATCH 1/2] =?UTF-8?q?feat(dq):=20=D0=B4=D0=BE=D0=B1=D0=B0=D0=B2?= =?UTF-8?q?=D0=BB=D0=B5=D0=BD=20=D0=BA=D0=BE=D0=BD=D1=82=D1=80=D0=BE=D0=BB?= =?UTF-8?q?=D1=8C=20=D0=BA=D0=B0=D1=87=D0=B5=D1=81=D1=82=D0=B2=D0=B0=20?= =?UTF-8?q?=D0=BA=D0=BE=D0=BD=D0=B2=D0=B5=D0=B9=D0=B5=D1=80=D0=B0?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Зачем: - конвейеру нужна публикуемая сводка текущего качества с красным завершением при нарушении договоров. - Что: - добавлены DDL, публичное представление и пять сверок на естественном деловом зерне. - даг dq_check встроен после DM и публикует полные дневные партиции до именованных утверждений. - параллелизм DDS, DM и DQ приведён к измеренному лимиту памяти планировщика. - Проверка: - make lint; make config-test; make rebuild-storage. - make smoke; make check-clickhouse; make check-services. - три семейства проверок покрашены искусственными расхождениями и восстановлены повторным полным прогоном. --- dags/dds_transform.py | 6 + dags/dm_transform.py | 7 +- dags/dq_check.py | 314 ++++++++++++++++++++++ dags/etl_pipeline.py | 14 +- docs/architecture/dm.md | 77 ++++-- docs/architecture/etl.md | 10 +- docs/architecture/storage.md | 3 + docs/specs/2026-07-30-stand-v2-realism.md | 4 +- sql/ddl/55-dm-tables.sql | 47 ++++ sql/ddl/60-dm-views.sql | 15 ++ sql/dq/classes_vs_inventory.sql | 17 ++ sql/dq/daily_traffic_vs_dds.sql | 46 ++++ sql/dq/prepare.sql | 7 + sql/dq/purchase_vs_orders_vs_dds.sql | 95 +++++++ sql/dq/replace.sql | 7 + sql/dq/revenue_daily_vs_dds.sql | 77 ++++++ sql/dq/sessions_vs_reference.sql | 45 ++++ 17 files changed, 753 insertions(+), 38 deletions(-) create mode 100644 dags/dq_check.py create mode 100644 sql/dq/classes_vs_inventory.sql create mode 100644 sql/dq/daily_traffic_vs_dds.sql create mode 100644 sql/dq/prepare.sql create mode 100644 sql/dq/purchase_vs_orders_vs_dds.sql create mode 100644 sql/dq/replace.sql create mode 100644 sql/dq/revenue_daily_vs_dds.sql create mode 100644 sql/dq/sessions_vs_reference.sql diff --git a/dags/dds_transform.py b/dags/dds_transform.py index 46ef624..d4ab708 100644 --- a/dags/dds_transform.py +++ b/dags/dds_transform.py @@ -355,6 +355,10 @@ def dds_transform(): conn_id=CLICKHOUSE_CONNECTION, sql="dds/order_replace.sql", do_xcom_push=False, + # Замены не зависят друг от друга, но каждая — отдельный процесс + # LocalExecutor. Три разом получили SIGKILL при лимите scheduler + # 640 МиБ; последовательность не меняет деловой результат. + max_active_tis_per_dag=1, ).expand_kwargs(replacements(days)) rebuild >> replace @@ -416,6 +420,8 @@ def dds_transform(): conn_id=CLICKHOUSE_CONNECTION, sql="dds/session_replace.sql", do_xcom_push=False, + # Причина единицы разобрана у order.replace выше. + max_active_tis_per_dag=1, ).expand_kwargs(replacements(days)) rebuild >> replace diff --git a/dags/dm_transform.py b/dags/dm_transform.py index d2b38a4..1b6ffbf 100644 --- a/dags/dm_transform.py +++ b/dags/dm_transform.py @@ -76,9 +76,10 @@ 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, + # Каждая задача LocalExecutor — отдельный процесс. После появления + # dq_check три одновременные сборки DM получили SIGKILL при лимите + # scheduler 640 МиБ; две оставляют запас и сохраняют параллельность. + max_active_tasks=2, template_searchpath=str(SQL_ROOT), tags=["dm"], ) diff --git a/dags/dq_check.py b/dags/dq_check.py new file mode 100644 index 0000000..febc5bd --- /dev/null +++ b/dags/dq_check.py @@ -0,0 +1,314 @@ +"""Проверки качества стендовых моделей и витрин. + +Пять проверок сначала собирают общий донор. После его полной сборки дневные +партиции публикуются одной серией замен, и только затем именованные задачи +утверждают результат. Расхождение поэтому остаётся в ``dm.dq_summary_v`` до +красного завершения дага; ошибка сборки прежнюю сводку не меняет. +""" + +from __future__ import annotations + +import datetime +import json +import logging +from collections import defaultdict +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 _write_summary( + check_name: str, + comparisons: list[tuple[object, ...]], +) -> list[str]: + """Свернуть строки деловых ключей в дневную сводку и малый срез ошибок.""" + if not comparisons: + raise AirflowException(f"{check_name}: область проверки пуста") + + totals: dict[datetime.date, list[int]] = defaultdict(lambda: [0, 0, 0]) + diagnostics: list[str] = [] + for ( + data_date, + business_key, + reference_present, + actual_present, + reference_value, + actual_value, + failed, + ) in comparisons: + counts = totals[data_date] + counts[0] += int(reference_present) + counts[1] += int(actual_present) + counts[2] += int(failed) + if failed and len(diagnostics) < DETAIL_LIMIT: + reference_text = reference_value if reference_present else "нет ключа" + actual_text = actual_value if actual_present else "нет ключа" + diagnostics.append( + f"{data_date} key={business_key}: " + f"эталон={reference_text}, объект={actual_text}" + ) + + context = get_current_context() + load_id = context["run_id"] + load_ts = datetime.datetime.now(datetime.UTC) + rows = [ + ( + data_date, + check_name, + reference_rows, + actual_rows, + failed_rows, + "pass" if failed_rows == 0 else "fail", + load_id, + load_ts, + ) + for data_date, (reference_rows, actual_rows, failed_rows) in sorted( + totals.items() + ) + ] + _clickhouse_hook().bulk_insert_rows( + "dm.dq_summary_stage_dist", + rows, + column_names=SUMMARY_COLUMNS, + ) + return diagnostics + + +@task +def build_sql_check(check_name: str, sql_name: str) -> list[str]: + """Выполнить одну SQL-сверку и записать ее дневные итоги в донор.""" + query = (SQL_ROOT / "dq" / sql_name).read_text(encoding="utf-8") + parameters = ( + {"position": _world_position()} + if check_name == "sessions_vs_reference" + else None + ) + comparisons = _clickhouse_hook().get_records(query, parameters=parameters) + return _write_summary(check_name, comparisons) + + +@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 + } + comparisons = [ + ( + data_date, + mismatch_class, + 1, + int((data_date, mismatch_class) in actual), + reference_count, + actual.get((data_date, mismatch_class)), + reference_count != actual.get((data_date, mismatch_class)), + ) + for (data_date, mismatch_class), reference_count in sorted(expected.items()) + ] + return _write_summary("classes_vs_inventory", comparisons) + + +@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, + # Три одновременные сверки на живом мире исчерпали 640 МиБ scheduler и + # получили SIGKILL. Последовательность здесь дешева: весь даг укладывается + # в один учебный прогон, а общему донору параллельность ничего не даёт. + max_active_tasks=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_sql_check.override(task_id="build_sessions_vs_reference")( + "sessions_vs_reference", "sessions_vs_reference.sql" + ) + 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, + # Каждая замена — отдельный процесс LocalExecutor. Последовательность + # держит даг в измеренном лимите памяти планировщика. + max_active_tis_per_dag=1, + ).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/architecture/dm.md b/docs/architecture/dm.md index c84e4cf..2379051 100644 --- a/docs/architecture/dm.md +++ b/docs/architecture/dm.md @@ -149,31 +149,58 @@ 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. Она +подтверждает, что до витрины доехал записанный в описи мир, но не проверяет +сам генератор: у настоящего магазина такого оракула нет. + +Пять сборок пишут в общий донор. Только после их успеха даг заменяет все +проверенные дневные партиции, затем именованные задачи читают опубликованную +сводку и падают при `failed_rows > 0`. Расхождение успевает сохраниться до +красного завершения, а ошибка сборки не затирает прежний результат. Красная +задача печатает не больше двадцати нарушенных ключей со значениями сторон; +полный список воспроизводится запросом проверки. ## Даг самопроверки @@ -188,12 +215,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..867bc04 --- /dev/null +++ b/sql/dq/daily_traffic_vs_dds.sql @@ -0,0 +1,46 @@ +-- 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 + ) +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) +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..730b749 --- /dev/null +++ b/sql/dq/purchase_vs_orders_vs_dds.sql @@ -0,0 +1,95 @@ +-- 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 + ) +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) +ORDER BY + data_date, + business_key +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..d5195b3 --- /dev/null +++ b/sql/dq/revenue_daily_vs_dds.sql @@ -0,0 +1,77 @@ +-- 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 + ) +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) +ORDER BY + data_date, + business_key +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..7394c6d --- /dev/null +++ b/sql/dq/sessions_vs_reference.sql @@ -0,0 +1,45 @@ +-- 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 + ) +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) +ORDER BY + data_date, + business_key +SETTINGS join_use_nulls = 1 -- 2.54.0 From 54686651dfbb4b40216fad8cc5126d2c3eca6ab6 Mon Sep 17 00:00:00 2001 From: Dmitry Dementev Date: Thu, 27 Aug 2026 11:25:19 -0400 Subject: [PATCH 2/2] =?UTF-8?q?fix(dq):=20=D1=83=D1=81=D1=82=D1=80=D0=B0?= =?UTF-8?q?=D0=BD=D0=B5=D0=BD=D0=BE=20=D0=BF=D0=B5=D1=80=D0=B5=D0=BF=D0=BE?= =?UTF-8?q?=D0=BB=D0=BD=D0=B5=D0=BD=D0=B8=D0=B5=20=D0=BF=D0=B0=D0=BC=D1=8F?= =?UTF-8?q?=D1=82=D0=B8=20=D0=BF=D1=80=D0=BE=D0=B2=D0=B5=D1=80=D0=BE=D0=BA?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Зачем: - параллельный DQ выгружал деловые ключи в Python и исчерпывал 640 МиБ планировщика. - Что: - четыре сверки агрегируют дневные итоги и ограниченную диагностику в ClickHouse. - коробка планировщика поднята до 1,5 ГиБ, локальные пределы задач удалены. - измерения и граница между SQL и Python записаны в документации. - Проверка: - make lint, make config-test, make smoke, make check-clickhouse, make check-services. - полный etl_pipeline, красный и восстановительный прогоны dq_check. --- compose.yaml | 5 +- dags/dds_transform.py | 11 -- dags/dm_transform.py | 11 -- dags/dq_check.py | 145 +++++++++++++-------------- docs/adr/0004-resource-limits.md | 25 ++++- docs/architecture/dm.md | 20 ++-- sql/dq/daily_traffic_vs_dds.sql | 59 +++++++---- sql/dq/purchase_vs_orders_vs_dds.sql | 89 ++++++++++------ sql/dq/revenue_daily_vs_dds.sql | 74 ++++++++++---- sql/dq/sessions_vs_reference.sql | 46 ++++++--- 10 files changed, 295 insertions(+), 190 deletions(-) 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 d4ab708..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"], ) @@ -355,10 +350,6 @@ def dds_transform(): conn_id=CLICKHOUSE_CONNECTION, sql="dds/order_replace.sql", do_xcom_push=False, - # Замены не зависят друг от друга, но каждая — отдельный процесс - # LocalExecutor. Три разом получили SIGKILL при лимите scheduler - # 640 МиБ; последовательность не меняет деловой результат. - max_active_tis_per_dag=1, ).expand_kwargs(replacements(days)) rebuild >> replace @@ -420,8 +411,6 @@ def dds_transform(): conn_id=CLICKHOUSE_CONNECTION, sql="dds/session_replace.sql", do_xcom_push=False, - # Причина единицы разобрана у order.replace выше. - max_active_tis_per_dag=1, ).expand_kwargs(replacements(days)) rebuild >> replace diff --git a/dags/dm_transform.py b/dags/dm_transform.py index 1b6ffbf..8e7210c 100644 --- a/dags/dm_transform.py +++ b/dags/dm_transform.py @@ -76,10 +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 — отдельный процесс. После появления - # dq_check три одновременные сборки DM получили SIGKILL при лимите - # scheduler 640 МиБ; две оставляют запас и сохраняют параллельность. - max_active_tasks=2, template_searchpath=str(SQL_ROOT), tags=["dm"], ) @@ -137,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 @@ -199,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 index febc5bd..accefcf 100644 --- a/dags/dq_check.py +++ b/dags/dq_check.py @@ -11,7 +11,6 @@ from __future__ import annotations import datetime import json import logging -from collections import defaultdict from pathlib import Path from airflow.exceptions import AirflowException @@ -69,55 +68,40 @@ def _world_position() -> int: return position -def _write_summary( +def _publish_summary( check_name: str, - comparisons: list[tuple[object, ...]], + summaries: list[tuple[object, ...]], ) -> list[str]: - """Свернуть строки деловых ключей в дневную сводку и малый срез ошибок.""" - if not comparisons: + """Записать дневную сводку и вернуть малый срез готовой диагностики.""" + if not summaries: raise AirflowException(f"{check_name}: область проверки пуста") - totals: dict[datetime.date, list[int]] = defaultdict(lambda: [0, 0, 0]) diagnostics: list[str] = [] - for ( - data_date, - business_key, - reference_present, - actual_present, - reference_value, - actual_value, - failed, - ) in comparisons: - counts = totals[data_date] - counts[0] += int(reference_present) - counts[1] += int(actual_present) - counts[2] += int(failed) - if failed and len(diagnostics) < DETAIL_LIMIT: - reference_text = reference_value if reference_present else "нет ключа" - actual_text = actual_value if actual_present else "нет ключа" - diagnostics.append( - f"{data_date} key={business_key}: " - f"эталон={reference_text}, объект={actual_text}" - ) - context = get_current_context() load_id = context["run_id"] load_ts = datetime.datetime.now(datetime.UTC) - rows = [ - ( - data_date, - check_name, - reference_rows, - actual_rows, - failed_rows, - "pass" if failed_rows == 0 else "fail", - load_id, - load_ts, + 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, + ) ) - for data_date, (reference_rows, actual_rows, failed_rows) in sorted( - totals.items() - ) - ] + diagnostics.extend(details[: DETAIL_LIMIT - len(diagnostics)]) + _clickhouse_hook().bulk_insert_rows( "dm.dq_summary_stage_dist", rows, @@ -126,17 +110,31 @@ def _write_summary( 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]: - """Выполнить одну SQL-сверку и записать ее дневные итоги в донор.""" - query = (SQL_ROOT / "dq" / sql_name).read_text(encoding="utf-8") - parameters = ( - {"position": _world_position()} - if check_name == "sessions_vs_reference" - else None + """Собрать проверку, которой не нужны координаты модельного мира.""" + 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()}, ) - comparisons = _clickhouse_hook().get_records(query, parameters=parameters) - return _write_summary(check_name, comparisons) @task @@ -158,19 +156,29 @@ def build_classes_vs_inventory() -> list[str]: for data_date, mismatch_class, count in _clickhouse_hook().get_records(query) if (data_date, mismatch_class) in expected } - comparisons = [ - ( - data_date, - mismatch_class, - 1, - int((data_date, mismatch_class) in actual), - reference_count, - actual.get((data_date, mismatch_class)), - reference_count != actual.get((data_date, mismatch_class)), - ) - for (data_date, mismatch_class), reference_count in sorted(expected.items()) - ] - return _write_summary("classes_vs_inventory", comparisons) + 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 @@ -186,7 +194,7 @@ def publication_scope() -> list[dict[str, dict[str, str]]]: @task def drop_stale_partitions() -> None: - """Убрать дни, которые после полного прогона не покрывает ни одна сторона.""" + """Убрать дни, исчезнувшие после починки с обеих сторон сверки.""" hook = _clickhouse_hook() stale_days = hook.get_records( """ @@ -237,10 +245,6 @@ def assert_check(check_name: str, diagnostics: list[str]) -> None: start_date=START_DATE, is_paused_upon_creation=False, max_active_runs=1, - # Три одновременные сверки на живом мире исчерпали 640 МиБ scheduler и - # получили SIGKILL. Последовательность здесь дешева: весь даг укладывается - # в один учебный прогон, а общему донору параллельность ничего не даёт. - max_active_tasks=1, template_searchpath=str(SQL_ROOT), tags=["dq"], ) @@ -253,9 +257,7 @@ def dq_check(): do_xcom_push=False, ) - sessions = build_sql_check.override(task_id="build_sessions_vs_reference")( - "sessions_vs_reference", "sessions_vs_reference.sql" - ) + sessions = build_sessions_vs_reference() classes = build_classes_vs_inventory.override( task_id="build_classes_vs_inventory" )() @@ -280,9 +282,6 @@ def dq_check(): conn_id=CLICKHOUSE_CONNECTION, sql="dq/replace.sql", do_xcom_push=False, - # Каждая замена — отдельный процесс LocalExecutor. Последовательность - # держит даг в измеренном лимите памяти планировщика. - max_active_tis_per_dag=1, ).expand_kwargs(days) sessions_assertion = assert_check.override(task_id="sessions_vs_reference")( 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 2379051..8c0f702 100644 --- a/docs/architecture/dm.md +++ b/docs/architecture/dm.md @@ -195,12 +195,20 @@ Distributed-таблица вернула бы каждый непарный з подтверждает, что до витрины доехал записанный в описи мир, но не проверяет сам генератор: у настоящего магазина такого оракула нет. -Пять сборок пишут в общий донор. Только после их успеха даг заменяет все -проверенные дневные партиции, затем именованные задачи читают опубликованную -сводку и падают при `failed_rows > 0`. Расхождение успевает сохраниться до -красного завершения, а ошибка сборки не затирает прежний результат. Красная -задача печатает не больше двадцати нарушенных ключей со значениями сторон; -полный список воспроизводится запросом проверки. +Четыре SQL-сверки считают дневные итоги в ClickHouse и возвращают Airflow +только сводку и малый срез диагностики. Техническая сверка с описью читает +небольшой файл JSON в Python. Все пять сборок пишут в общий донор. После их +успеха даг заменяет проверенные дневные партиции. Затем именованные задачи +читают опубликованную сводку и падают при `failed_rows > 0`. Расхождение успевает +сохраниться до красного завершения, а ошибка сборки не затирает прежний +результат. Красная задача печатает не больше двадцати нарушенных ключей за +проверку со значениями сторон. Для полного списка итоговую агрегацию SQL-сверки +заменяют на `SELECT * FROM comparison WHERE failed ORDER BY data_date, +business_key`. + +Если после починки день исчез с обеих сторон, его больше нет и в новом доноре. +Такой день удаляется из сводки отдельно: старая красная строка не должна +превращаться в историю попыток. ## Даг самопроверки diff --git a/sql/dq/daily_traffic_vs_dds.sql b/sql/dq/daily_traffic_vs_dds.sql index 867bc04..9861076 100644 --- a/sql/dq/daily_traffic_vs_dds.sql +++ b/sql/dq/daily_traffic_vs_dds.sql @@ -23,24 +23,49 @@ WITH 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 - 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) + 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/purchase_vs_orders_vs_dds.sql b/sql/dq/purchase_vs_orders_vs_dds.sql index 730b749..cc03c78 100644 --- a/sql/dq/purchase_vs_orders_vs_dds.sql +++ b/sql/dq/purchase_vs_orders_vs_dds.sql @@ -57,39 +57,64 @@ WITH 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 - 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) -ORDER BY data_date, - business_key + 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/revenue_daily_vs_dds.sql b/sql/dq/revenue_daily_vs_dds.sql index d5195b3..1a9bae5 100644 --- a/sql/dq/revenue_daily_vs_dds.sql +++ b/sql/dq/revenue_daily_vs_dds.sql @@ -49,29 +49,59 @@ WITH 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 - 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) -ORDER BY data_date, - business_key + 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 index 7394c6d..cb03c2a 100644 --- a/sql/dq/sessions_vs_reference.sql +++ b/sql/dq/sessions_vs_reference.sql @@ -1,7 +1,7 @@ -- DQ: эталонная сессия Метрики против сессии, нарезанной слоем DDS. -- --- Строка результата — одна кука одного завершённого дня. Сравнение по ключу --- не даёт лишней сессии одной куки скрыть потерянную сессию другой. +-- Сопоставление идет по куке внутри завершенного дня: лишняя сессия одной +-- куки не скрывает потерянную сессию другой. Результат содержит итог дня. WITH (SELECT min(EventDate) FROM ods.event_v) AS first_source_day, first_source_day + ({position:UInt16} - 1) AS last_lived_day, @@ -28,18 +28,38 @@ WITH 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 - 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) -ORDER BY data_date, - business_key + 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 -- 2.54.0