From 7918eeed254dbf4c79beedd85a805ea1266db603 Mon Sep 17 00:00:00 2001 From: Dmitry Dementev Date: Sun, 6 Sep 2026 17:33:00 -0400 Subject: [PATCH 1/2] =?UTF-8?q?docs(labs):=20=D0=B4=D0=BE=D0=B1=D0=B0?= =?UTF-8?q?=D0=B2=D0=BB=D0=B5=D0=BD=D0=BE=20=D0=B7=D0=B0=D0=B4=D0=B0=D0=BD?= =?UTF-8?q?=D0=B8=D0=B5=20=D0=BF=D0=BE=20=D1=80=D0=B0=D0=B7=D0=B1=D0=BE?= =?UTF-8?q?=D1=80=D1=83=20=D1=81=D0=BE=D0=B1=D1=8B=D1=82=D0=B8=D0=B9?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Зачем: - ступени 6 не хватало ТЗ для пакетного разбора сырого среза рядом с живым приемом. - Что: - добавлено задание с одной собственной таблицей, строгим входом и сверкой всех 47 полей с ODS. - индекс основной программы дополнен ссылкой на ступень 6 и актуальным статусом ступени 7. - Проверка: - выполнен `git diff --check`, проверены локальные ссылки и договор задания по DDL STG и ODS. --- docs/labs/6-1-event-parsing.md | 215 +++++++++++++++++++++++++++++++++ docs/labs/README.md | 4 +- 2 files changed, 217 insertions(+), 2 deletions(-) create mode 100644 docs/labs/6-1-event-parsing.md diff --git a/docs/labs/6-1-event-parsing.md b/docs/labs/6-1-event-parsing.md new file mode 100644 index 0000000..cd0ab8b --- /dev/null +++ b/docs/labs/6-1-event-parsing.md @@ -0,0 +1,215 @@ +# Разберите сырые события рядом со стендом + +**Чему учит:** пакетно разобрать сообщения отдельно от живого приема и +построчно проверить свой результат по нетронутому 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`. Если недавних строк нет, запустите +`world_live_day`: он распределит события модельного дня по реальным минутам. +Затем повторите поиск. Службы для задания останавливать не нужно. + +Зафиксируйте до разбора: + +- обе границы с миллисекундами и число строк; +- минимальный и максимальный `kafka_offset` по каждой паре + `kafka_topic`, `kafka_partition`. + +Во всех дальнейших запросах читайте `stg.hits_raw_dist` только с этими +границами. Зафиксированная верхняя граница защищает срез от нового потока. +Повторить опыт можно, пока строки не удалил 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` разбираются + в ожидаемые типы. + +Пять опорных полей сначала разбирайте в `Nullable`. К преобразованию в +необнуляемый тип переходите после фильтра, чтобы грязная строка не остановила +весь пакет исключением. Остальные поля извлеките по договору формата. Выберите +сами форму общих выражений и способ не повторять их внутри запроса. Итог +предиката всегда должен быть равен 0 или 1: `NULL` не должен потеряться между +предикатом и его отрицанием. + +Перенесите `_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` в `Nullable` и выберите эти события из свежего `ods.event_v`. + +Здесь `WatchID` служит мостом между сырой доставкой и ODS: по договору +генератора это уникальный идентификатор события. Он задает только область +эталона. Дедупликация и последующее соединение по-прежнему используют полный +физический ключ. Не берите область из своей таблицы и не применяйте к ней свой +предикат. Иначе ошибочно отвергнутая строка исчезнет с обеих сторон сверки. + +## Сверьте 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 дает образец для сравнения. Оба разбора могли повторить одну +ошибку. Для пяти опорных полей у вас есть проверка разбираемости и исходный +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 -- 2.54.0 From 6e4680d12dcef649ad24b5e42df2df3c70852edd Mon Sep 17 00:00:00 2001 From: Dmitry Dementev Date: Mon, 7 Sep 2026 03:19:58 -0400 Subject: [PATCH 2/2] =?UTF-8?q?docs(labs):=20=D1=83=D1=82=D0=BE=D1=87?= =?UTF-8?q?=D0=BD=D0=B5=D0=BD=D0=BE=20=D0=A2=D0=97=20=D1=80=D0=B0=D0=B7?= =?UTF-8?q?=D0=B1=D0=BE=D1=80=D0=B0=20=D1=81=D0=BE=D0=B1=D1=8B=D1=82=D0=B8?= =?UTF-8?q?=D0=B9=20=D0=BF=D0=BE=20=D1=80=D0=B5=D0=B2=D1=8C=D1=8E?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Зачем: - сделать постановку ступени 6 понятнее после редакторского ревью, сохранив её учебный договор и границы. - Что: - оставлена свобода выбрать форму разбора без навязывания Nullable и готовой схемы 0/1. - уточнены пакетная обработка, баланс годных строк, модельное и ручное время и убрана лишняя персонификация. - Проверка: - проверены SHA256, `git diff --check` и изменение только `docs/labs/6-1-event-parsing.md`. --- docs/labs/6-1-event-parsing.md | 85 +++++++++++++++++----------------- 1 file changed, 42 insertions(+), 43 deletions(-) diff --git a/docs/labs/6-1-event-parsing.md b/docs/labs/6-1-event-parsing.md index cd0ab8b..b817729 100644 --- a/docs/labs/6-1-event-parsing.md +++ b/docs/labs/6-1-event-parsing.md @@ -49,9 +49,12 @@ ORDER BY minute DESC; Выберите короткий интервал `[slice_from, slice_to)`, который уже закончился и дает удобное для построчной проверки число строк. Если одна минута велика, -сузьте ее точными границами `_load_ts`. Если недавних строк нет, запустите -`world_live_day`: он распределит события модельного дня по реальным минутам. -Затем повторите поиск. Службы для задания останавливать не нужно. +сузьте ее точными границами `_load_ts`. Если недавних строк нет, откройте +Airflow. При включенном `world_live` посмотрите его текущий запуск и дочерний +`world_live_day`: дождитесь свежих строк от текущего или следующего планового +запуска, а при ошибке сначала разберите ее причину. Запустите `world_live_day` +вручную, только если `world_live` стоит на паузе и активного `world_live_day` +нет. Успешный прогон распределяет события модельного дня по реальным минутам. Зафиксируйте до разбора: @@ -60,9 +63,8 @@ ORDER BY minute DESC; `kafka_topic`, `kafka_partition`. Во всех дальнейших запросах читайте `stg.hits_raw_dist` только с этими -границами. Зафиксированная верхняя граница защищает срез от нового потока. -Повторить опыт можно, пока строки не удалил TTL; после этого прежний срез уже -не восстановить из ODS. +границами. Новые строки с более поздним `_load_ts` в срез не попадут. Повторите +опыт до очистки по TTL; после нее прежний срез уже не восстановить из ODS. ## Соберите собственную таблицу @@ -70,47 +72,45 @@ ORDER BY minute DESC; `ods.event_rep`, а также `_load_ts`. Тип `purchaseRevenue` возьмите из ODS: `Array(Decimal(18, 2))`. -Повтор доставки должен схлопываться так же, как в ODS. Возьмите +Схлопните повтор доставки по договору ODS. Возьмите `ReplacingMergeTree(_load_ts)`, партицию `EventDate` и полный ключ замены: `(CounterID, EventDate, intHash32(ClientID), WatchID)`. Достаточно одной локальной таблицы: отдельная Distributed-таблица не добавит урока к пакетному опыту на одной ноде. -`WatchID` - логический идентификатор события, уникальный по +`WatchID` — логический идентификатор события, уникальный по [договору генератора](../specs/2026-08-01-generator.md). Физический ключ ODS -шире: `ReplacingMergeTree` считает версиями строки с одинаковым полным ключом -сортировки. Используйте его при загрузке и сверке. +шире: в `ReplacingMergeTree` версиями считаются строки с одинаковым полным +ключом сортировки. Используйте его при загрузке и сверке. ## Разберите годные строки Для каждой строки среза примите решение по тому же строгому входу, что и стенд: -1. корень `raw` - JSON-объект; +1. корень `raw` — JSON-объект; 2. набор верхнеуровневых ключей в точности равен 47 ключам формата; 3. `WatchID`, `VisitID`, `ClientID`, `EventDate` и `UTCEventTime` разбираются в ожидаемые типы. -Пять опорных полей сначала разбирайте в `Nullable`. К преобразованию в -необнуляемый тип переходите после фильтра, чтобы грязная строка не остановила -весь пакет исключением. Остальные поля извлеките по договору формата. Выберите -сами форму общих выражений и способ не повторять их внутри запроса. Итог -предиката всегда должен быть равен 0 или 1: `NULL` не должен потеряться между -предикатом и его отрицанием. +Напишите разбор, который не прерывает пакет из-за одной грязной строки. Сами +выберите способ безопасного извлечения пяти опорных полей, форму общих выражений +и место фильтра. Остальные поля извлеките по договору формата. Перенесите `_load_ts` из STG без пересчета: это время исходной доставки и колонка версии. Время запуска вашего запроса сюда не входит. Вставьте в свою таблицу только годные строки выбранного среза. -Отдельную таблицу брака строить не нужно. Но решение строгого входа должно -быть наблюдаемым. Одним запросом посчитайте для исходного среза: +Считайте отвергнутые строки прямо в исходном срезе: отдельная таблица брака в +задание не входит. Проверьте работу строгого входа одним запросом. Посчитайте: - все доставки; - строки, прошедшие ваш предикат; - строки, отвергнутые его отрицанием. -Первая величина должна равняться сумме двух следующих. Если отвергнутые строки -есть, сохраните их `raw`, `_load_ts` и координаты Kafka. Найдите те же +Проверьте равенство общего числа строк сумме прошедших и отвергнутых. Так каждая +исходная строка учитывается ровно в одной группе. Если отвергнутые строки есть, +сохраните их `raw`, `_load_ts` и координаты Kafka. Найдите те же координаты в `ods.event_errors_dist` и запишите стендовый `error_class`. Если брака нет, так и зафиксируйте: опыт проверил успешную ветвь и выполнение предиката на этом срезе, но не классификатор ошибок. Подкладывать грязные @@ -127,24 +127,25 @@ ORDER BY minute DESC; счетом показывает повторы в исходном срезе. Область эталона постройте независимо от своего предиката и результата вставки. -Снова возьмите весь исходный STG-срез. По координатам Kafka отделите доставки, -которые стенд записал в `ods.event_errors_dist`. У оставшихся доставок извлеките -`WatchID` в `Nullable` и выберите эти события из свежего `ods.event_v`. +Снова возьмите весь исходный STG-срез. По координатам Kafka исключите доставки, +которые есть в `ods.event_errors_dist`. У оставшихся доставок извлеките +`WatchID` без риска прервать запрос и выберите эти события из свежего +`ods.event_v`. -Здесь `WatchID` служит мостом между сырой доставкой и ODS: по договору -генератора это уникальный идентификатор события. Он задает только область -эталона. Дедупликация и последующее соединение по-прежнему используют полный -физический ключ. Не берите область из своей таблицы и не применяйте к ней свой -предикат. Иначе ошибочно отвергнутая строка исчезнет с обеих сторон сверки. +Свяжите сырую доставку с ODS по `WatchID`: по договору генератора это уникальный +идентификатор события. Он задает только область эталона. Дедупликация и +последующее соединение используют полный физический ключ. Не берите область из +своей таблицы и не применяйте к ней свой предикат. Иначе ошибочно отвергнутая +строка исчезнет с обеих сторон сверки. ## Сверьте 47 полей Соедините свою таблицу и выбранные строки `ods.event_v` через `FULL OUTER -JOIN` по полному ключу замены. Полное соединение должно сохранить строки, +JOIN` по полному ключу замены. Полное соединение сохраняет строки, которых нет на одной из сторон. Разделите результат на три исхода: -- строка есть только в ODS - вы ее потеряли; -- строка есть только у вас - вы приняли лишнее или иначе разобрали часть +- строка есть только в ODS — вы ее потеряли; +- строка есть только у вас — вы приняли лишнее или иначе разобрали часть ключа; - ключ есть с обеих сторон, но различается хотя бы одно из 47 полей. @@ -154,16 +155,14 @@ ClickHouse по умолчанию заполняет отсутствующую Сравните все поля явно, включая массивы, `ecommerce` и точное десятичное `purchaseRevenue`. Удобный вид диагностики выберите сами: важно, чтобы по -результату можно было назвать отличающиеся колонки одной строки. Итоговая -сводка должна показать число строк каждого из трех исходов. В верном разборе -все три числа равны нулю. +результату можно было назвать отличающиеся колонки одной строки. Выведите число +строк каждого из трех исходов. В верном разборе все три числа равны нулю. `_load_ts` не входит в 47 полей источника и не проверяет JSON-разбор. -Проверьте его отдельно: у каждой строки вашей таблицы метка должна быть меткой -одной из доставок того же полного ключа в исходном срезе, а после `FINAL` -должна остаться самая поздняя метка этого ключа внутри среза. С ODS ее не -сравнивайте: там могла победить более поздняя доставка того же события из-за -границ среза. +Проверьте его отдельно. Свяжите каждую строку своей таблицы с меткой одной из +доставок того же полного ключа в исходном срезе. После `FINAL` остается самая +поздняя метка этого ключа внутри среза. С ODS ее не сравнивайте: там может +храниться более поздняя доставка того же события из-за границ среза. ## Расследуйте расхождение @@ -172,8 +171,8 @@ ClickHouse по умолчанию заполняет отсутствующую Для измененной строки назовите поля, которые расходятся, и сравните каждое с сырой строкой. Для пропавшей или лишней строки проверьте пять опорных полей и -решение строгого предиката. Если строку отверг стенд, найдите ее в -`ods.event_errors_dist` по координатам доставки и посмотрите `error_class`. +решение строгого предиката. Если координаты доставки есть в +`ods.event_errors_dist`, посмотрите `error_class`. Исправьте разбор, удалите и заново создайте только собственную таблицу, затем повторите вставку того же среза. Остановитесь, когда счет исходных доставок по @@ -182,7 +181,7 @@ ClickHouse по умолчанию заполняет отсутствующую ## Где заканчивается проверка -Стендовый ODS дает образец для сравнения. Оба разбора могли повторить одну +`ods.event_v` — образец для сравнения. Оба разбора могли повторить одну ошибку. Для пяти опорных полей у вас есть проверка разбираемости и исходный JSON. Остальные 42 поля можно вручную сопоставить с `raw`, однако второго независимого источника для всех значений на стенде нет. -- 2.54.0