diff --git a/docs/labs/6-1-event-parsing.md b/docs/labs/6-1-event-parsing.md new file mode 100644 index 0000000..b817729 --- /dev/null +++ b/docs/labs/6-1-event-parsing.md @@ -0,0 +1,214 @@ +# Разберите сырые события рядом со стендом + +**Чему учит:** пакетно разобрать сообщения отдельно от живого приема и +построчно проверить свой результат по нетронутому ODS. + +## Сначала прочитать + +- [формат события](../formats/clickstream-event.md); +- [справочник хранилища](../architecture/storage.md), разделы про имена + объектов, актуальные события, прием потока и срок жизни сырых данных; +- [ADR о приеме событий](../adr/0005-event-ingestion.md), раздел с принятым + решением; +- схему [`ods.event_rep`](../../sql/ddl/20-ods-tables.sql). + +Готовый стендовый разбор находится в `sql/ddl/30-ods-views.sql`. Откройте его +после собственной сверки: в нем лежит ответ на основную часть задания. + +## Граница задания + +Создайте в базе `default` одну свою таблицу годных событий и наполните ее одним +пакетным `INSERT SELECT` из `stg.hits_raw_dist`. Работайте с первой нодой под +пользователем `etl`. + +Матвью `ods.event_mv` и `ods.event_errors_mv` продолжают принимать живой +поток. Оставьте их, STG, ODS, Kafka и даги без изменений. Пересоздание матвью +на работающем стенде оставило бы зазор в приеме, а для пакетного опыта оно не +нужно. + +Дайте своей таблице имя, которое не пересечется с чужой работой, и запишите +его вместе с запросами. В конце вы удалите только эту таблицу. + +## Выберите сырой срез + +STG хранит строки примерно три дня по `_load_ts`. Эта колонка отмечает время +поступления в хранилище. Модельный день события лежит в `EventDate`. Сначала +найдите среди недавних данных завершившийся интервал со строками. Посмотрите +счет по минутам: + +```sql +SELECT + toStartOfMinute(_load_ts) AS minute, + count() AS rows +FROM stg.hits_raw_dist +WHERE _load_ts >= now() - INTERVAL 30 MINUTE + AND _load_ts < toStartOfMinute(now()) +GROUP BY minute +ORDER BY minute DESC; +``` + +Выберите короткий интервал `[slice_from, slice_to)`, который уже закончился и +дает удобное для построчной проверки число строк. Если одна минута велика, +сузьте ее точными границами `_load_ts`. Если недавних строк нет, откройте +Airflow. При включенном `world_live` посмотрите его текущий запуск и дочерний +`world_live_day`: дождитесь свежих строк от текущего или следующего планового +запуска, а при ошибке сначала разберите ее причину. Запустите `world_live_day` +вручную, только если `world_live` стоит на паузе и активного `world_live_day` +нет. Успешный прогон распределяет события модельного дня по реальным минутам. + +Зафиксируйте до разбора: + +- обе границы с миллисекундами и число строк; +- минимальный и максимальный `kafka_offset` по каждой паре + `kafka_topic`, `kafka_partition`. + +Во всех дальнейших запросах читайте `stg.hits_raw_dist` только с этими +границами. Новые строки с более поздним `_load_ts` в срез не попадут. Повторите +опыт до очистки по TTL; после нее прежний срез уже не восстановить из ODS. + +## Соберите собственную таблицу + +Таблица содержит все 47 полей источника с именами и типами +`ods.event_rep`, а также `_load_ts`. Тип `purchaseRevenue` возьмите из ODS: +`Array(Decimal(18, 2))`. + +Схлопните повтор доставки по договору ODS. Возьмите +`ReplacingMergeTree(_load_ts)`, партицию `EventDate` и полный ключ замены: +`(CounterID, EventDate, intHash32(ClientID), WatchID)`. Достаточно одной +локальной таблицы: отдельная Distributed-таблица не добавит урока к пакетному +опыту на одной ноде. + +`WatchID` — логический идентификатор события, уникальный по +[договору генератора](../specs/2026-08-01-generator.md). Физический ключ ODS +шире: в `ReplacingMergeTree` версиями считаются строки с одинаковым полным +ключом сортировки. Используйте его при загрузке и сверке. + +## Разберите годные строки + +Для каждой строки среза примите решение по тому же строгому входу, что и +стенд: + +1. корень `raw` — JSON-объект; +2. набор верхнеуровневых ключей в точности равен 47 ключам формата; +3. `WatchID`, `VisitID`, `ClientID`, `EventDate` и `UTCEventTime` разбираются + в ожидаемые типы. + +Напишите разбор, который не прерывает пакет из-за одной грязной строки. Сами +выберите способ безопасного извлечения пяти опорных полей, форму общих выражений +и место фильтра. Остальные поля извлеките по договору формата. + +Перенесите `_load_ts` из STG без пересчета: это время исходной доставки и +колонка версии. Время запуска вашего запроса сюда не входит. Вставьте в свою +таблицу только годные строки выбранного среза. + +Считайте отвергнутые строки прямо в исходном срезе: отдельная таблица брака в +задание не входит. Проверьте работу строгого входа одним запросом. Посчитайте: + +- все доставки; +- строки, прошедшие ваш предикат; +- строки, отвергнутые его отрицанием. + +Проверьте равенство общего числа строк сумме прошедших и отвергнутых. Так каждая +исходная строка учитывается ровно в одной группе. Если отвергнутые строки есть, +сохраните их `raw`, `_load_ts` и координаты Kafka. Найдите те же +координаты в `ods.event_errors_dist` и запишите стендовый `error_class`. Если +брака нет, так и зафиксируйте: опыт проверил успешную ветвь и выполнение +предиката на этом срезе, но не классификатор ошибок. Подкладывать грязные +сообщения ради задания не требуется. + +## Приведите стороны к одному зерну + +В STG одна строка представляет доставку. Повтор сообщения остается второй +строкой. В `ods.event_v` одна строка представляет актуальное событие, потому +что представление читает физическую таблицу с `FINAL`. + +Свою сторону тоже читайте с `FINAL`. Ожидаемый счет равен числу различных +полных ключей среди годных доставок. Разница между числом доставок и этим +счетом показывает повторы в исходном срезе. + +Область эталона постройте независимо от своего предиката и результата вставки. +Снова возьмите весь исходный STG-срез. По координатам Kafka исключите доставки, +которые есть в `ods.event_errors_dist`. У оставшихся доставок извлеките +`WatchID` без риска прервать запрос и выберите эти события из свежего +`ods.event_v`. + +Свяжите сырую доставку с ODS по `WatchID`: по договору генератора это уникальный +идентификатор события. Он задает только область эталона. Дедупликация и +последующее соединение используют полный физический ключ. Не берите область из +своей таблицы и не применяйте к ней свой предикат. Иначе ошибочно отвергнутая +строка исчезнет с обеих сторон сверки. + +## Сверьте 47 полей + +Соедините свою таблицу и выбранные строки `ods.event_v` через `FULL OUTER +JOIN` по полному ключу замены. Полное соединение сохраняет строки, +которых нет на одной из сторон. Разделите результат на три исхода: + +- строка есть только в ODS — вы ее потеряли; +- строка есть только у вас — вы приняли лишнее или иначе разобрали часть + ключа; +- ключ есть с обеих сторон, но различается хотя бы одно из 47 полей. + +ClickHouse по умолчанию заполняет отсутствующую сторону значениями типов. +Включите для сверки `join_use_nulls = 1` или добавьте явный признак присутствия +в каждую сторону. Не определяйте отсутствие по значениям полей события. + +Сравните все поля явно, включая массивы, `ecommerce` и точное десятичное +`purchaseRevenue`. Удобный вид диагностики выберите сами: важно, чтобы по +результату можно было назвать отличающиеся колонки одной строки. Выведите число +строк каждого из трех исходов. В верном разборе все три числа равны нулю. + +`_load_ts` не входит в 47 полей источника и не проверяет JSON-разбор. +Проверьте его отдельно. Свяжите каждую строку своей таблицы с меткой одной из +доставок того же полного ключа в исходном срезе. После `FINAL` остается самая +поздняя метка этого ключа внутри среза. С ODS ее не сравнивайте: там может +храниться более поздняя доставка того же события из-за границ среза. + +## Расследуйте расхождение + +Если сверка не пуста, возьмите одну строку. Вернитесь к независимому списку +ключей и исходному `raw`, выведите рядом координаты Kafka и `_load_ts`. + +Для измененной строки назовите поля, которые расходятся, и сравните каждое с +сырой строкой. Для пропавшей или лишней строки проверьте пять опорных полей и +решение строгого предиката. Если координаты доставки есть в +`ods.event_errors_dist`, посмотрите `error_class`. + +Исправьте разбор, удалите и заново создайте только собственную таблицу, затем +повторите вставку того же среза. Остановитесь, когда счет исходных доставок по +прежним границам не изменился, баланс строгого входа сошелся, а построчная +сверка пуста. + +## Где заканчивается проверка + +`ods.event_v` — образец для сравнения. Оба разбора могли повторить одну +ошибку. Для пяти опорных полей у вас есть проверка разбираемости и исходный +JSON. Остальные 42 поля можно вручную сопоставить с `raw`, однако второго +независимого источника для всех значений на стенде нет. + +## Уберите свой объект + +Сохраните DDL таблицы, запрос разбора, границы среза, баланс строгого входа и +пустую итоговую сверку. Затем удалите свою таблицу командой `DROP TABLE` с ее +точным именем. Убедитесь через `system.tables`, что объекта больше нет. +Прикладные строки стенда вы не меняли, поэтому удалять данные из STG, ODS или +Kafka после опыта не нужно. + +## Приемка + +- [ ] записаны границы малого завершенного среза, его счет и координаты + доставки по партициям Kafka; +- [ ] создана одна собственная таблица со всеми 47 полями, перенесенным + `_load_ts` и физическим договором дедупликации ODS; +- [ ] пакетный разбор принимает только JSON-объекты с точным набором ключей и + пятью разбираемыми опорными полями; +- [ ] счет исходных строк равен сумме годных и отвергнутых, а найденный брак + рассмотрен по `raw`, координатам и стендовому классу; +- [ ] область ODS получена из всех координат исходного среза и стендовой + таблицы ошибок, независимо от собственного предиката и таблицы; +- [ ] обе стороны приведены к зерну полного физического ключа; +- [ ] `FULL OUTER JOIN` сравнивает все 47 полей, а числа пропавших, лишних и + измененных строк равны нулю; +- [ ] отдельно подтвержден перенос `_load_ts` и учтены повторы доставки; +- [ ] границы стендового оракула и TTL записаны вместе с результатом; +- [ ] собственная таблица удалена, объекты и данные стенда не изменены. diff --git a/docs/labs/README.md b/docs/labs/README.md index 4aaaf29..39d4c8a 100644 --- a/docs/labs/README.md +++ b/docs/labs/README.md @@ -13,10 +13,10 @@ | 3 | [Нарежь сессии сам](3-1-sessions-by-hand.md) | сверить число сессий в своей нарезке с эталоном и найти события, из-за которых оно расходится | | 4 | [Перепиши нарезку стенда](4-1-session-pipeline.md) | изменить расчетную сущность внутри готового конвейера и проверить ее внешним эталоном | | 5 | [Сначала сторож, потом ломай: заказы](5-1-order-checks.md)
[Перепишите разбор заказов в ODS](5-2-order-parsing.md)
[Перепишите сборку заказов в DDS](5-3-order-pipeline.md) | написать проверки качества до рискованного изменения, затем переписать разбор и сборку заказов | -| 6 | Разбери сырые события и сверь результат со стендом | изучить разбор потока отдельно от живого приема и сверить его со стендовым результатом | +| 6 | [Разбери сырые события и сверь результат со стендом](6-1-event-parsing.md) | изучить разбор потока отдельно от живого приема и сверить его со стендовым результатом | | 7 | Мониторинг: куда смотреть | выбрать, куда и на что смотреть в живой системе | -ТЗ для ступеней 6-7 появятся по мере продвижения программы. +ТЗ для ступени 7 появится по мере продвижения программы. ## Ступень 1: пять витрин слоя DM