clickstream-data-platform/docs/labs/4-1-session-pipeline.md

Расчет сессий в стенде

На прошлой ступени вы рассчитали сессии отдельным запросом. Теперь перепишете расчет внутри стенда, запустите его через Airflow и проверите результат по исходным событиям.

Подготовка

• задание «Сессии из событий»;
• модель сессии DDS;
• дневной конвейер трансформаций;
• раскладка по шардам.

Файлы для изменения

Даг, таблицы сессий и проверка их числа уже созданы. Перепишите только три файла:

• sql/dds/session_scope.sql;
• sql/dds/session_rebuild.sql;
• sql/dds/session_replace.sql.

Удалите из них готовый SQL и напишите свой. Даг dds_transform передаст параметры и выполнит запросы по порядку. Таблицы уже созданы: основные хранят готовые сессии, промежуточные нужны для нового расчета перед заменой данных. Запрос sessions_vs_reference.sql сравнивает число сессий с VisitID из ODS.

Выполнение запросов в Airflow

Объем прогона — дни, которые соберет один запуск. Его вычисляет первый файл.

Файл	Вход	Результат
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_replace.sql	День в {{ params.day }}	Партиция дня в dds.session_rep

first_day должен быть первым завершившимся модельным днем, которого еще нет в DDS, а last_day последним завершившимся днем. Начальную дату возьмите из событий ODS: position считает дни от нее. Текущий модельный день еще не закончен, поэтому сессии за него не рассчитываются.

Если в ODS нет ни одного события, запрос не возвращает строк. Если все завершившиеся дни уже есть в DDS, запрос возвращает строку с first_day = NULL. Тогда Airflow пропускает группу задач расчета сессий: ее состояние будет skipped.

session_rebuild.sql сначала очищает dds.session_stage_rep на всем кластере. Затем он записывает в dds.session_stage_dist расчет за все выбранные дни. Добавьте к вставке SETTINGS distributed_foreground_insert = 1. Так строки успеют попасть на шарды до замены партиций.

Запрос session_replace.sql заменяет партицию в основной dds.session_rep готовой партицией из промежуточной dds.session_stage_rep. Операция выполняется на всем кластере, на локальных таблицах каждого шарда.

Не ставьте точку с запятой в конце session_scope.sql. Драйвер добавляет формат ответа. С точкой с запятой получится несколько команд, которые ClickHouse отклонит. День для замены подставляет шаблон Jinja в Airflow. Параметр ClickHouse в ALTER ... PARTITION здесь не работает.

Правила расчета сессий

Источник сборки — dds.event_v. Одна строка результата представляет один визит. На третьей ступени вы уже получили правила разделения событий на сессии:

• сессия не пересекает модельные сутки;
• пауза больше 30 минут открывает новую сессию;
• в расчете участвуют события всех типов.

Повторный расчет должен давать тот же результат. При одинаковом времени упорядочивайте события по event_id: порядок в оконных функциях должен быть (event_time, event_id). Один визит должен получать одинаковый session_id при каждом запуске. Вычислите идентификатор как cityHash64(client_id, min(event_time)).

Состав и типы строки заданы в dds.session_rep. Заполните начало, конец и длительность. Посчитайте все события, просмотры, добавления в корзину и покупки. Возьмите входную и выходную страницы, UTM первого события, устройство и город. Заполните также служебные колонки из схемы таблицы.

Пишите через dds.session_stage_dist: шард для строки определяется по client_id. При записи прямо в локальную таблицу он зависел бы от узла, который выполнил запрос.

Запуск нового расчета

Готовые дни повторно не собираются. Чтобы новый SQL точно выполнился, выберите последний собранный день:

SELECT max(session_date)
FROM dds.session_v;

Берите последний день: если удалить более ранний, расчет придется повторить и за все последующие завершившиеся дни.

Сохраните десять строк с идентификатором, границами, длительностью и счетчиками:

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, session_id
LIMIT 10;

Создайте в DBeaver подключение к первому узлу по инструкции из README. Укажите пользователя etl и пароль CLICKHOUSE_ETL_PASSWORD из .env. У этой учетной записи есть право менять таблицы DDS на кластере. Подставьте найденную дату и удалите только ее партицию:

ALTER TABLE dds.session_rep ON CLUSTER clickstream_cluster
DROP PARTITION 'YYYY-MM-DD';

Откройте Airflow и запустите etl_pipeline. В даге dds_transform раскройте группу session: в ней должны выполниться задачи scope и rebuild, а также replace для выбранного дня. Если после третьей ступени мир ушел вперед, в группе появятся задачи replace и для последующих прожитых дней. Следом обновятся витрины и выполнится dq_check.

Если прогон дошел до замены, но упал на проверке, ошибочная партиция уже находится в DDS. После правки снова удалите тот же день перед следующим запуском. Иначе день останется заполненным, а новая сборка для него не выполнится.

Проверка результата

Сначала повторите запрос с десятью строками. session_id, границы, длительность и счетчики должны совпасть с сохраненным результатом. Это проверит, что ваш расчет восстановил те же сессии с теми же идентификаторами.

Задача dq_check.sessions_vs_reference должна стать зеленой. Та же проверка видна в хранилище:

SELECT data_date, reference_rows, actual_rows, failed_rows, status
FROM dm.dq_summary_v
WHERE check_name = 'sessions_vs_reference'
ORDER BY data_date;

Эта сверка сравнивает число сессий каждого client_id за день с VisitID. При failed_rows = 0 число сессий совпало, но их границы, страницы и UTM еще могут различаться. Как на третьей ступени, проверьте отдельные сессии. Для одного client_id выбранного дня найдите начало и конец каждого VisitID по событиям и сравните с границами строк dds.session_v. По исходным событиям проверьте страницы, UTM, устройство и город. Страницы и UTM не входят в сохраненные десять строк: готовый SQL стенда выбирал их через argMin и argMax только по event_time. При равном времени ожидаемое значение определяйте по исходным событиям и полному порядку (event_time, event_id).

Проверьте также, на каком шарде лежит каждая строка. Напишите отдельный запрос к dds.session_dist. Он должен:

1. получить фактический узел строки через hostName();
2. определить ожидаемый узел по cityHash64(client_id);
3. посчитать строки, у которых фактический и ожидаемый узлы не совпали.

У кластера два равных шарда: остаток 0 соответствует clickhouse-01, остаток 1 — clickhouse-02. Сгруппируйте ответ по фактическому узлу. На каждом узле число строк на чужом шарде должно быть нулевым. Проверка числа сессий читает Distributed-таблицу и такую ошибку не заметит.

После зеленого результата запустите etl_pipeline еще раз, не удаляя партицию. Группа сессий должна быть пропущена: все завершившиеся дни уже рассчитаны, новых дней и пропусков нет.

Результат работы

• вы переписали три файла sql/dds/session_*.sql, сохранив даг, таблицы и проверку стенда;
• ваш расчет восстановил удаленный день, etl_pipeline завершился успешно;
• вы сверили десять сохраненных строк: session_id, границы, длительность и счетчики совпали после пересборки;
• вы получили ноль расхождений в sessions_vs_reference за все прожитые дни и вручную сверили границы и поля сессии одного client_id;
• ваш запрос с hostName() не нашел строк на чужом шарде;
• повторный прогон без удаления данных пропустил группу сессий.