From 741820789f4efc84bc52e40ec949580ef8fef191 Mon Sep 17 00:00:00 2001 From: Dmitry Dementev Date: Wed, 26 Aug 2026 14:56:33 -0400 Subject: [PATCH] =?UTF-8?q?refactor(dm):=20=D1=83=D1=82=D0=BE=D1=87=D0=BD?= =?UTF-8?q?=D0=B5=D0=BD=D1=8B=20=D0=BE=D0=B3=D1=80=D0=B0=D0=BD=D0=B8=D1=87?= =?UTF-8?q?=D0=B5=D0=BD=D0=B8=D1=8F=20DAG=20=D0=B8=20=D0=BF=D0=BE=D1=8F?= =?UTF-8?q?=D1=81=D0=BD=D0=B5=D0=BD=D0=B8=D1=8F=20SQL?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Зачем: - Ревью выявило неясные причины ограничений и защитных условий в учебном образце. - Что: - Спекулятивный общий потолок задач заменен адресным лимитом размноженной замены с точным описанием OOM. - Объяснены нулевое значение непарного LEFT JOIN и опасность пустого DDS при замене партиций. - Добавлена ссылка на разбор области пересборки, а общий штамп про Context7 удален. - Проверка: - make lint, make config-test, make smoke; успешный запуск dm_transform. --- dags/dm_transform.py | 16 +++++++--------- sql/dm/daily_traffic_rebuild.sql | 2 ++ sql/dm/revenue_daily_scope.sql | 4 ++++ 3 files changed, 13 insertions(+), 9 deletions(-) diff --git a/dags/dm_transform.py b/dags/dm_transform.py index 066b73f..4915ef5 100644 --- a/dags/dm_transform.py +++ b/dags/dm_transform.py @@ -3,9 +3,6 @@ Выручка наследует ритм заказов и заменяет дневные партиции изменяемого окна. Трафик наследует ретроспективность карты идентичностей и пересчитывает всю историю новой версией. Объем каждого прогона выводится из состояния слоев. - -Форма DAG и динамическая замена сверены с публичным API Airflow 3 через -Context7; DDL и REPLACE PARTITION — с документацией ClickHouse. """ from __future__ import annotations @@ -26,7 +23,7 @@ CLICKHOUSE_CONNECTION = "clickhouse_default" def _days(first_day: datetime.date, last_day: datetime.date) -> list[str]: - """Дни замены строками от левой до правой границы включительно.""" + """Дни отрезка строками - в таком виде их ждет замена партиции.""" return [ (first_day + datetime.timedelta(days=shift)).isoformat() for shift in range((last_day - first_day).days + 1) @@ -49,7 +46,6 @@ def replacements(scope: dict[str, object]) -> list[dict[str, object]]: start_date=START_DATE, is_paused_upon_creation=False, max_active_runs=1, - max_active_tasks=3, template_searchpath=str(SQL_ROOT), tags=["dm"], ) @@ -70,7 +66,8 @@ def dm_transform(): row = hook.get_first(query) if row is None: raise AirflowException( - "нет слепка заказов в ODS или готовых заказов в DDS" + "нет слепка заказов в ODS или готовых заказов в DDS; " + "дождитесь загрузки слепка и сначала выполните dds_transform" ) first_day, last_day = row days = _days(first_day, last_day) @@ -106,9 +103,10 @@ 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)) diff --git a/sql/dm/daily_traffic_rebuild.sql b/sql/dm/daily_traffic_rebuild.sql index 02318fc..83d3e1c 100644 --- a/sql/dm/daily_traffic_rebuild.sql +++ b/sql/dm/daily_traffic_rebuild.sql @@ -15,6 +15,8 @@ INSERT INTO dm.daily_traffic_dist SELECT sessions.session_date, uniqExact(sessions.client_id), + -- При стендовом join_use_nulls=0 непарный UInt64 получает 0. Условие + -- не дает анонимной сессии превратиться в известного пользователя. uniqExactIf(identities.user_id, identities.user_id != 0), {load_id:String}, now64(3, 'UTC') diff --git a/sql/dm/revenue_daily_scope.sql b/sql/dm/revenue_daily_scope.sql index 2eea500..c15154f 100644 --- a/sql/dm/revenue_daily_scope.sql +++ b/sql/dm/revenue_daily_scope.sql @@ -4,6 +4,8 @@ -- изменить. Сам агрегат читает только DDS; ODS здесь нужен лишь для границ. -- Первый запуск и повтор после частичного отказа расширяют левую границу до -- первого дня оплаченных заказов, которого еще нет в витрине. +-- Почему хвост начинается с первого пропуска и чем за это платит ODS, +-- разобрано в sql/dds/order_scope.sql. -- -- Файл кончается без точки с запятой: даг читает его одним запросом и ждет -- одну строку с границами. @@ -31,5 +33,7 @@ WITH SELECT least(window_start, coalesce(first_missing_day, window_start)) AS first_day, window_end AS last_day +-- Без ODS неизвестно окно. Без готового DDS донор останется пустым, а замена +-- сотрет уже собранные дни выручки. Поэтому для прогона нужны оба источника. WHERE (SELECT count() FROM dds.order_v) > 0 AND (SELECT count() FROM ods.order_dist) > 0