clickstream-data-platform/docs/labs/6-1-event-parsing.md

Разберите сырые события рядом со стендом

Чему учит: пакетно разобрать сообщения отдельно от живого приема и построчно проверить свой результат по нетронутому ODS.

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

• формат события;
• справочник хранилища, разделы про имена объектов, актуальные события, прием потока и срок жизни сырых данных;
• ADR о приеме событий, раздел с принятым решением;
• схему ods.event_rep.

Готовый стендовый разбор находится в 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. Сначала найдите среди недавних данных завершившийся интервал со строками. Посмотрите счет по минутам:

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). Достаточно одной локальной таблицы: загрузка и сверка в этом опыте выполняются на одной ноде.

WatchID — логический идентификатор события, уникальный по договору генератора. Физический ключ 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. Ожидаемый счет равен числу различных полных ключей среди годных доставок. Разница между числом доставок и этим счетом показывает повторы в исходном срезе.

Выберите события ODS для сравнения

Теперь нужно найти в ODS события из выбранного среза. Список стройте по исходным доставкам: если взять его из своей таблицы, ошибочно отвергнутое событие исчезнет с обеих сторон сверки и потеря останется незамеченной.

Снова возьмите весь исходный STG-срез. По координатам Kafka исключите доставки, которые есть в ods.event_errors_dist. У оставшихся извлеките WatchID без риска прервать запрос и выберите эти события из свежего 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

ods.event_v — образец для сравнения. Оба разбора могли повторить одну ошибку. Для пяти опорных полей у вас есть проверка преобразования в нужный тип и исходный JSON. Остальные 42 поля можно вручную сопоставить с raw, однако второго независимого источника для всех значений на стенде нет.

Уберите свой объект

Сохраните DDL таблицы, запрос разбора, границы среза, число исходных, годных и отвергнутых строк и итоговую сверку без расхождений. Запишите также, чего не подтверждает сравнение с ODS и как TTL ограничивает время для повторения опыта. Затем удалите свою таблицу командой DROP TABLE с ее точным именем. Убедитесь через system.tables, что объекта больше нет. Прикладные строки стенда вы не меняли, поэтому удалять данные из STG, ODS или Kafka после опыта не нужно.

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

• у вас сохранены границы выбранного среза, число строк и координаты доставки по партициям Kafka;
• ваша таблица содержит все 47 полей и _load_ts и обрабатывает повторы по тем же правилам, что и ODS;
• ваш разбор принимает только JSON-объекты с точным набором ключей и пятью опорными полями, которые преобразуются в нужные типы;
• вы сверили число исходных строк с суммой годных и отвергнутых, а найденный брак разобрали по raw, координатам и классу ошибки стенда;
• вы выбрали события ODS по исходному срезу и таблице ошибок, независимо от своей таблицы и условия отбора;
• вы сравнили обе стороны по полному физическому ключу через FULL OUTER JOIN: во всех 47 полях нет расхождений, пропавших и лишних строк тоже нет;
• вы отдельно проверили перенос _load_ts и учли повторные доставки;
• вы записали ограничения сравнения с ODS и срок повторения опыта;
• после опыта вы удалили свою таблицу, сохранив объекты и данные стенда.