clickstream-ch-kafka-superset-demo/docs/course/lessons/08_lab_continue.md

# Лаба 08. Живой поток и свежесть данных

> Формат: **практика** — запустишь живой поток, остановишь его и продолжишь с
> сохранённого состояния.
> Пререквизит: пройдена лаба 07; стенд возвращён к эталонному миру, а
> `world_next_day` стоит на паузе.
> Эталонные пути:
> [`Makefile`](../../../Makefile) и
> [`airflow/dags/etl_pipeline_dag.py`](../../../airflow/dags/etl_pipeline_dag.py).
>
> Поток данных одной строкой:
> `generator-continue → Kafka → STG (сразу) → etl_pipeline (по запуску) → ODS → DDS → DM`
>
> О чём лаба простыми словами: события будут приезжать непрерывно. Ты увидишь,
> что Kafka и STG уже обновились, а аналитические слои ещё ждут пакетного
> запуска.

---

## 1. Зачем и где в проде

У данных есть **свежесть** — задержка между событием в источнике и моментом,
когда его увидит потребитель. Для одного отчёта допустимы сутки, для другого
нужны минуты. Обещание «данные готовы не позже чем через N минут» влияет на
всю цепочку.

В этом стенде приём потоковый: ClickHouse постоянно читает Kafka, а
Materialized View сразу складывает сообщения в STG. Дальнейшая обработка
пакетная: ODS, DDS и DM обновляет только запуск `etl_pipeline`.

Поэтому **потоковый приём не равен потоковой обработке**. Новые строки уже
могут лежать в STG, пока витрина для аналитика показывает старую правую
границу.

> **В проде иначе.** Требование к свежести задают вместе с потребителем данных.
> Под него выбирают расписание, размер порции, проверки и допустимое
> отставание. Слово «поток» само по себе ничего не обещает о времени появления
> данных в отчёте.

---

## 2. Руки: оживляем мониторинг и догоняем витрины

