815 lines
34 KiB
Markdown
815 lines
34 KiB
Markdown
# ClickHouse mini DWH (STG → ODS → DDS → DM) для событий из Kafka (JSON per message)
|
||
|
||
Набор данных состоит из 4 потоков (предполагаемые Kafka topics):
|
||
|
||
- `browser_events` — событие (event_id) + время + тип + click_id + браузер
|
||
- `location_events` — детализация страницы/UTM (event_id)
|
||
- `device_events` — атрибуты устройства/пользователя (click_id)
|
||
- `geo_events` — гео/IP (click_id)
|
||
|
||
Ключи стыковки:
|
||
|
||
- `event_id` связывает `browser_events` ↔ `location_events`
|
||
- `click_id` связывает `browser_events` ↔ (`device_events`, `geo_events`)
|
||
|
||
Ниже — DDL “по классике”:
|
||
|
||
- **STG** хранит сырые JSON как есть (1 Kafka message = 1 row) + метаданные доставки.
|
||
- **ODS** типизирует поля, добавляет базовую DQ-диагностику, дедуплицирует “последней версией” (ReplacingMergeTree).
|
||
- **DDS** собирает “детальные сущности” (событие и click-контекст) и поддерживает дообогащение при опоздавших кусках.
|
||
- **DM** — витрины/представления для Superset (enriched view + несколько готовых срезов).
|
||
|
||
> Примечание про “грязные” данные: в ODS мы используем `…OrNull`-парсинг и заполняем `parse_errors`. Некорректные строки не блокируют пайплайн.
|
||
|
||
## Как это согласуется с заданием (docx)
|
||
|
||
В `data/DE-task.md` просят:
|
||
|
||
- развернуть инфраструктуру, загрузить данные в Kafka и “прогнать” их в ClickHouse по слоям;
|
||
- реализовать расчёты/алгоритмы и “регулярный процесс”;
|
||
- подготовить агрегаты под дашборд (дашборд сам по себе “не оценивается”).
|
||
|
||
Эта схема укладывается в задание так:
|
||
|
||
- **Kafka → STG** реализуется внутри ClickHouse через `ENGINE = Kafka` + MV (стриминг, 1 json = 1 row).
|
||
- “**регулярный процесс**” — батч‑трансформации `STG → ODS → DDS (→ DM)` в виде SQL (`INSERT INTO … SELECT …`) по расписанию (Airflow).
|
||
- Airflow используется как основной оркестратор batch‑шагов и DQ‑проверок.
|
||
|
||
### Выбранное решение (MVP)
|
||
|
||
Чтобы сделать “хорошо, но без оверинжиниринга”, фиксируем такое MVP:
|
||
|
||
- **STG → ODS** — batch через SQL (`sql/ods/20_stg_to_ods.sql`) внутри `etl_pipeline`.
|
||
- **DDS** — batch‑сборка из ODS через SQL (`INSERT INTO … SELECT …`) как “регулярный процесс”.
|
||
- Причина: `MV + JOIN` в `ODS → DDS` плохо переносит произвольный порядок прихода данных и может давать некорректные результаты (eventual consistency ODS, версии в разных партициях и т.п.).
|
||
- **DM** — `VIEW` (витрины “на чтении”) поверх DDS, чтобы не плодить лишние таблицы и джобы под демо.
|
||
- На стороне BI считаем, что запросы всегда идут с фильтрами по времени (`event_date`/`event_ts`) и не сканируют всю историю.
|
||
|
||
Диаграмма витрин и потоков данных:
|
||
|
||
```mermaid
|
||
flowchart LR
|
||
subgraph STG
|
||
stg_browser[stg.browser_raw]
|
||
stg_location[stg.location_raw]
|
||
stg_device[stg.device_raw]
|
||
stg_geo[stg.geo_raw]
|
||
end
|
||
|
||
subgraph ODS
|
||
ods_browser[ods.browser_event]
|
||
ods_location[ods.location_event]
|
||
ods_device[ods.device_by_click]
|
||
ods_geo[ods.geo_by_click]
|
||
end
|
||
|
||
subgraph DDS
|
||
dds_event[dds.event]
|
||
dds_click[dds.click]
|
||
end
|
||
|
||
subgraph DM["DM (views)"]
|
||
v_enriched[dm.v_events_enriched]
|
||
v_daily[dm.v_daily_traffic]
|
||
v_pages[dm.v_top_pages_daily]
|
||
v_dq[dm.v_dq_errors_daily]
|
||
end
|
||
|
||
stg_browser -->|Batch SQL| ods_browser
|
||
stg_location -->|Batch SQL| ods_location
|
||
stg_device -->|Batch SQL| ods_device
|
||
stg_geo -->|Batch SQL| ods_geo
|
||
|
||
ods_browser -->|Batch SQL| dds_event
|
||
ods_location -->|Batch SQL| dds_event
|
||
|
||
ods_device -->|Batch SQL| dds_click
|
||
ods_geo -->|Batch SQL| dds_click
|
||
|
||
dds_event -->|VIEW join| v_enriched
|
||
dds_click -->|VIEW join| v_enriched
|
||
|
||
v_enriched --> v_daily
|
||
v_enriched --> v_pages
|
||
v_enriched --> v_dq
|
||
```
|
||
|
||
### Почему DDS батчами (а не MV join)
|
||
|
||
Мы сознательно уходим от `MV + JOIN` в `ODS → DDS`, потому что это решение:
|
||
|
||
- чувствительно к произвольному порядку прихода сообщений между топиками;
|
||
- может давать неконсистентные “снимки” из‑за версионирования в `ReplacingMergeTree` и отсутствия гарантий “последней версии” в момент выполнения MV;
|
||
- может порождать дубли, если одна и та же сущность попадает в разные партиции (например, когда часть полей для партиционирования появляется “позже”).
|
||
|
||
Поэтому DDS считаем батчами: сначала получаем “current snapshot” ODS (например, через `argMax(..., src_ingest_ts)` по ключу), потом делаем join и грузим результат в DDS.
|
||
|
||
Дальше, при росте нагрузки:
|
||
|
||
- делаем инкрементальные батчи по watermark/окнам (а не full rebuild);
|
||
- материализуем самые тяжёлые витрины в `DM` (daily/topN), чтобы BI не джойнил “деталь” на лету;
|
||
- вводим ресурсные лимиты для BI-пользователя ClickHouse (time/memory/rows), чтобы Superset не “утопил” БД.
|
||
|
||
---
|
||
|
||
## План актуализации DDL (target state репозитория)
|
||
|
||
Цель: перестать исполнять DDL из markdown и хранить **исполняемые** DDL в отдельных `sql/*/*.sql` (по слоям), чтобы:
|
||
|
||
- применять их “тонким раннером” через `clickhouse-client` (через `make ddl`);
|
||
- в будущем легко перенести выполнение в Airflow (1 файл = 1 task, линейные зависимости).
|
||
|
||
Важно: Kafka-объекты STG включаем **по умолчанию** (как часть `sql/ddl/stg/10_stg.sql`).
|
||
|
||
### Артефакты DDL (планируемые файлы)
|
||
|
||
- `sql/ddl/00_databases.sql` — базы `stg/ods/dds/dm`.
|
||
- `sql/ddl/stg/10_stg.sql` — STG raw (`stg.*_raw`) + Kafka source tables (`ENGINE = Kafka`) + MV `Kafka → STG`.
|
||
- `sql/ddl/ods/20_ods.sql` — ODS таблицы типизации + DQ (`parse_errors`) + удаление legacy MV `STG → ODS`.
|
||
- `sql/ddl/dds/30_dds.sql` — DDS таблицы (`dds.event`, `dds.click`) **без MV** (только `CREATE TABLE`).
|
||
- `sql/ddl/dm/40_dm.sql` — витрины `VIEW` для Superset (`dm.v_*`).
|
||
|
||
BI-ограничения (ресурсы/пользователь) **не выносим в `sql/*/*.sql`**: оставляем это только как текст/пример в этом плане, чтобы не смешивать инфраструктуру доступа с DDL витрин.
|
||
|
||
### Артефакты batch-трансформаций (планируемые файлы)
|
||
|
||
- `sql/dds/30_ods_to_dds.sql` — регулярная батч‑сборка DDS из ODS:
|
||
- получить “последнюю версию” строк по ключам (`event_id`/`click_id`) через `argMax(..., src_ingest_ts)` (или эквивалент);
|
||
- выполнить join snapshot’ов и загрузить в `dds.event`/`dds.click` (для демо возможно “full rebuild”; позже — инкрементально).
|
||
- `sql/ods/20_stg_to_ods.sql` — регулярная батч‑сборка ODS из STG:
|
||
- очистить ODS (`TRUNCATE`) перед пересборкой;
|
||
- типизировать валидные строки в `ods.*`;
|
||
- сложить критичные ошибки парсинга в `ods.*_errors`.
|
||
|
||
### Исполнение DDL (make сейчас / Airflow потом)
|
||
|
||
Требования к файлам `sql/*/*.sql`:
|
||
|
||
- идемпотентность (`IF NOT EXISTS`), чтобы повторные прогоны были безопасны;
|
||
- строгий порядок исполнения: `00 → 10 → 20 → 30 → 40` (из‑за зависимостей объектов);
|
||
- единые имена топиков Kafka: `browser_events`, `location_events`, `device_events`, `geo_events` (их создаёт `make data`).
|
||
|
||
### Дедупликация и обработка “битых” ключей (ODS)
|
||
|
||
Проблема: данные “грязные”, а ключи стыковки (`event_id`, `click_id`) могут быть `NULL`/невалидными. Если хранить такие строки в основной ODS‑таблице на `ReplacingMergeTree` с `ORDER BY (event_id/click_id)`, то строки с `NULL` ключом могут схлопываться друг с другом на мерджах, и мы потеряем часть ошибок.
|
||
|
||
Решение в target state:
|
||
|
||
- основная ODS (`ods.browser_event`, `ods.location_event`, `ods.device_by_click`, `ods.geo_by_click`) хранит только строки с валидными ключами (key `IS NOT NULL`) и подходит для join’ов/сборки DDS;
|
||
- отдельные таблицы `ods.*_errors` хранят строки с битыми ключами (и/или критичными ошибками парсинга) для DQ‑аналитики и дебага; в них важно сохранять “уникальность строки” через Kafka‑метаданные (`kafka_topic/partition/offset`, `kafka_ts`) + `src_ingest_ts` + `raw`.
|
||
|
||
Про дедуп:
|
||
|
||
- STG хранит все сообщения как есть; при чтении из Kafka уникальность сообщения определяется `(kafka_topic, kafka_partition, kafka_offset)`.
|
||
- В основной ODS дедупликация — по бизнес‑ключу (`event_id` / `click_id`) с версией `src_ingest_ts` (ReplacingMergeTree).
|
||
- Для `ods.*_errors` дедуп/уникальность (если потребуется) делаем по Kafka‑метаданным; но в демо допустимо хранить “как пришло” без схлопывания.
|
||
|
||
Текущее “как запускаем” (целевое, для реализации следующим шагом):
|
||
|
||
- `make ddl` вызывает `scripts/apply_clickhouse_ddl.sh`;
|
||
- скрипт прогоняет `sql/*/*.sql` по порядку через `clickhouse-client --multiquery` внутри контейнера ClickHouse.
|
||
|
||
Batch‑трансформации (целевое, для реализации следующим шагом):
|
||
|
||
- `make transform` (или аналогичная команда) запускает `sql/ods/20_stg_to_ods.sql`, `sql/dds/30_ods_to_dds.sql` и `sql/dm/40_dds_to_dm.sql` через `clickhouse-client`;
|
||
- Airflow может выполнять те же шаги как отдельные task (с ретраями и мониторингом).
|
||
|
||
### Параметры окружения (docker compose)
|
||
|
||
- Kafka для подключений **из контейнера ClickHouse**: `kafka:29092` (а `localhost:9092` — только для клиентов на хосте).
|
||
- ClickHouse порты на хосте: native `localhost:8002`, HTTP `localhost:9123` (см. `docker-compose.yml`).
|
||
- Пользователь ClickHouse: `default`, пароль задан в `configs/default_user.xml` (сейчас `123456`).
|
||
|
||
---
|
||
|
||
## Приложение A: текущий inline DDL (legacy; будет вынесен в `sql/*/*.sql`)
|
||
|
||
### 0) Базы данных
|
||
|
||
```sql
|
||
CREATE DATABASE IF NOT EXISTS stg;
|
||
CREATE DATABASE IF NOT EXISTS ods;
|
||
CREATE DATABASE IF NOT EXISTS dds;
|
||
CREATE DATABASE IF NOT EXISTS dm;
|
||
```
|
||
|
||
---
|
||
|
||
### 1) STG — сырой JSON (+ метаданные доставки)
|
||
|
||
STG-таблицы делаем максимально простыми и “дешевыми”: строка JSON + время приёма + (опционально) Kafka-метаданные.
|
||
|
||
```sql
|
||
CREATE TABLE IF NOT EXISTS stg.browser_raw
|
||
(
|
||
ingest_ts DateTime64(3) DEFAULT now64(3),
|
||
kafka_topic LowCardinality(String) DEFAULT '',
|
||
kafka_partition Int32 DEFAULT -1,
|
||
kafka_offset Int64 DEFAULT -1,
|
||
kafka_ts DateTime64(3) DEFAULT ingest_ts,
|
||
raw String
|
||
)
|
||
ENGINE = MergeTree
|
||
PARTITION BY toYYYYMM(ingest_ts)
|
||
ORDER BY (kafka_topic, kafka_partition, kafka_offset, ingest_ts);
|
||
|
||
CREATE TABLE IF NOT EXISTS stg.location_raw
|
||
(
|
||
ingest_ts DateTime64(3) DEFAULT now64(3),
|
||
kafka_topic LowCardinality(String) DEFAULT '',
|
||
kafka_partition Int32 DEFAULT -1,
|
||
kafka_offset Int64 DEFAULT -1,
|
||
kafka_ts DateTime64(3) DEFAULT ingest_ts,
|
||
raw String
|
||
)
|
||
ENGINE = MergeTree
|
||
PARTITION BY toYYYYMM(ingest_ts)
|
||
ORDER BY (kafka_topic, kafka_partition, kafka_offset, ingest_ts);
|
||
|
||
CREATE TABLE IF NOT EXISTS stg.device_raw
|
||
(
|
||
ingest_ts DateTime64(3) DEFAULT now64(3),
|
||
kafka_topic LowCardinality(String) DEFAULT '',
|
||
kafka_partition Int32 DEFAULT -1,
|
||
kafka_offset Int64 DEFAULT -1,
|
||
kafka_ts DateTime64(3) DEFAULT ingest_ts,
|
||
raw String
|
||
)
|
||
ENGINE = MergeTree
|
||
PARTITION BY toYYYYMM(ingest_ts)
|
||
ORDER BY (kafka_topic, kafka_partition, kafka_offset, ingest_ts);
|
||
|
||
CREATE TABLE IF NOT EXISTS stg.geo_raw
|
||
(
|
||
ingest_ts DateTime64(3) DEFAULT now64(3),
|
||
kafka_topic LowCardinality(String) DEFAULT '',
|
||
kafka_partition Int32 DEFAULT -1,
|
||
kafka_offset Int64 DEFAULT -1,
|
||
kafka_ts DateTime64(3) DEFAULT ingest_ts,
|
||
raw String
|
||
)
|
||
ENGINE = MergeTree
|
||
PARTITION BY toYYYYMM(ingest_ts)
|
||
ORDER BY (kafka_topic, kafka_partition, kafka_offset, ingest_ts);
|
||
```
|
||
|
||
### (Опционально) Kafka-таблицы-источники
|
||
|
||
Если ClickHouse читает Kafka напрямую, обычно создают `ENGINE = Kafka` и Materialized View в STG.
|
||
|
||
Вариант A (предпочтительно): формат `JSONAsString` (если доступен в вашей версии) — читает весь JSON объект в `raw`.
|
||
|
||
```sql
|
||
-- Пример: одна колонка raw, один message = одна строка.
|
||
-- Замените broker/topic/group под вашу инфраструктуру.
|
||
-- Если ClickHouse запущен в docker compose в одной сети с Kafka — обычно это `kafka:29092`.
|
||
-- Если ClickHouse подключается к Kafka с хоста — обычно это `localhost:9092`.
|
||
CREATE TABLE IF NOT EXISTS stg.kafka_browser_raw
|
||
(
|
||
raw String
|
||
)
|
||
ENGINE = Kafka
|
||
SETTINGS
|
||
kafka_broker_list = 'kafka:29092',
|
||
kafka_topic_list = 'browser_events',
|
||
kafka_group_name = 'ch_stg_browser',
|
||
kafka_format = 'JSONAsString',
|
||
kafka_num_consumers = 1,
|
||
kafka_handle_error_mode = 'stream';
|
||
|
||
CREATE MATERIALIZED VIEW IF NOT EXISTS stg.mv_kafka_browser_to_stg
|
||
TO stg.browser_raw
|
||
AS
|
||
SELECT
|
||
now64(3) AS ingest_ts,
|
||
_topic AS kafka_topic,
|
||
_partition AS kafka_partition,
|
||
_offset AS kafka_offset,
|
||
toDateTime64(_timestamp_ms / 1000.0, 3) AS kafka_ts,
|
||
raw
|
||
FROM stg.kafka_browser_raw;
|
||
```
|
||
|
||
По аналогии для остальных топиков:
|
||
|
||
```sql
|
||
CREATE TABLE IF NOT EXISTS stg.kafka_location_raw (raw String)
|
||
ENGINE = Kafka
|
||
SETTINGS
|
||
kafka_broker_list = 'kafka:29092',
|
||
kafka_topic_list = 'location_events',
|
||
kafka_group_name = 'ch_stg_location',
|
||
kafka_format = 'JSONAsString',
|
||
kafka_num_consumers = 1,
|
||
kafka_handle_error_mode = 'stream';
|
||
|
||
CREATE TABLE IF NOT EXISTS stg.kafka_device_raw (raw String)
|
||
ENGINE = Kafka
|
||
SETTINGS
|
||
kafka_broker_list = 'kafka:29092',
|
||
kafka_topic_list = 'device_events',
|
||
kafka_group_name = 'ch_stg_device',
|
||
kafka_format = 'JSONAsString',
|
||
kafka_num_consumers = 1,
|
||
kafka_handle_error_mode = 'stream';
|
||
|
||
CREATE TABLE IF NOT EXISTS stg.kafka_geo_raw (raw String)
|
||
ENGINE = Kafka
|
||
SETTINGS
|
||
kafka_broker_list = 'kafka:29092',
|
||
kafka_topic_list = 'geo_events',
|
||
kafka_group_name = 'ch_stg_geo',
|
||
kafka_format = 'JSONAsString',
|
||
kafka_num_consumers = 1,
|
||
kafka_handle_error_mode = 'stream';
|
||
|
||
CREATE MATERIALIZED VIEW IF NOT EXISTS stg.mv_kafka_location_to_stg
|
||
TO stg.location_raw
|
||
AS
|
||
SELECT
|
||
now64(3) AS ingest_ts,
|
||
_topic AS kafka_topic,
|
||
_partition AS kafka_partition,
|
||
_offset AS kafka_offset,
|
||
toDateTime64(_timestamp_ms / 1000.0, 3) AS kafka_ts,
|
||
raw
|
||
FROM stg.kafka_location_raw;
|
||
|
||
CREATE MATERIALIZED VIEW IF NOT EXISTS stg.mv_kafka_device_to_stg
|
||
TO stg.device_raw
|
||
AS
|
||
SELECT
|
||
now64(3) AS ingest_ts,
|
||
_topic AS kafka_topic,
|
||
_partition AS kafka_partition,
|
||
_offset AS kafka_offset,
|
||
toDateTime64(_timestamp_ms / 1000.0, 3) AS kafka_ts,
|
||
raw
|
||
FROM stg.kafka_device_raw;
|
||
|
||
CREATE MATERIALIZED VIEW IF NOT EXISTS stg.mv_kafka_geo_to_stg
|
||
TO stg.geo_raw
|
||
AS
|
||
SELECT
|
||
now64(3) AS ingest_ts,
|
||
_topic AS kafka_topic,
|
||
_partition AS kafka_partition,
|
||
_offset AS kafka_offset,
|
||
toDateTime64(_timestamp_ms / 1000.0, 3) AS kafka_ts,
|
||
raw
|
||
FROM stg.kafka_geo_raw;
|
||
```
|
||
|
||
> Почему сырой `raw String` в STG полезен для задания: так можно сохранять “грязные” и даже невалидные JSON как есть (в STG), а ошибки парсинга фиксировать уже в ODS (`parse_errors`) без падения потребления Kafka.
|
||
|
||
> Если `JSONAsString` недоступен в вашей версии ClickHouse, чаще всего можно заменить на `kafka_format = 'RawBLOB'` (просто сообщение как строка байтов). Альтернатива — читать `JSONEachRow` сразу в колонки, но тогда вы хуже сохраняете “как пришло” и сложнее разбирать ошибки.
|
||
|
||
### Операционка: как проверить, что Kafka-консьюмеры живы
|
||
|
||
1) Убедиться, что Kafka-таблицы существуют и читаются:
|
||
|
||
```sql
|
||
SHOW TABLES FROM stg LIKE 'kafka_%';
|
||
```
|
||
|
||
2) Посмотреть состояние консьюмеров (названия/колонки зависят от версии ClickHouse, поэтому сначала можно посмотреть схему):
|
||
|
||
```sql
|
||
DESCRIBE TABLE system.kafka_consumers;
|
||
|
||
SELECT *
|
||
FROM system.kafka_consumers
|
||
WHERE database = 'stg'
|
||
ORDER BY table, consumer_id
|
||
LIMIT 50;
|
||
```
|
||
|
||
3) Самый простой smoke-test без системных таблиц — растут ли STG/ODS:
|
||
|
||
```sql
|
||
SELECT count() AS rows, max(kafka_ts) AS max_kafka_ts
|
||
FROM stg.browser_raw;
|
||
|
||
SELECT count() AS rows, max(src_ingest_ts) AS max_ingest
|
||
FROM ods.browser_event;
|
||
```
|
||
|
||
> Практика: для “параллелизма” увеличивайте `kafka_num_consumers` и/или число партиций топика. Для демо обычно достаточно `1`.
|
||
|
||
### Предохранители для BI (Superset), чтобы не “утопить” БД
|
||
|
||
Для тестового достаточно ограничить ресурсы для BI-пользователя. Пример (пароль/сеть/имена подставьте свои):
|
||
|
||
```sql
|
||
CREATE USER IF NOT EXISTS superset IDENTIFIED WITH sha256_password BY 'REPLACE_ME';
|
||
CREATE ROLE IF NOT EXISTS bi_readonly;
|
||
|
||
GRANT SELECT ON dm.* TO bi_readonly;
|
||
GRANT SELECT ON dds.* TO bi_readonly;
|
||
GRANT bi_readonly TO superset;
|
||
|
||
CREATE SETTINGS PROFILE IF NOT EXISTS superset_profile SETTINGS
|
||
max_execution_time = 30,
|
||
max_threads = 4,
|
||
max_memory_usage = 4000000000,
|
||
max_rows_to_read = 200000000,
|
||
max_bytes_to_read = 5000000000,
|
||
max_result_rows = 200000,
|
||
result_overflow_mode = 'break';
|
||
|
||
ALTER USER superset SETTINGS PROFILE superset_profile;
|
||
```
|
||
|
||
---
|
||
|
||
### 2) ODS — типизация + дедупликация + DQ
|
||
|
||
Принцип: на выходе ODS — “как в источнике, но типизировано и пригодно для джойнов”.
|
||
Дедупликация — по бизнес-ключу (`event_id` или `click_id`) с версией `src_ingest_ts`.
|
||
|
||
### 2.1 ODS: browser_events
|
||
|
||
```sql
|
||
CREATE TABLE IF NOT EXISTS ods.browser_event
|
||
(
|
||
event_id Nullable(UUID),
|
||
event_ts Nullable(DateTime64(6)),
|
||
event_date Date MATERIALIZED ifNull(toDate(event_ts), toDate(src_ingest_ts)),
|
||
event_type LowCardinality(Nullable(String)),
|
||
click_id Nullable(UUID),
|
||
browser_name LowCardinality(Nullable(String)),
|
||
browser_user_agent Nullable(String),
|
||
browser_language LowCardinality(Nullable(String)),
|
||
|
||
src_ingest_ts DateTime64(3),
|
||
src_raw String,
|
||
parse_errors Array(LowCardinality(String))
|
||
)
|
||
ENGINE = ReplacingMergeTree(src_ingest_ts)
|
||
PARTITION BY toYYYYMM(event_date)
|
||
ORDER BY (event_id);
|
||
|
||
CREATE MATERIALIZED VIEW IF NOT EXISTS stg.mv_browser_raw_to_ods_browser_event
|
||
TO ods.browser_event
|
||
AS
|
||
WITH
|
||
toUUIDOrNull(JSONExtractString(raw, 'event_id')) AS event_id,
|
||
parseDateTime64BestEffortOrNull(JSONExtractString(raw, 'event_timestamp'), 6) AS event_ts,
|
||
JSONExtractString(raw, 'event_type') AS event_type,
|
||
toUUIDOrNull(JSONExtractString(raw, 'click_id')) AS click_id,
|
||
JSONExtractString(raw, 'browser_name') AS browser_name,
|
||
JSONExtractString(raw, 'browser_user_agent') AS browser_user_agent,
|
||
JSONExtractString(raw, 'browser_language') AS browser_language
|
||
SELECT
|
||
event_id,
|
||
event_ts,
|
||
event_type,
|
||
click_id,
|
||
browser_name,
|
||
browser_user_agent,
|
||
browser_language,
|
||
ingest_ts AS src_ingest_ts,
|
||
raw AS src_raw,
|
||
arrayFilter(x -> x != '', [
|
||
if(event_id IS NULL, 'bad_event_id', ''),
|
||
if(event_ts IS NULL, 'bad_event_timestamp', ''),
|
||
if(click_id IS NULL, 'bad_click_id', '')
|
||
]) AS parse_errors
|
||
FROM stg.browser_raw;
|
||
```
|
||
|
||
### 2.2 ODS: location_events
|
||
|
||
```sql
|
||
CREATE TABLE IF NOT EXISTS ods.location_event
|
||
(
|
||
event_id Nullable(UUID),
|
||
|
||
page_url Nullable(String),
|
||
page_url_path LowCardinality(Nullable(String)),
|
||
referer_url Nullable(String),
|
||
referer_medium LowCardinality(Nullable(String)),
|
||
utm_medium LowCardinality(Nullable(String)),
|
||
utm_source LowCardinality(Nullable(String)),
|
||
utm_content LowCardinality(Nullable(String)),
|
||
utm_campaign LowCardinality(Nullable(String)),
|
||
|
||
src_ingest_ts DateTime64(3),
|
||
src_raw String,
|
||
parse_errors Array(LowCardinality(String))
|
||
)
|
||
ENGINE = ReplacingMergeTree(src_ingest_ts)
|
||
PARTITION BY toYYYYMM(toDate(src_ingest_ts))
|
||
ORDER BY (event_id);
|
||
|
||
CREATE MATERIALIZED VIEW IF NOT EXISTS stg.mv_location_raw_to_ods_location_event
|
||
TO ods.location_event
|
||
AS
|
||
WITH
|
||
toUUIDOrNull(JSONExtractString(raw, 'event_id')) AS event_id
|
||
SELECT
|
||
event_id,
|
||
JSONExtractString(raw, 'page_url') AS page_url,
|
||
JSONExtractString(raw, 'page_url_path') AS page_url_path,
|
||
JSONExtractString(raw, 'referer_url') AS referer_url,
|
||
JSONExtractString(raw, 'referer_medium') AS referer_medium,
|
||
JSONExtractString(raw, 'utm_medium') AS utm_medium,
|
||
JSONExtractString(raw, 'utm_source') AS utm_source,
|
||
JSONExtractString(raw, 'utm_content') AS utm_content,
|
||
JSONExtractString(raw, 'utm_campaign') AS utm_campaign,
|
||
ingest_ts AS src_ingest_ts,
|
||
raw AS src_raw,
|
||
arrayFilter(x -> x != '', [
|
||
if(event_id IS NULL, 'bad_event_id', '')
|
||
]) AS parse_errors
|
||
FROM stg.location_raw;
|
||
```
|
||
|
||
### 2.3 ODS: device_events
|
||
|
||
```sql
|
||
CREATE TABLE IF NOT EXISTS ods.device_by_click
|
||
(
|
||
click_id Nullable(UUID),
|
||
|
||
os Nullable(String),
|
||
os_name LowCardinality(Nullable(String)),
|
||
os_timezone LowCardinality(Nullable(String)),
|
||
device_type LowCardinality(Nullable(String)),
|
||
device_is_mobile Nullable(UInt8),
|
||
user_custom_id Nullable(String),
|
||
user_domain_id Nullable(UUID),
|
||
|
||
src_ingest_ts DateTime64(3),
|
||
src_raw String,
|
||
parse_errors Array(LowCardinality(String))
|
||
)
|
||
ENGINE = ReplacingMergeTree(src_ingest_ts)
|
||
PARTITION BY toYYYYMM(toDate(src_ingest_ts))
|
||
ORDER BY (click_id);
|
||
|
||
CREATE MATERIALIZED VIEW IF NOT EXISTS stg.mv_device_raw_to_ods_device_by_click
|
||
TO ods.device_by_click
|
||
AS
|
||
WITH
|
||
toUUIDOrNull(JSONExtractString(raw, 'click_id')) AS click_id,
|
||
JSONExtract(raw, 'device_is_mobile', 'Nullable(UInt8)') AS device_is_mobile,
|
||
toUUIDOrNull(JSONExtractString(raw, 'user_domain_id')) AS user_domain_id
|
||
SELECT
|
||
click_id,
|
||
JSONExtractString(raw, 'os') AS os,
|
||
JSONExtractString(raw, 'os_name') AS os_name,
|
||
JSONExtractString(raw, 'os_timezone') AS os_timezone,
|
||
JSONExtractString(raw, 'device_type') AS device_type,
|
||
device_is_mobile,
|
||
JSONExtractString(raw, 'user_custom_id') AS user_custom_id,
|
||
user_domain_id,
|
||
ingest_ts AS src_ingest_ts,
|
||
raw AS src_raw,
|
||
arrayFilter(x -> x != '', [
|
||
if(click_id IS NULL, 'bad_click_id', ''),
|
||
if(user_domain_id IS NULL, 'bad_user_domain_id', '')
|
||
]) AS parse_errors
|
||
FROM stg.device_raw;
|
||
```
|
||
|
||
### 2.4 ODS: geo_events
|
||
|
||
```sql
|
||
CREATE TABLE IF NOT EXISTS ods.geo_by_click
|
||
(
|
||
click_id Nullable(UUID),
|
||
|
||
geo_latitude Nullable(Float64),
|
||
geo_longitude Nullable(Float64),
|
||
geo_country LowCardinality(Nullable(String)),
|
||
geo_timezone LowCardinality(Nullable(String)),
|
||
geo_region_name Nullable(String),
|
||
ip_address Nullable(String),
|
||
|
||
src_ingest_ts DateTime64(3),
|
||
src_raw String,
|
||
parse_errors Array(LowCardinality(String))
|
||
)
|
||
ENGINE = ReplacingMergeTree(src_ingest_ts)
|
||
PARTITION BY toYYYYMM(toDate(src_ingest_ts))
|
||
ORDER BY (click_id);
|
||
|
||
CREATE MATERIALIZED VIEW IF NOT EXISTS stg.mv_geo_raw_to_ods_geo_by_click
|
||
TO ods.geo_by_click
|
||
AS
|
||
WITH
|
||
toUUIDOrNull(JSONExtractString(raw, 'click_id')) AS click_id,
|
||
toFloat64OrNull(JSONExtractString(raw, 'geo_latitude')) AS geo_latitude,
|
||
toFloat64OrNull(JSONExtractString(raw, 'geo_longitude')) AS geo_longitude
|
||
SELECT
|
||
click_id,
|
||
geo_latitude,
|
||
geo_longitude,
|
||
JSONExtractString(raw, 'geo_country') AS geo_country,
|
||
JSONExtractString(raw, 'geo_timezone') AS geo_timezone,
|
||
JSONExtractString(raw, 'geo_region_name') AS geo_region_name,
|
||
JSONExtractString(raw, 'ip_address') AS ip_address,
|
||
ingest_ts AS src_ingest_ts,
|
||
raw AS src_raw,
|
||
arrayFilter(x -> x != '', [
|
||
if(click_id IS NULL, 'bad_click_id', ''),
|
||
if(geo_latitude IS NULL, 'bad_geo_latitude', ''),
|
||
if(geo_longitude IS NULL, 'bad_geo_longitude', '')
|
||
]) AS parse_errors
|
||
FROM stg.geo_raw;
|
||
```
|
||
|
||
---
|
||
|
||
### 3) DDS — детальный слой (event + click context)
|
||
|
||
DDS хранит **минимально необходимую детализацию для аналитики**, но уже “собранную”:
|
||
|
||
- `dds.event` — 1 строка на `event_id` (browser + location).
|
||
- `dds.click` — 1 строка на `click_id` (device + geo + user).
|
||
|
||
Важно: MV с `JOIN` для `ODS → DDS` мы в target state **не используем** (см. обоснование выше). DDS собирается батчами из “снапшота” ODS. Поэтому в приложении ниже оставляем только DDL таблиц DDS.
|
||
|
||
### 3.1 DDS: click
|
||
|
||
```sql
|
||
CREATE TABLE IF NOT EXISTS dds.click
|
||
(
|
||
click_id Nullable(UUID),
|
||
|
||
user_domain_id Nullable(UUID),
|
||
user_custom_id Nullable(String),
|
||
|
||
device_type LowCardinality(Nullable(String)),
|
||
device_is_mobile Nullable(UInt8),
|
||
os_name LowCardinality(Nullable(String)),
|
||
os Nullable(String),
|
||
os_timezone LowCardinality(Nullable(String)),
|
||
|
||
geo_country LowCardinality(Nullable(String)),
|
||
geo_region_name Nullable(String),
|
||
geo_timezone LowCardinality(Nullable(String)),
|
||
geo_latitude Nullable(Float64),
|
||
geo_longitude Nullable(Float64),
|
||
ip_address Nullable(String),
|
||
|
||
dds_update_ts DateTime64(3),
|
||
parse_errors Array(LowCardinality(String))
|
||
)
|
||
ENGINE = ReplacingMergeTree(dds_update_ts)
|
||
PARTITION BY toYYYYMM(toDate(dds_update_ts))
|
||
ORDER BY (click_id);
|
||
```
|
||
|
||
### 3.2 DDS: event
|
||
|
||
```sql
|
||
CREATE TABLE IF NOT EXISTS dds.event
|
||
(
|
||
event_id Nullable(UUID),
|
||
event_ts Nullable(DateTime64(6)),
|
||
event_date Date MATERIALIZED ifNull(toDate(event_ts), toDate(dds_update_ts)),
|
||
event_type LowCardinality(Nullable(String)),
|
||
click_id Nullable(UUID),
|
||
|
||
page_url Nullable(String),
|
||
page_url_path LowCardinality(Nullable(String)),
|
||
referer_url Nullable(String),
|
||
referer_medium LowCardinality(Nullable(String)),
|
||
utm_medium LowCardinality(Nullable(String)),
|
||
utm_source LowCardinality(Nullable(String)),
|
||
utm_content LowCardinality(Nullable(String)),
|
||
utm_campaign LowCardinality(Nullable(String)),
|
||
|
||
browser_name LowCardinality(Nullable(String)),
|
||
browser_user_agent Nullable(String),
|
||
browser_language LowCardinality(Nullable(String)),
|
||
|
||
dds_update_ts DateTime64(3),
|
||
parse_errors Array(LowCardinality(String))
|
||
)
|
||
ENGINE = ReplacingMergeTree(dds_update_ts)
|
||
PARTITION BY toYYYYMM(event_date)
|
||
ORDER BY (event_id);
|
||
```
|
||
|
||
---
|
||
|
||
### 4) DM — витрины для Superset
|
||
|
||
### 4.1 Enriched view (удобная “таблица фактов” для аналитики)
|
||
|
||
```sql
|
||
CREATE VIEW IF NOT EXISTS dm.v_events_enriched AS
|
||
SELECT
|
||
e.event_id,
|
||
e.event_ts,
|
||
e.event_date,
|
||
e.event_type,
|
||
e.click_id,
|
||
|
||
e.page_url,
|
||
e.page_url_path,
|
||
e.referer_url,
|
||
e.referer_medium,
|
||
e.utm_medium,
|
||
e.utm_source,
|
||
e.utm_content,
|
||
e.utm_campaign,
|
||
|
||
e.browser_name,
|
||
e.browser_language,
|
||
e.browser_user_agent,
|
||
|
||
c.user_domain_id,
|
||
c.user_custom_id,
|
||
c.device_type,
|
||
c.device_is_mobile,
|
||
c.os_name,
|
||
c.os_timezone,
|
||
|
||
c.geo_country,
|
||
c.geo_region_name,
|
||
c.geo_timezone,
|
||
c.geo_latitude,
|
||
c.geo_longitude,
|
||
c.ip_address,
|
||
|
||
e.dds_update_ts,
|
||
arrayConcat(e.parse_errors, c.parse_errors) AS parse_errors
|
||
FROM dds.event AS e
|
||
LEFT JOIN dds.click AS c
|
||
ON c.click_id = e.click_id;
|
||
```
|
||
|
||
### 4.2 Несколько полезных срезов (views)
|
||
|
||
```sql
|
||
CREATE VIEW IF NOT EXISTS dm.v_daily_traffic AS
|
||
SELECT
|
||
event_date,
|
||
geo_country,
|
||
device_type,
|
||
browser_name,
|
||
utm_source,
|
||
utm_medium,
|
||
count() AS events,
|
||
uniqExact(click_id) AS uniq_clicks,
|
||
uniqExact(user_domain_id) AS uniq_users
|
||
FROM dm.v_events_enriched
|
||
WHERE event_ts IS NOT NULL
|
||
GROUP BY
|
||
event_date,
|
||
geo_country,
|
||
device_type,
|
||
browser_name,
|
||
utm_source,
|
||
utm_medium;
|
||
|
||
CREATE VIEW IF NOT EXISTS dm.v_top_pages_daily AS
|
||
SELECT
|
||
event_date,
|
||
page_url_path,
|
||
count() AS pageviews,
|
||
uniqExact(click_id) AS uniq_clicks
|
||
FROM dm.v_events_enriched
|
||
WHERE event_type = 'pageview'
|
||
GROUP BY event_date, page_url_path;
|
||
|
||
CREATE VIEW IF NOT EXISTS dm.v_dq_errors_daily AS
|
||
SELECT
|
||
event_date,
|
||
arrayJoin(parse_errors) AS error_code,
|
||
count() AS rows_cnt
|
||
FROM dm.v_events_enriched
|
||
WHERE length(parse_errors) > 0
|
||
GROUP BY event_date, error_code;
|
||
```
|
||
|
||
> Если потребуется “настоящая” материализация DM (таблицы с предагрегацией), её лучше делать либо периодическим пересчётом, либо через подход с агрегатными состояниями (`AggregatingMergeTree`). Для мини-демо Superset обычно достаточно `VIEW`.
|
||
|
||
---
|
||
|
||
### 5) Практические заметки для демо
|
||
|
||
- Для быстрой локальной загрузки **первых N строк** (без Kafka) удобно использовать формат `LineAsString`, он кладёт каждую строку файла как `String` в колонку `raw`:
|
||
|
||
```bash
|
||
# 50 строк, как вы просили — не грузим всё
|
||
head -n 50 data/browser_events.jsonl \
|
||
| clickhouse-client --query="INSERT INTO stg.browser_raw (raw) FORMAT LineAsString"
|
||
|
||
head -n 50 data/location_events.jsonl \
|
||
| clickhouse-client --query="INSERT INTO stg.location_raw (raw) FORMAT LineAsString"
|
||
|
||
head -n 50 data/device_events.jsonl \
|
||
| clickhouse-client --query="INSERT INTO stg.device_raw (raw) FORMAT LineAsString"
|
||
|
||
head -n 50 data/geo_events.jsonl \
|
||
| clickhouse-client --query="INSERT INTO stg.geo_raw (raw) FORMAT LineAsString"
|
||
```
|
||
|
||
- `ReplacingMergeTree` “схлопывает” версии во время мерджей. Для строго “последнего состояния” в демо-запросах используйте `FINAL` (дорого) или выполните `OPTIMIZE TABLE … FINAL` после загрузки сэмпла.
|
||
- Для Superset удобнее всего датасеты: `dm.v_events_enriched`, `dm.v_daily_traffic`, `dm.v_top_pages_daily`, `dm.v_dq_errors_daily`.
|