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. Если недавних строк нет, запустите 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 - логический идентификатор события, уникальный по договору генератора. Физический ключ 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 записаны вместе с результатом;
• собственная таблица удалена, объекты и данные стенда не изменены.