Подготовь стенд по
[канонической инструкции курса](../README.md#подготовка-и-канонический-сброс).
После `world_init` живой генератор выключен, а эталонный мир заканчивается на
одной и той же модельной границе у всех менти.

### Запоминаем начальную свежесть

Открой SQL-консоль `http://localhost:9123/play` (пользователь `default`, пароль
`123456`) и выполни:

```sql
SELECT
    'STG' AS layer,
    max(parseDateTime64BestEffortOrNull(
        JSONExtractString(raw, 'event_timestamp'), 6, 'UTC'
    )) AS max_event_ts
FROM stg.browser_raw

UNION ALL

SELECT 'ODS', max(event_ts)
FROM ods.browser_event

UNION ALL

SELECT 'DDS', max(event_ts)
FROM dds.event

UNION ALL

SELECT 'DM', max(event_ts)
FROM dm.v_events_enriched;
```

На эталонном мире правые границы слоёв должны совпадать.
<!-- сверить-на-стенде -->

### Запускаем живое продолжение

В терминале выполни:

```bash
make generator-continue
```

Команда запускает генератор в режиме `continue`: он берёт сохранённое состояние
эталонного мира и продолжает его, не создавая новый мир с нуля.

Вернись к экранам из урока 5:

- в Prometheus target `generator` переходит в `UP`;
- в `Generator Overview` оживает `Total Events/min (all 4 topics)`;
- в `Kafka Overview` растут сообщения и конечные offset-ы партиций;
- в Kafka UI из урока 0 у новых сообщений растут offset-ы;
- `Kafka No Messages Produced` выходит из `Alerting`, когда Grafana увидит
  устойчивый поток.

У алерта есть окно `for: 10m`: в `Alerting` он переходит только после десяти
минут тишины. Когда поток вернётся, условие перестанет выполняться при
ближайшей оценке. График событий и offset-ы должны начать двигаться раньше.

Подожди несколько минут и снова выполни запрос по слоям. Правая граница STG
уйдёт вперёд. ODS, DDS и DM останутся на границе последнего
`etl_pipeline`.

### Догоняем пакетные слои

Открой Airflow и запусти `etl_pipeline` через **Trigger DAG** с пустой формой.
По умолчанию это полный пересчёт.

После зелёного прогона повтори запрос. ODS, DDS и DM догонят данные, которые
успели попасть в STG к началу обработки. Генератор всё ещё работает, поэтому
STG вскоре снова может оказаться чуть свежее. Это ожидаемое расслоение, а не
потеря строк.

---

## 3. Загляни внутрь

Открой [`Makefile`](../../../Makefile) и найди две команды этой лабы:

```make
generator-continue:
	COMPOSE_BIN="$(COMPOSE)" bash ./scripts/run_generator.sh continue

generator-down:
	$(COMPOSE) stop generator
```

`generator-down` останавливает только контейнер генератора. Kafka, ClickHouse и
читатели STG продолжают работать. Поэтому они успевают дочитать уже записанные
сообщения.

`generator-continue` запускает генератор с сохранением прежнего состояния. В нём
остаются популяция пользователей, активные визиты, случайное состояние и
модельные часы. После рестарта продолжается тот же мир.

<a id="modelnoe-vremya"></a>
> **Модельное время.** Профиль `daily-wave` ускоряет внутренние часы стенда в
> 60 раз: модельные сутки проходят примерно за полчаса. Поэтому «сейчас» на
> дашборде может обгонять настенные часы. Ручка тренажёра называется
> `GEN_MODEL_TIME_SPEED`; в этой лабе её не меняем.

### Где проходит граница между потоком и пакетом

В уроке 1 ты видел Kafka engine и Materialized View. Пока сервисы работают,
этот путь сам переносит сообщения из Kafka в `stg.*_raw`.

У `etl_pipeline` в
[`airflow/dags/etl_pipeline_dag.py`](../../../airflow/dags/etl_pipeline_dag.py)
стоит `schedule=None`. ODS, DDS и DM не догоняют поток сами: нужен ручной
запуск DAG. Получается две разные гарантии свежести:

- Kafka → STG работает постоянно;
- STG → ODS → DDS → DM обновляется по запуску `etl_pipeline`.

Если потребителю нужна витрина не старше пяти минут, ручной DAG такую
гарантию не даёт. Понадобится расписание или другая обработка, а затем
измерение фактического отставания.

---

## 4. Управляемая правка: останавливаем и продолжаем поток

Проверим, переживает ли мир остановку генератора.

### Шаг 1. Зафиксируй состояние до остановки

Пока генератор работает, запиши:

- правую границу STG из запроса секции 2;
- значение `Total Events/min (all 4 topics)`;
- текущие конечные offset-ы в `Kafka Overview`.

### Шаг 2. Останови генератор

```bash
make generator-down
```

После остановки:

- target `generator` переходит в `DOWN`;
- `Total Events/min (all 4 topics)` падает к нулю или перестаёт обновляться;
- конечные offset-ы перестают расти;
- читатели ClickHouse дочитывают уже записанные сообщения, и отставание
  стекает к нулю;
- после окна ожидания `Kafka No Messages Produced` переходит в `Alerting`.

Как и в уроке 5, Kafka UI или панель lag могут не показать прогресс групп
`ch_stg_*`. Тогда смотри `system.kafka_consumers` со стороны ClickHouse.
Главный признак здесь такой: новые offset-ы больше не появляются, а строки,
уже записанные до остановки, доходят в STG.

### Шаг 3. Продолжи тот же мир

```bash
make generator-continue
```

Проверь обратную картину:

- target `generator` снова в `UP`;
- график событий и offset-ы снова растут;
- `Kafka No Messages Produced` возвращается в норму при ближайшей оценке;
- правая граница STG стала больше значения, записанного до остановки.

Генератор взял прежнюю популяцию и прежнюю точку продолжения. Остановка не
превратила поток в новый независимый мир.

### Шаг 4. Догони аналитику и увидь расхождение мира

Запусти `etl_pipeline` с пустой формой. После зелёного прогона выполни:

```sql
SELECT
    uniqExact(user_domain_id) AS users,
    uniqExact(click_id) AS sessions
FROM dm.v_events_enriched
WHERE user_domain_id IS NOT NULL;
```

В живом мире один пользователь может вернуться и открыть новый визит. Поэтому
после достаточного продолжения ожидаем `users < sessions`. На эталонной
статике было `users == sessions`.

Точные числа здесь не печатаем: длительность живого запуска у каждого своя.
`users = <значение>`, `sessions = <значение>`.
<!-- сверить-на-стенде -->

Если равенство ещё сохранилось, дай генератору поработать несколько минут,
снова запусти `etl_pipeline` и повтори запрос. Важно увидеть сам возврат
пользователя, а не угадать конкретное число.

### Верни как было

Останови живой поток:

```bash
make generator-down
```

После `continue` мир стал недетерминированным: его числа зависят от длительности
запуска. Верни стенд к эталону по
[канонической инструкции курса](../README.md#подготовка-и-канонический-сброс).
Не пересказывай шаги по памяти: эта ссылка остаётся единственным учебным
описанием полного сброса.

---

## 5. Проверь себя

| Действие | Где смотреть | Что ожидать |
|----------|--------------|-------------|
| выполнить `make generator-continue` | Prometheus Targets | `generator` переходит в `UP` |
| посмотреть живой генератор | `Generator Overview` | `Total Events/min (all 4 topics)` становится ненулевым |
| сравнить свежесть до ETL | SQL-запрос по слоям | STG свежее ODS, DDS и DM <!-- сверить-на-стенде --> |
| запустить `etl_pipeline` | Airflow и SQL-запрос | пакетные слои догоняют снимок STG |
| выполнить `make generator-down` | Grafana и Kafka | события перестают поступать, отставание читателей стекает к нулю |
| снова выполнить `make generator-continue` | Grafana, Kafka и STG | поток продолжается, offset-ы и правая граница снова растут |
| пересчитать аналитику | запрос `users` и `sessions` | после возвратов пользователей выполняется `users < sessions` <!-- сверить-на-стенде --> |
| остановить поток и пройти сброс | Grafana и SQL | `generator` выключен, слои снова совпадают с эталонным миром |

Ответь своими словами:

- что означает свежесть данных и кто задаёт требование к ней;
- почему новые строки уже есть в STG, но их ещё нет в DDS;
- почему запуск `etl_pipeline` догоняет поток лишь до очередного снимка;
- что сохраняет режим `continue`;
- почему после живого продолжения числа разных менти расходятся;
- откуда берётся `users < sessions`.

---

## 6. Что должно получиться

После лабы сохрани:

- скрин Prometheus Targets с `generator` в `UP`;
- скрин живого `Total Events/min (all 4 topics)`;
- два результата запроса свежести: до и после `etl_pipeline`;
- наблюдение остановки и продолжения по offset-ам или правой границе STG;
- результат запроса с `users < sessions`;
- короткое объяснение своими словами: почему потоковый приём не гарантирует
  потоковую обработку до витрины.

В конце `generator` должен быть остановлен, а стенд — возвращён к эталонному
миру.

---

## Мост после курса

Теперь у тебя есть два способа растить один мир: повторяемая суточная порция и
живое продолжение. Следующий практический вопрос уже зависит от продукта:
какую свежесть обещать потребителю и какую цену платить за пересчёт.