From 54686651dfbb4b40216fad8cc5126d2c3eca6ab6 Mon Sep 17 00:00:00 2001 From: Dmitry Dementev Date: Thu, 27 Aug 2026 11:25:19 -0400 Subject: [PATCH] =?UTF-8?q?fix(dq):=20=D1=83=D1=81=D1=82=D1=80=D0=B0=D0=BD?= =?UTF-8?q?=D0=B5=D0=BD=D0=BE=20=D0=BF=D0=B5=D1=80=D0=B5=D0=BF=D0=BE=D0=BB?= =?UTF-8?q?=D0=BD=D0=B5=D0=BD=D0=B8=D0=B5=20=D0=BF=D0=B0=D0=BC=D1=8F=D1=82?= =?UTF-8?q?=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