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

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

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