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

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

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

## Подготовка

- [задание «Сессии из событий»](3-1-sessions-by-hand.md);
- [модель сессии DDS](../architecture/dds/session.md);
- [дневной конвейер трансформаций](../architecture/etl.md#объём-прогона);
- [раскладка по шардам](../architecture/storage.md#раскладка-по-шардам).

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

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

- `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 точно выполнился, выберите
последний собранный день:

```sql
SELECT max(session_date)
FROM dds.session_v;
```

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

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

```sql
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](../../README.md#подключиться-из-dbeaver). Укажите пользователя `etl` и
пароль `CLICKHOUSE_ETL_PASSWORD` из `.env`. У этой учетной записи есть право
менять таблицы DDS на кластере. Подставьте найденную дату и удалите только ее
партицию:

```sql
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` должна стать зеленой. Та же проверка
видна в хранилище:

```sql
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()` не нашел строк на чужом шарде;
- [ ] повторный прогон без удаления данных пропустил группу сессий.