diff --git a/dags/dds_transform.py b/dags/dds_transform.py index bae503f..9036377 100644 --- a/dags/dds_transform.py +++ b/dags/dds_transform.py @@ -1,11 +1,12 @@ -"""Трансформации слоя DDS: пересборка дней, которые ещё могут измениться. +"""Трансформации слоя DDS: сборка модели заменой дневных партиций. -Даг собирает модель поверх принятого ODS. Первая его сущность — заказ; сессии -и карта идентичностей встанут соседними группами задач. +Даг собирает модель поверх принятого ODS. Сущностей у него две — заказ и +сессия, зависимости по данным между ними нет; карта идентичностей встанет +соседней группой задач. Расписания у дага нет и параметров тоже: объём каждого прогона он выводит сам -из состояния слоёв, поэтому один и тот же запуск одинаково верен из триггера -дневного конвейера, с кнопки менти и после падения +из состояния слоёв и позиции мира, поэтому один и тот же запуск одинаково +верен из триггера дневного конвейера, с кнопки менти и после падения (docs/architecture/etl.md, «Объём прогона»). Повтор прогона гасит не колонка версии, а замена партиции целиком: день @@ -24,7 +25,11 @@ 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 dag, task, task_group +from airflow.sdk import Variable, dag, task, task_group + +# Пропуск приезжает из Task SDK: в airflow.exceptions он объявлен устаревшим +# и говорит об этом в журнал задачи (проверено на 3.3.0). +from airflow.sdk.exceptions import AirflowSkipException START_DATE = datetime.datetime(2026, 1, 1, tzinfo=datetime.UTC) @@ -36,6 +41,61 @@ CLICKHOUSE_CONNECTION = "clickhouse_default" # почему у стенда лежит копия и чем она рискует — compose.yaml. ORDER_WINDOW_DAYS = int(os.environ["ORDER_WINDOW_DAYS"]) +# Позиция на оси модельного времени: номер первого несыгранного дня. Её ставит +# пульт мира, и только она отличает прожитый день от живого — зачем это +# сессиям, разобрано в sql/dds/session_scope.sql. +WORLD_POSITION = "world_position" + + +def _world_position() -> int: + """Позиция мира числом, с внятным отказом на непригодном значении. + + Ту же переменную читает пульт, но общего модуля у дагов нет намеренно: + связь между пультом и трансформацией держит сама переменная, а не импорт. + """ + raw_position = Variable.get(WORLD_POSITION, default=None) + if raw_position is None: + raise AirflowException( + "мир ещё не создан: выполните make up и дождитесь world_initialize" + ) + try: + position = int(raw_position) + except (TypeError, ValueError) as error: + raise AirflowException( + f"world_position={raw_position!r} не является номером дня; " + "выполните make rebuild-storage" + ) from error + if position < 1: + raise AirflowException( + f"мир не готов: world_position={position}; " + "выполните make up или make rebuild-storage" + ) + return position + + +def _days(first_day: datetime.date, last_day: datetime.date) -> list[str]: + """Дни отрезка строками — в таком виде их ждёт замена партиции.""" + return [ + (first_day + datetime.timedelta(days=shift)).isoformat() + for shift in range((last_day - first_day).days + 1) + ] + + +@task +def replacements(scope: dict[str, object]) -> list[dict[str, object]]: + """Разложить дни в аргументы размноженной задачи замены. + + Шаг существует ради ограничения Airflow: размножать задачу можно + только по целому возврату другой задачи — ни по ключу словаря, ни по + карте поверх него. Отсюда отдельный возврат под один список. + + День едет в запрос подстановкой Jinja, а не параметром ClickHouse: + в ALTER … PARTITION сервер параметр не подставляет — замерено, + отвечает синтаксической ошибкой. Значение при этом не внешнее, его + вернул сам ClickHouse запросом объёма. + """ + return [{"params": {"day": day}} for day in scope["days"]] + @dag( dag_id="dds_transform", @@ -45,6 +105,11 @@ ORDER_WINDOW_DAYS = int(os.environ["ORDER_WINDOW_DAYS"]) # Два прогона разом собирали бы один и тот же двойник: второй затёр бы # чужую сборку между вставкой и заменой партиций. max_active_runs=1, + # Потолок на задачи в прогоне. Размноженная замена норовит выстрелить + # всеми днями сразу, а каждая задача — свой процесс исполнителя, и + # планировщику отведено 640 МиБ (compose.yaml). Замерено: тринадцать дней + # двух сущностей без потолка кончаются SIGKILL от контрольной группы. + max_active_tasks=3, template_searchpath=str(SQL_ROOT), tags=["dds"], ) @@ -69,10 +134,7 @@ def dds_transform(): "проверьте, дошёл ли слепок до хранилища" ) first_day, last_day = row - days = [ - (first_day + datetime.timedelta(days=shift)).isoformat() - for shift in range((last_day - first_day).days + 1) - ] + days = _days(first_day, last_day) logging.info( "объём пересборки: %s … %s, дней %d", first_day, @@ -85,21 +147,6 @@ def dds_transform(): "days": days, } - @task - def replacements(scope: dict[str, object]) -> list[dict[str, object]]: - """Разложить дни в аргументы размноженной задачи замены. - - Шаг существует ради ограничения Airflow: размножать задачу можно - только по целому возврату другой задачи — ни по ключу словаря, ни - по карте поверх него. Отсюда отдельный возврат под один список. - - День едет в запрос подстановкой Jinja, а не параметром ClickHouse: - в ALTER … PARTITION сервер параметр не подставляет — замерено, - отвечает синтаксической ошибкой. Значение при этом не внешнее, его - вернул сам ClickHouse запросом объёма. - """ - return [{"params": {"day": day}} for day in scope["days"]] - days = scope() rebuild = SQLExecuteQueryOperator( @@ -127,7 +174,69 @@ def dds_transform(): rebuild >> replace + @task_group(group_id="session") + def session(): + """Сессия: дни, которые мир уже прожил, а слой ещё не собрал.""" + + @task + def scope() -> dict[str, object]: + """Спросить у хранилища и у позиции мира, какие дни собирает запуск.""" + query = (SQL_ROOT / "dds" / "session_scope.sql").read_text(encoding="utf-8") + hook = ClickHouseHook(clickhouse_conn_id=CLICKHOUSE_CONNECTION) + row = hook.get_first(query, parameters={"position": _world_position()}) + if row is None: + raise AirflowException( + "в ods.event нет ни одного события: собирать нечего; " + "проверьте, дошёл ли поток до хранилища" + ) + first_day, last_day = row + # Пропуск, а не отказ: слой полон по последний прожитый день, а + # сегодняшний ещё идёт. Это обычное состояние повторного прогона, + # и в интерфейсе оно видно пропущенной группой. + if first_day is None: + raise AirflowSkipException( + f"сессии собраны по {last_day} включительно, " + "новых прожитых дней нет" + ) + days = _days(first_day, last_day) + logging.info( + "объём сборки: %s … %s, дней %d", + first_day, + last_day, + len(days), + ) + return { + "first_day": first_day.isoformat(), + "last_day": last_day.isoformat(), + "days": days, + } + + days = scope() + + rebuild = SQLExecuteQueryOperator( + task_id="rebuild", + conn_id=CLICKHOUSE_CONNECTION, + sql="dds/session_rebuild.sql", + parameters={ + "load_id": "{{ run_id }}", + "first_day": days["first_day"], + "last_day": days["last_day"], + }, + split_statements=True, + do_xcom_push=False, + ) + + replace = SQLExecuteQueryOperator.partial( + task_id="replace", + conn_id=CLICKHOUSE_CONNECTION, + sql="dds/session_replace.sql", + do_xcom_push=False, + ).expand_kwargs(replacements(days)) + + rebuild >> replace + order() + session() dds_transform() diff --git a/docs/architecture/dds/session.md b/docs/architecture/dds/session.md index 0099260..b632714 100644 --- a/docs/architecture/dds/session.md +++ b/docs/architecture/dds/session.md @@ -73,15 +73,18 @@ шарде, поэтому сессионизация локальна по природе — капкан смены ключа при вставке, который дока хранилища обещает сборкам DDS, у сессий не стреляет. -Сборка — дневной батч после прихода слепка заказов: хранилище выше ODS по -смыслу батчевое, раз второй источник приходит раз в сутки. К этому моменту -события дня полны, и сессии только что прожитого дня собираются один раз и -целиком; идемпотентность держит замена партиции. В отличие от заказов, где -батч пересобирает до K «дышащих» дней, у сессий пересобирается только одна -партиция: день самодостаточен и замерзает сразу (правило спеки, раздел 6). -Внутри живого дня модель сессии про сегодня честно молчит — «сегодня» -смотрят через `dds.event_v`; сессия и по смыслу не финальна, пока день не -прожит. +Сборка — дневной батч: хранилище выше ODS по смыслу батчевое, раз второй +источник приходит раз в сутки. Собирается день, который мир уже прожил, и +собирается один раз и целиком; какой день прожит, слой спрашивает у позиции +мира ([дневной конвейер](../etl.md), «Объём прогона»). Идемпотентность держит +замена партиции. В отличие от заказов, где батч пересобирает до K «дышащих» +дней, у сессии дышащего окна нет: прожитый день самодостаточен (правило +спеки, раздел 6). Второй раз он соберётся только +вынужденно — когда упавшая замена оставила позади дыру и хвост от неё накрыл +соседние дни. Дней в одном прогоне при этом бывает несколько — столько, +сколько мир успел прожить без слоя. Внутри живого дня +модель сессии про сегодня честно молчит — «сегодня» смотрят через +`dds.event_v`; сессия и по смыслу не финальна, пока день не прожит. Публичная поверхность — `dds.session_v`: обычное представление с явным списком деловых и собственных служебных колонок над `dds.session_dist`. diff --git a/docs/architecture/etl.md b/docs/architecture/etl.md index 1d57ec1..85832ad 100644 --- a/docs/architecture/etl.md +++ b/docs/architecture/etl.md @@ -40,8 +40,8 @@ ## Объём прогона Параметров у конвейера нет: объём он выводит сам из `world_position` и -состояния слоёв. Сессии догоняют хвост — от последнего собранного дня до -последнего прожитого; заказы пересобирают дышащее окно K и тот же хвост; карта +состояния слоёв. Сессии догоняют хвост — от первого несобранного дня по +последний прожитый; заказы пересобирают дышащее окно K и тот же хвост; карта идентичностей пересчитывается целиком; правила витрин — у [слоя DM](dm.md). Поэтому один и тот же запуск одинаково верен из триггера, с кнопки менти и @@ -56,6 +56,16 @@ неудачного приёма она обгоняет хранилище. Слой обязан считать объём по тому, что у него есть, а не по тому, что мир о себе заявляет. +У сессий всё наоборот: им отвечает только позиция, потому что вопрос у них +другой. Им нужен не день, о котором рассказал источник, а день, который мир +**прожил**: события живой день от прожитого не отличают — и посреди дня D, и +сразу после его конца позднейший день событий равен D. Соберись сессии посреди дня, визит, начатый в +23:50, остался бы обрезанным навсегда: замёрзший день слой не пересобирает. +Отсюда правило — последний прожитый день равен первому дню событий плюс позиция +минус один ([`sql/dds/session_scope.sql`](../../sql/dds/session_scope.sql)). +Оговорка про «доехал» остаётся в силе и здесь: за хвост потока отвечает не +позиция, а «Готовность источника». + Ярлыки доставки — `snapshot_date`, `_load_id`, `_load_ts` — не отвечают тоже, и это свойство ODS, а не недосмотр: слой схлопывает доставки, сохраняя деловые версии, поэтому у заказа, который не менялся, ярлык выжившей строки достаётся @@ -67,14 +77,18 @@ ## Готовность источника -Первый шаг `dds_transform` — дождаться, пока хвост топика `hits` дожёван: -отставание его потребителей нулевое (`system.kafka_consumers`). Зелёный -приём заказов о событиях не говорит ничего — топики разные и потребители -разные. В живом режиме зазор ничтожен, но в пакетном генератор выстреливает -дни в топик быстрее, чем чтец их прожёвывает, и сессии собрались бы по -недожёванному дню. DQ-сверка такой день подсветит, но починить его до +Первым шагом `dds_transform` обязан дождаться, пока хвост топика `hits` +дожёван: отставание его потребителей нулевое (`system.kafka_consumers`). +Зелёный приём заказов о событиях не говорит ничего — топики разные и +потребители разные. В живом режиме зазор ничтожен, но в пакетном генератор +выстреливает дни в топик быстрее, чем чтец их прожёвывает, и сессии собираются +по недожёванному дню. DQ-сверка такой день подсветит, но починить его до этапа 5 нечем: профилактика дешевле ремонта, которого ещё нет. +Шага этого в даге ещё нет — он приезжает отдельным тикетом (#156), и до тех +пор дыра открыта: сессии считают день прожитым по позиции мира, а хвост потока +не стережёт никто. На приёмке отставание смотрят руками. + ## Идемпотентность Партиция не заменяется простой вставкой. День собирается в таблицу-двойник — diff --git a/docs/architecture/storage.md b/docs/architecture/storage.md index c21d7b1..7c44f4f 100644 --- a/docs/architecture/storage.md +++ b/docs/architecture/storage.md @@ -16,8 +16,9 @@ keeper, Kafka, каркас сервисов. Этап 2 собран: в `sql/d CSV-файла. Живёт он вне цепочки слоёв, в зоне `dic`, и читает не файл, а подложку над ним ([ADR 0012](../adr/0012-dictionary-home.md)). Этап 4 начат публичным договором события `dds.event_v` над точным состоянием ODS, а следом -получил первую физическую модель слоя — заказ: его пересобирает заменой -партиций даг `dds_transform`. +получил две физические модели слоя — заказ и сессию: их собирает заменой +партиций даг `dds_transform`. Сессия среди них особая: в потоке такой сущности +нет, стенд режет её из событий сам. Дальше по тексту устройство описано так, как оно проектируется; построенное от заложенного отличает карта таблиц в конце. @@ -518,8 +519,8 @@ SQL-файлы исполняет штатный `SQLExecuteQueryOperator`. Пр | `20-ods-tables.sql` | типизированное событие, версии заказа и обе таблицы ошибок | | `30-ods-views.sql` | актуальные события, текущие заказы и матвью разбора в ODS | | `40-stg-views.sql` | матвью приёма: чтец в сырьё | -| `45-dds-tables.sql` | физическая модель заказа и таблица-двойник, донор её партиций | -| `50-dds-views.sql` | публичные договоры события и заказа на языке модели DDS | +| `45-dds-tables.sql` | физические модели заказа и сессии, доноры их партиций | +| `50-dds-views.sql` | публичные договоры события, заказа и сессии на языке модели DDS | Справочники идут сразу за базами: зона `dic` не зависит ни от одного слоя, и правила порядка слоёв к ней не применяются. @@ -581,6 +582,9 @@ SQL-файлы исполняет штатный `SQLExecuteQueryOperator`. Пр | `dds` | `dds.order_rep` / `_dist` | заказ в текущем состоянии, партиция — день заказа | | `dds` | `dds.order_stage_rep` / `_dist` | двойник заказа: в нём день собирается перед заменой партиции | | `dds` | `dds.order_v` | публичный договор заказа | +| `dds` | `dds.session_rep` / `_dist` | визит, нарезанный слоем; партиция — день сессии | +| `dds` | `dds.session_stage_rep` / `_dist` | двойник сессии: в нём день собирается перед заменой партиции | +| `dds` | `dds.session_v` | публичный договор сессии | | `dic` | `dic.products_file` | чтец CSV-каталога, общего с генератором | | `dic` | `dic.products` | словарь товаров поверх подложки | @@ -588,10 +592,11 @@ SQL-файлы исполняет штатный `SQLExecuteQueryOperator`. Пр `INSERT SELECT` одной задачи `parse_batch` из файла `sql/ods/order_parse.sql`. -В слое DDS построены событие и заказ, и устроены они намеренно по-разному: -событие остаётся представлением над ODS, заказ получил физическую модель. -Сессии, карту идентичностей и слой DM принесут следующие тикеты этапа. Их -состав задан разделом 7 мастер-спеки и переносится сюда по мере постройки. +В слое DDS построены событие, заказ и сессия, и устроены они намеренно +по-разному: событие остаётся представлением над ODS, у заказа и сессии — +физические модели. Карту идентичностей и слой DM принесут следующие тикеты +этапа. Их состав задан разделом 7 мастер-спеки и переносится сюда по мере +постройки. ## Что проверено diff --git a/sql/ddl/45-dds-tables.sql b/sql/ddl/45-dds-tables.sql index 9e22762..41d7bd3 100644 --- a/sql/ddl/45-dds-tables.sql +++ b/sql/ddl/45-dds-tables.sql @@ -1,18 +1,18 @@ --- DDS: физическая модель заказа и донор её партиций. +-- DDS: физические модели слоя и доноры их партиций. +-- +-- Идемпотентность здесь устроена иначе, чем в приёме, и это главный урок обеих +-- сущностей файла. В ODS повтор гасит колонка версии: +-- ReplacingMergeTree(updated_at) оставляет позднюю строку. Здесь колонки +-- версии нет вовсе — повтор гасит замена партиции целиком: день собирается +-- заново и подменяется одним атомарным движением. Поэтому строка таблицы +-- ровно одна на своё зерно, а чтению не нужен FINAL. + +-- Локальная таблица заказа. -- -- Зерно — заказ в текущем состоянии, ключ order_id. Историю версий слой не -- хранит: у неё нет читателя ни в одной витрине, а ODS её и не обещает -- (docs/architecture/dds/order.md, «Зерно»). -- --- Идемпотентность здесь устроена иначе, чем в приёме, и это главный урок --- сущности. В ODS повтор гасит колонка версии: ReplacingMergeTree(updated_at) --- оставляет позднюю строку. Здесь колонки версии нет вовсе — повтор гасит --- замена партиции целиком: день собирается заново и подменяется одним --- атомарным движением. Поэтому каждая строка таблицы — ровно один заказ, а --- чтению не нужен FINAL. - --- Локальная таблица шарда. --- -- Партиция — order_date, день заказа в поясе счётчика. Партиция дневная не -- ради размера куска, а потому что день — единица пересборки: пока заказ -- внутри окна изменяемости, его день «дышит» и собирается заново, за окном @@ -93,3 +93,80 @@ ORDER BY order_id; CREATE TABLE IF NOT EXISTS dds.order_stage_dist ON CLUSTER clickstream_cluster AS dds.order_stage_rep ENGINE = Distributed('clickstream_cluster', 'dds', 'order_stage_rep', cityHash64(order_id)); + +-- Локальная таблица сессии. +-- +-- Зерно — визит: подряд идущие события одной куки без пауз длиннее тридцати +-- минут, не пересекающие границу суток (CONTEXT.md, «Визит»). Сущность +-- расчётная, в потоке её нет — слой режет и чеканит её сам +-- (docs/architecture/dds/session.md). +-- +-- Партиция — день сессии. У заказа день дышит внутри окна изменяемости, у +-- сессии замерзает сразу: визит суток не пересекает, поэтому прожитый день +-- собирается один раз и целиком. Отсюда и цена отказа: пропущенный день +-- заказа вернуло бы в работу следующее окно, а день сессии вернёт только счёт +-- хвоста от первой дыры (sql/dds/session_scope.sql). +-- +-- Ключ сортировки — кука и начало сессии: тем же порядком идёт и нарезка, и +-- чтение «визиты посетителя подряд». session_id в ключ не входит: он ручка +-- для витрин и разбора расхождений, а не адрес строки. +CREATE TABLE IF NOT EXISTS dds.session_rep ON CLUSTER clickstream_cluster +( + -- Суррогат слоя: cityHash64 от куки и начала сессии. Детерминированность + -- несущая — пересборка партиции обязана отчеканить те же id + -- (docs/architecture/dds/session.md, «Идентификатор»). + session_id UInt64, + client_id UInt64, + session_date Date, + started_at DateTime('Europe/Samara'), + finished_at DateTime('Europe/Samara'), + duration_seconds UInt32, + -- Счётчики по типам событий: у визита Метрики на этом месте достигнутые + -- цели, которых стенд не шлёт. + events_total UInt32, + pageviews UInt32, + cart_adds UInt32, + purchases UInt32, + -- Вход и выход визита, как StartURL и EndURL у Метрики. Полного списка + -- страниц нет и у неё: топы считает витрина по событиям. + entry_url String, + exit_url String, + -- Атрибуция визита — метки его первого события. + utm_source String, + utm_medium String, + utm_campaign String, + utm_content String, + utm_term String, + -- Паспорт куки: устройство и город приписаны ей на всю жизнь, поэтому + -- берутся с первого события и внутри визита не меняются. + device_category LowCardinality(String), + region_city String, + _load_id String, + _load_ts DateTime64(3, 'UTC') +) +ENGINE = ReplicatedMergeTree('/clickhouse/tables/{shard}/{database}/{table}', '{replica}') +PARTITION BY session_date +ORDER BY (client_id, started_at); + +-- Лицо загрузки и диагностики. Ключ шардирования — cityHash64(client_id), то +-- же выражение, что у событий и у карты идентичностей +-- (docs/architecture/storage.md, «Раскладка по шардам»). Ко-локация здесь не +-- только про соединения: на ней стоит сама сборка — все события куки лежат на +-- одном шарде, и нарезать их можно, не собирая куку с двух нод +-- (sql/dds/session_rebuild.sql). +CREATE TABLE IF NOT EXISTS dds.session_dist ON CLUSTER clickstream_cluster +AS dds.session_rep +ENGINE = Distributed('clickstream_cluster', 'dds', 'session_rep', cityHash64(client_id)); + +-- Донор партиций сессии. Правило донора — в комментарии к dds.order_stage_rep +-- выше: структура и ключи обязаны совпадать с целью, отсюда AS dds.session_rep +-- и тот же ключ шардирования. +CREATE TABLE IF NOT EXISTS dds.session_stage_rep ON CLUSTER clickstream_cluster +AS dds.session_rep +ENGINE = ReplicatedMergeTree('/clickhouse/tables/{shard}/{database}/{table}', '{replica}') +PARTITION BY session_date +ORDER BY (client_id, started_at); + +CREATE TABLE IF NOT EXISTS dds.session_stage_dist ON CLUSTER clickstream_cluster +AS dds.session_stage_rep +ENGINE = Distributed('clickstream_cluster', 'dds', 'session_stage_rep', cityHash64(client_id)); diff --git a/sql/ddl/50-dds-views.sql b/sql/ddl/50-dds-views.sql index dcfed0b..d8eba06 100644 --- a/sql/ddl/50-dds-views.sql +++ b/sql/ddl/50-dds-views.sql @@ -99,3 +99,38 @@ SELECT _load_id, _load_ts FROM dds.order_dist; + +-- DDS: публичный договор сессии на языке бизнес-модели. +-- +-- Под представлением — физическая модель, как у заказа: оконная нарезка +-- визитов на каждое чтение означала бы считать сущность заново в каждом +-- отчёте (docs/architecture/dds/session.md, «Отклонённые варианты»). +-- +-- Эталонного VisitID в договоре нет намеренно: наша нарезка обязана уметь +-- разойтись с эталоном, на этом стоит сверка. За эталоном она ходит в +-- ods.event_v. +CREATE VIEW IF NOT EXISTS dds.session_v ON CLUSTER clickstream_cluster +AS +SELECT + session_id, + client_id, + session_date, + started_at, + finished_at, + duration_seconds, + events_total, + pageviews, + cart_adds, + purchases, + entry_url, + exit_url, + utm_source, + utm_medium, + utm_campaign, + utm_content, + utm_term, + device_category, + region_city, + _load_id, + _load_ts +FROM dds.session_dist; diff --git a/sql/dds/session_rebuild.sql b/sql/dds/session_rebuild.sql new file mode 100644 index 0000000..daef9d2 --- /dev/null +++ b/sql/dds/session_rebuild.sql @@ -0,0 +1,131 @@ +-- DDS: нарезка событий на визиты и сборка дней в таблицу-двойник. +-- +-- Двойник очищается целиком в начале, дни собираются одной вставкой на весь +-- отрезок — по правилу заказа (sql/dds/order_rebuild.sql). +-- +-- Нарезка — центральный расчёт слоя, и стоит она на двух окнах подряд. +-- Первое спрашивает у каждого события, сколько прошло с предыдущего события +-- той же куки в тот же день, и ставит единицу там, где пауза длиннее +-- таймаута: это шов между визитами. Второе считает нарастающий итог этих +-- единиц — номер визита внутри дня куки. Дальше визит становится обычной +-- группировкой. +-- +-- Границу суток окна не проверяют: она уже в разбиении PARTITION BY +-- (client_id, event_date), поэтому визит через полночь не склеится и при +-- короткой паузе. Так правило визита ложится в запрос целиком: одна кука, +-- один день, паузы не длиннее таймаута (CONTEXT.md, «Визит»). +-- +-- Первое событие куки за день сравнивается само с собой — третий аргумент +-- lagInFrame. Паузы у него нет, шов ему не нужен, и номер первого визита +-- получается нулём: номера различают визиты внутри дня куки, а не считают их +-- от единицы. +-- +-- Порядок в окнах полный — метка времени и event_id. Окна считаются двумя +-- разными сортировками (видно в EXPLAIN PIPELINE), и при равных метках они +-- вправе разойтись: шов достался бы одной строке, а нарастающий итог другой, +-- и сессия разъехалась бы надвое. +-- +-- Тридцать минут написаны числом, а не приехали параметром: это правило +-- аналитики, а не константа мира. Генератор режет визиты тем же порогом; +-- разойдись эти два числа — увидит счётная сверка с VisitID +-- (docs/architecture/dds/session.md, «Сверка с эталоном»). + +ALTER TABLE dds.session_stage_rep ON CLUSTER clickstream_cluster DROP PARTITION ALL; + +INSERT INTO dds.session_stage_dist +( + session_id, + client_id, + session_date, + started_at, + finished_at, + duration_seconds, + events_total, + pageviews, + cart_adds, + purchases, + entry_url, + exit_url, + utm_source, + utm_medium, + utm_campaign, + utm_content, + utm_term, + device_category, + region_city, + _load_id, + _load_ts +) +WITH + seams AS ( + SELECT + client_id, + event_date, + event_time, + event_id, + event_type, + url, + utm_source, + utm_medium, + utm_campaign, + utm_content, + utm_term, + device_category, + region_city, + event_time - lagInFrame(event_time, 1, event_time) OVER visit > 1800 + AS opens_visit + FROM dds.event_v + WHERE event_date BETWEEN {first_day:Date} AND {last_day:Date} + WINDOW visit AS ( + PARTITION BY client_id, event_date + ORDER BY event_time, event_id + ROWS BETWEEN 1 PRECEDING AND CURRENT ROW + ) + ), + numbered AS ( + SELECT + *, + sum(opens_visit) OVER ( + PARTITION BY client_id, event_date + ORDER BY event_time, event_id + ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW + ) AS visit_no + FROM seams + ) +SELECT + -- Чекан визита: кука и его начало. Время входит в хеш секундой эпохи, + -- поэтому id держится за содержание сессии, а не за порядок сборки. + cityHash64(client_id, min(event_time)), + client_id, + event_date, + min(event_time), + max(event_time), + max(event_time) - min(event_time), + count(), + countIf(event_type = 'pageview'), + countIf(event_type = 'add_to_cart'), + countIf(event_type = 'purchase'), + argMin(url, event_time), + argMax(url, event_time), + -- Атрибуция и паспорт куки берутся с первого события визита. У паспорта + -- любое событие дало бы то же: устройство и город у куки постоянны. + argMin(utm_source, event_time), + argMin(utm_medium, event_time), + argMin(utm_campaign, event_time), + argMin(utm_content, event_time), + argMin(utm_term, event_time), + argMin(device_category, event_time), + argMin(region_city, event_time), + {load_id:String}, + now64(3, 'UTC') +FROM numbered +GROUP BY client_id, event_date, visit_no +-- Капкана parallel_distributed_insert_select здесь нет, и это выигрыш +-- ко-локации: ключ шардирования у источника и цели один — cityHash64 от +-- куки, — поэтому визит ляжет туда же, где лежали его события, исполнись +-- вставка хоть на инициаторе, хоть локально на шардах. Заказу приходится +-- называть ноль вслух: совпадение ключей у него случайное +-- (sql/dds/order_rebuild.sql). Здесь выбор сделан обратный и он тоже +-- осознанный: настройку не называть, чтобы не притворяться, будто её значение +-- что-то решает. +SETTINGS distributed_foreground_insert = 1; diff --git a/sql/dds/session_replace.sql b/sql/dds/session_replace.sql new file mode 100644 index 0000000..6b220ba --- /dev/null +++ b/sql/dds/session_replace.sql @@ -0,0 +1,11 @@ +-- DDS: подмена одного дня сессий собранной партицией двойника. +-- +-- Механика и доводы целиком — sql/dds/order_replace.sql: замена идёт по +-- локальным таблицам, шард за шардом, атомарность пошардовая, день +-- подставляет Jinja, потому что параметров ALTER … PARTITION не берёт. +-- +-- Разница одна, и она про отказ. Несобранный день заказа вернуло бы в работу +-- следующее окно; сессиям возвращать нечем, и упавшая замена оставляет дыру +-- до тех пор, пока её не найдёт счёт хвоста (sql/dds/session_scope.sql). +ALTER TABLE dds.session_rep ON CLUSTER clickstream_cluster +REPLACE PARTITION '{{ params.day }}' FROM dds.session_stage_rep; diff --git a/sql/dds/session_scope.sql b/sql/dds/session_scope.sql new file mode 100644 index 0000000..fc86d8d --- /dev/null +++ b/sql/dds/session_scope.sql @@ -0,0 +1,50 @@ +-- DDS: объём пересборки сессий — какие дни собирает этот запуск. +-- +-- Ответ — две даты, границы непрерывного отрезка; список дней между ними +-- разворачивает даг. Пустой ответ означает пустой источник. Пустой первый +-- день — что собирать нечего: все прожитые дни уже в слое, и прогон честно +-- пропускает сборку. +-- +-- Файл кончается без точки с запятой: почему — sql/dds/order_scope.sql. +-- +-- Правый край — последний прожитый день, и знает его только позиция мира. +-- События живой день от прожитого не отличают: и посреди дня D, и сразу после +-- его конца max(EventDate) = D. А разница несущая — визит, начатый в 23:50, +-- посреди дня ещё идёт, и собранная сессия оказалась бы обрезанной навсегда: +-- пересобирать замёрзший день слой не станет +-- (docs/architecture/dds/session.md, «Движок и наполнение»). +-- +-- Позиция — номер первого несыгранного дня, счётчик прожитого. Своей копии +-- начала оси стенд из-за неё не заводит: начало он спрашивает у данных, и +-- дата дня N в них — константа, начало оси плюс N. Держится якорь на том, что +-- ранние дни из ods.event не исчезают: срок жизни назначен сырью и таблицам +-- ошибок, самому событию — нет (sql/ddl/20-ods-tables.sql). +-- +-- За «доехал» позиция не отвечает — только за «прожит». Дождаться хвоста +-- потока — работа отдельного шага готовности источника, и в даге его пока нет +-- (#156): до него недожёванный хвост даёт замёрзший недособранный день +-- (docs/architecture/etl.md, «Готовность источника»). +WITH + (SELECT min(EventDate) FROM ods.event_dist) AS first_source_day, + first_source_day + ({position:UInt16} - 1) AS last_lived_day, + -- Хвост считается от первой дыры, а не от последнего собранного дня: + -- замены партиций падают порознь, и счёт по максимуму собранного дня + -- дыру позади себя не заметил бы (доводы целиком — sql/dds/order_scope.sql). + -- У сессий это единственный способ дыру найти: дышащего окна, которое + -- пересобрало бы день заодно, здесь нет. + -- + -- GLOBAL обязателен: обе таблицы распределённые, и обычный NOT IN сверял + -- бы дни источника с локальной частью слоя на каждом шарде. + -- minOrNull отличает «дыр нет» от даты по умолчанию. + ( + SELECT minOrNull(day) + FROM (SELECT DISTINCT EventDate AS day FROM ods.event_dist) + WHERE day GLOBAL NOT IN (SELECT session_date FROM dds.session_dist) + ) AS first_missing_day +SELECT + -- Живой день в слое отсутствует всегда, и работой это не считается: + -- дырой он станет, только когда мир его проживёт. + if(first_missing_day <= last_lived_day, first_missing_day, NULL) AS first_day, + last_lived_day AS last_day +-- FINAL не нужен: повтор доставки не двигает ни край оси, ни набор дней. +WHERE (SELECT count() FROM ods.event_dist) > 0