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

Разбор сырых событий

В этой работе вы разберете сохраненные сообщения о событиях в свою таблицу. Каждую полученную строку сравните с результатом обработки в стенде.

Подготовка

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

Готовый стендовый разбор находится в sql/ddl/30-ods-views.sql. Откройте его после собственной сверки: в нем лежит ответ на основную часть задания.

Порядок работы

Создайте в базе default свою таблицу для правильно заполненных событий. Выберите небольшой набор сообщений из stg.hits_raw_dist, разберите их и загрузите в таблицу одним INSERT SELECT. Работайте с первым узлом под пользователем etl. Затем сравните результат с событиями в ODS.

Материализованные представления 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 обозначает событие и уникален по правилам генератора. Но ReplacingMergeTree считает версиями строки, у которых совпадает весь ключ сортировки. Поэтому для загрузки и сверки используйте все четыре части ключа ODS, перечисленные выше.

Разбор сообщений

Для каждой строки среза проверьте те же условия приема, что и на стенде:

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

Сверка полей

Соедините свою таблицу и выбранные строки ods.event_v через FULL OUTER JOIN по полному ключу замены. Полное соединение сохраняет строки, которых нет на одной из сторон. Найдите три вида расхождений:

• строка есть только в ODS — вы ее потеряли;
• строка есть только у вас — вы приняли лишнее или иначе разобрали часть ключа;
• ключ есть с обеих сторон, но различается хотя бы одно из 47 полей.

ClickHouse по умолчанию заполняет отсутствующую сторону значениями по умолчанию для соответствующих типов, например нулем или пустой строкой. Включите для сверки join_use_nulls = 1 или добавьте явный признак присутствия в каждую сторону. Не определяйте отсутствие по значениям полей события.

Сравните все поля явно, включая массивы, ecommerce и точное десятичное purchaseRevenue. Выберите, как показать расхождения, чтобы было видно, какие колонки различаются в конкретной строке. Выведите число строк каждого из трех видов расхождений. В верном разборе все три числа равны нулю.

_load_ts не входит в 47 полей источника и не проверяет JSON-разбор. Проверьте его отдельно. Для каждой строки своей таблицы найдите самую позднюю метку доставки того же полного ключа в исходном срезе. Она должна совпасть с _load_ts, который вы читаете после FINAL. С ODS эту метку не сравнивайте: там может храниться более поздняя доставка того же события из-за границ среза.

Поиск ошибок разбора

Если нашлись расхождения, выберите одну строку. Найдите ее ключ в списке, который вы получили из STG, и исходное сообщение raw. Выведите рядом координаты Kafka и _load_ts.

Для измененной строки назовите поля, которые расходятся, и сравните каждое с сырой строкой. Для пропавшей или лишней строки проверьте пять опорных полей и результат условия отбора. Если координаты доставки есть в ods.event_errors_dist, посмотрите error_class.

Исправьте разбор, удалите и заново создайте только собственную таблицу, затем повторите вставку того же среза. Остановитесь, когда число исходных доставок по прежним границам не изменилось, число годных и отвергнутых строк в сумме дает число исходных доставок, а построчная сверка не находит расхождений.

Ограничения проверки

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 и срок повторения опыта;
• после опыта вы удалили свою таблицу, сохранив объекты и данные стенда.