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

Перепиши нарезку стенда

Чему учит: изменить расчетную сущность внутри готового конвейера и проверить результат внешним эталоном.

Сначала прочитать

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

Граница задания

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

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

Удалите из них готовый SQL и соберите свою реализацию. Параметры и порядок запросов заданы в dds_transform. DDL задает целевые таблицы и донор. Донор — служебная таблица, где день собирается перед заменой. В sessions_vs_reference.sql число сессий сравнивается с VisitID из ODS.

Договор с дагом

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

Файл	Вход	Результат
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

Первый прожитый день, которого еще нет в DDS, становится first_day. 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, границы, длительность и счетчики должны совпасть с сохраненным результатом. Так вы завершите отложенное с третьей ступени сопоставление со стендовой нарезкой по 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;

Эта сверка сравнивает число сессий каждой куки за день с VisitID. Нулевой failed_rows не подтверждает границы, страницы и UTM: две ошибки могут оставить тот же счет. Возьмите поштучную сверку из третьей ступени. Для одной куки выбранного дня сопоставьте границы 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 дает ноль расхождений во всех прожитых днях;
• для одной куки вручную сверены границы и атрибуты сессии;
• пересчет ключа по hostName() не нашел строк на чужом шарде;
• повторный прогон без новой дыры пропустил группу сессий.