From 05f4e775e2ab013bd9e6ac43f3963c5cd67b1eb3 Mon Sep 17 00:00:00 2001 From: Dmitry Dementev Date: Sun, 6 Sep 2026 12:22:34 -0400 Subject: [PATCH] =?UTF-8?q?docs(labs):=20=D1=83=D1=82=D0=BE=D1=87=D0=BD?= =?UTF-8?q?=D0=B5=D0=BD=D0=BE=20=D0=A2=D0=97=20=D0=BA=D0=BE=D0=BD=D0=B2?= =?UTF-8?q?=D0=B5=D0=B9=D0=B5=D1=80=D0=B0=20=D1=81=D0=B5=D1=81=D1=81=D0=B8?= =?UTF-8?q?=D0=B9?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Зачем: - связать четвертую ступень с результатом третьей и сделать пересборку проверяемой. - Что: - уточнены переход от третьей ступени, договор с дагом и синхронная вставка в донор; - ограничен сохраненный срез строк и вынесена независимая проверка атрибутов; - упрощены формулировки и исправлены инструкции по DBeaver и удалению партиции. - Проверка: - git diff --check; - проверены ссылки документа и якорь README#подключиться-из-dbeaver. - узкая повторная вычитка Claude Opus одобрила все девять исправлений. --- docs/labs/4-1-session-pipeline.md | 122 ++++++++++++++++++------------ 1 file changed, 73 insertions(+), 49 deletions(-) diff --git a/docs/labs/4-1-session-pipeline.md b/docs/labs/4-1-session-pipeline.md index f35ac12..61ce3c1 100644 --- a/docs/labs/4-1-session-pipeline.md +++ b/docs/labs/4-1-session-pipeline.md @@ -12,17 +12,17 @@ ## Граница задания -Стенд уже умеет запускать трансформацию, хранить сессии и проверять их число. +Даг, таблицы сессий и проверка их числа уже созданы. Перепишите только три файла: - `sql/dds/session_scope.sql`; - `sql/dds/session_rebuild.sql`; - `sql/dds/session_replace.sql`. -Удалите из них готовый SQL и соберите свою реализацию. Даг `dds_transform` -передаст ей параметры и запустит запросы в нужном порядке. DDL уже создает -целевые таблицы и донор. Донор — служебная таблица, где день собирается перед -заменой. Запрос `sessions_vs_reference.sql` проверит опубликованный результат. +Удалите из них готовый SQL и соберите свою реализацию. Параметры и порядок +запросов заданы в `dds_transform`. DDL задает целевые таблицы и донор. Донор — +служебная таблица, где день собирается перед заменой. В +`sessions_vs_reference.sql` число сессий сравнивается с `VisitID` из ODS. ## Договор с дагом @@ -31,49 +31,54 @@ | Файл | Вход | Результат | |---|---|---| | `session_scope.sql` | `{position:UInt16}` — номер первого несыгранного дня | `first_day` и `last_day` | -| `session_rebuild.sql` | `{first_day:Date}`, `{last_day:Date}`, `{load_id:String}` | Собранные сессии в `dds.session_stage_dist` | +| `session_rebuild.sql` | `{first_day:Date}`, `{last_day:Date}`, `{load_id:String}` | Сессии в `dds.session_stage_dist` | | `session_replace.sql` | День в `{{ params.day }}` | Партиция дня в `dds.session_rep` | -`session_scope.sql` ищет первый прожитый день, которого еще нет в DDS. От него -до последнего прожитого дня получается непрерывный отрезок. Начало шкалы нужно -взять из событий ODS: `position` считает дни от него. Живой модельный день еще -не закончен, поэтому сессии за него не собираются. +Первый прожитый день, которого еще нет в DDS, становится `first_day`. +`last_day` — последний прожитый день. Начало шкалы нужно взять из событий ODS: +`position` считает дни от него. Живой модельный день еще не закончен, поэтому +сессии за него не собираются. Если в ODS нет ни одного события, запрос не возвращает строк. Если DDS уже -догнал мир, запрос возвращает строку с `first_day = NULL`. Даг поймет, что -работы нет, и пропустит группу сессий. +догнал мир, запрос возвращает строку с `first_day = NULL`. В этом случае группа +сессий в Airflow получает состояние `skipped`. `session_rebuild.sql` сначала очищает `dds.session_stage_rep` на всем кластере. -Затем он собирает в `dds.session_stage_dist` все дни отрезка. Вставка должна -закончиться до замены партиций. +Затем он собирает в `dds.session_stage_dist` все дни отрезка. Добавьте к +вставке `SETTINGS distributed_foreground_insert = 1`. Так строки успеют попасть +на шарды до замены партиций. -`session_replace.sql` заменяет один день. Цель — локальная таблица -`dds.session_rep`, источник — локальный донор `dds.session_stage_rep`. Операция -выполняется на всем кластере. +Цель `session_replace.sql` — локальная таблица `dds.session_rep`, источник — +локальный донор `dds.session_stage_rep`. Операция выполняется на всем кластере. -`session_scope.sql` читается как один запрос, поэтому в конце ему не нужна -точка с запятой. День для замены подставляет шаблон Jinja в Airflow. Параметр +Не ставьте точку с запятой в конце `session_scope.sql`. Драйвер добавляет формат +ответа, и точка с запятой превратит строку в мультизапрос, который ClickHouse +отклонит. День для замены подставляет шаблон Jinja в Airflow. Параметр ClickHouse в `ALTER ... PARTITION` здесь не работает. ## Нарежьте сессии Источник сборки — `dds.event_v`. Одна строка результата представляет один -визит. Используйте правила, которые уточнили на третьей ступени: +визит. На третьей ступени вы уже нарезали сессии: - сессия не пересекает модельные сутки; - пауза больше 30 минут открывает новую сессию; -- в нарезке участвуют события всех типов; -- при одинаковом времени порядок событий задает `event_id`, поэтому полный - порядок окон — `(event_time, event_id)`; -- `session_id` детерминирован и зависит от куки и начала сессии. +- в нарезке участвуют события всех типов. + +Пересборка добавляет требования к повторяемости. При одинаковом времени +порядок событий задает `event_id`, поэтому полный порядок окон — +`(event_time, event_id)`. Один визит должен получать одинаковый `session_id` +при каждом запуске. Тогда его можно сравнить с сохраненным результатом. +Вычислите идентификатор как `cityHash64(client_id, min(event_time))`. Состав и типы строки заданы в `dds.session_rep`. Заполните начало, конец и длительность. Посчитайте все события, просмотры, добавления в корзину и покупки. Возьмите входную и выходную страницы, UTM первого события, устройство и город. Служебные колонки тоже входят в договор таблицы. -Пишите через `dds.session_stage_dist`. Эта таблица разложит строки по ключу -донора. +Пишите через `dds.session_stage_dist`: место строки определит ключ +`client_id`. При записи прямо в локальную таблицу оно зависело бы от ноды, +которая выполнила запрос. ## Запустите свою реализацию @@ -85,42 +90,57 @@ SELECT max(session_date) FROM dds.session_v; ``` -Сохраните десять строк этого дня без служебных колонок загрузки: +Берите последний день: более ранняя дыра расширит объем до всех последующих +прожитых дней. + +Сохраните десять строк с идентификатором, границами, длительностью и счетчиками: ```sql -SELECT * EXCEPT (_load_id, _load_ts) +SELECT + session_id, + client_id, + session_date, + started_at, + finished_at, + duration_seconds, + events_total, + pageviews, + cart_adds, + purchases FROM dds.session_v WHERE session_date = 'YYYY-MM-DD' -ORDER BY client_id, started_at +ORDER BY client_id, started_at, session_id LIMIT 10; ``` -Подставьте найденную дату и удалите только ее партицию. Команда идет под -служебным пользователем `etl`, потому что `analyst` имеет доступ только на -чтение: +Создайте в DBeaver подключение к первой ноде по [инструкции из +README](../../README.md#подключиться-из-dbeaver). Укажите пользователя `etl` и +пароль `CLICKHOUSE_ETL_PASSWORD` из `.env`. У этой учетной записи есть право +менять таблицы DDS на кластере. Подставьте найденную дату и удалите только ее +партицию: -```bash -docker compose exec -T clickhouse-01 sh -c \ - 'clickhouse-client --user etl --password "$CLICKHOUSE_ETL_PASSWORD" --query "$1"' \ - _ "ALTER TABLE dds.session_rep ON CLUSTER clickstream_cluster DROP PARTITION 'YYYY-MM-DD'" +```sql +ALTER TABLE dds.session_rep ON CLUSTER clickstream_cluster +DROP PARTITION 'YYYY-MM-DD'; ``` Откройте Airflow и запустите `etl_pipeline`. В даге `dds_transform` раскройте группу `session`: в ней должны выполниться задачи `scope` и `rebuild`, а также `replace` для выбранного дня. Если после третьей ступени мир ушел вперед, -Airflow создаст `replace` и для последующих прожитых дней. Следом конвейер -запустит витрины и `dq_check`. +в группе появятся задачи `replace` и для последующих прожитых дней. Следом +выполнятся витрины и `dq_check`. Если прогон дошел до замены, но упал на проверке, ошибочная партиция уже находится в DDS. После правки снова удалите тот же день перед следующим -запуском. Иначе расчет объема сочтет день готовым и пропустит вашу новую -сборку. +запуском. Иначе день останется заполненным, а новая сборка для него не +выполнится. -## Проверьте результат +## Как проверить себя -Сначала повторите запрос с десятью строками. `session_id` и все деловые -колонки должны совпасть с сохраненным результатом. `_load_id` и `_load_ts` -изменятся: день собрал другой запуск. +Сначала повторите запрос с десятью строками. `session_id`, границы, +длительность и счетчики должны совпасть с сохраненным результатом. Так вы +завершите отложенное с третьей ступени сопоставление со стендовой нарезкой по +`session_id`. Задача `dq_check.sessions_vs_reference` должна стать зеленой. Та же проверка видна в хранилище: @@ -136,7 +156,11 @@ ORDER BY data_date; `failed_rows` не подтверждает границы, страницы и UTM: две ошибки могут оставить тот же счет. Возьмите поштучную сверку из третьей ступени. Для одной куки выбранного дня сопоставьте границы `VisitID` по событиям с границами -строк `dds.session_v`. Проверьте начало, конец и атрибуты сессии. +строк `dds.session_v`. По исходным событиям проверьте страницы, UTM, устройство +и город. Страницы и UTM не входят в сохраненные десять строк: готовый SQL +стенда выбирал их через `argMin` и `argMax` только по `event_time`. При равном +времени ожидаемое значение определяйте по исходным событиям и полному порядку +`(event_time, event_id)`. Осталась физическая раскладка. Напишите отдельный запрос к `dds.session_dist`. Для каждой строки он должен: @@ -147,7 +171,7 @@ ORDER BY data_date; У кластера два равных шарда: остаток 0 соответствует `clickhouse-01`, остаток 1 — `clickhouse-02`. Сгруппируйте ответ по фактической ноде. На каждой ноде -число строк на чужом шарде должно быть нулевым. Счетная DQ-проверка читает +число строк на чужом шарде должно быть нулевым. Проверка числа сессий читает Distributed-таблицу и такую ошибку не заметит. После зеленого результата запустите `etl_pipeline` еще раз, не удаляя @@ -158,10 +182,10 @@ Distributed-таблицу и такую ошибку не заметит. - [ ] переписаны только три файла `sql/dds/session_*.sql`, инфраструктура упражнения не изменена; -- [ ] адресный прогон восстановил выбранный день и завершил +- [ ] запуск после удаления партиции восстановил выбранный день и завершил `etl_pipeline` успешно; -- [ ] сохраненные `session_id` и деловые колонки десяти строк совпали после - пересборки; +- [ ] сохраненные `session_id`, границы, длительность и счетчики десяти строк + совпали после пересборки; - [ ] `sessions_vs_reference` дает ноль расхождений во всех прожитых днях; - [ ] для одной куки вручную сверены границы и атрибуты сессии; - [ ] пересчет ключа по `hostName()` не нашел строк на чужом шарде;