ddmitry/clickstream-ch-kafka-superset-demo
1
0
Fork 0
Code Issues 1 Pull Requests Actions Packages Projects Releases Wiki Activity

docs(architecture): update ODS error handling and DDS partial data support

Browse Source 
Refine data flow diagrams and documentation to clarify error handling
in the ODS layer and partial data processing in the DDS layer. Add
detailed explanations for materialized views, batch SQL transformations,
and data quality metrics. Split DDS entity assembly diagrams for better
readability of event and click processing pipelines.
This commit is contained in:
ddadmin
2026-02-06 22:21:16 +03:00
parent e44b988d76
commit cbf5f22064
1 changed files with 187 additions and 81 deletions
Download Patch File Download Diff File Expand all files Collapse all files
docs/ARCHITECTURE.md
+187 -81
View File
@@ -24,91 +24,91 @@
### Общая схема потока данных ### Общая схема потока данных
```mermaid ```mermaid
flowchart TB flowchart LR
subgraph Sources["📁 Источники (JSONL)"] subgraph Sources["📁 Источники (JSONL)"]
BE[browser_events.jsonl] BE[browser_events]
LE[location_events.jsonl] LE[location_events]
DE[device_events.jsonl] DE[device_events]
GE[geo_events.jsonl] GE[geo_events]
end end
subgraph Kafka["🚀 Kafka Topics"] subgraph Kafka["🚀 Kafka"]
KT1[browser_events] K1[browser_events]
KT2[location_events] K2[location_events]
KT3[device_events] K3[device_events]
KT4[geo_events] K4[geo_events]
end end
subgraph STG["📦 STG (Staging)"] subgraph STG["📦 STG"]
BR[browser_raw] S1[browser_raw]
LR[location_raw] S2[location_raw]
DR[device_raw] S3[device_raw]
GR[geo_raw] S4[geo_raw]
MV1[mv_*_to_ods]
MV2[mv_*_to_errors]
end end
subgraph ODS["🔧 ODS (Operational Data Store)"] subgraph ODS["🔧 ODS"]
BE_O[browser_event] O1[browser_event]
LE_O[location_event] O2[location_event]
DE_O[device_by_click] O3[device_by_click]
GE_O[geo_by_click] O4[geo_by_click]
ERR[error_tables] OE[error_tables]
end end
subgraph DDS["🎯 DDS (Detailed Data Store)"] subgraph DDS["🎯 DDS"]
E[event] DE1[event]
C[click] DC1[click]
end end
subgraph DM["📊 DM (Data Marts)"] subgraph DM["📊 DM"]
VE[v_events_enriched] DM1[v_events_enriched]
VDT[v_daily_traffic] DM2[v_daily_traffic]
VTP[v_top_pages_daily] DM3[v_utm_effectiveness]
VUTM[v_utm_effectiveness] DM4[v_top_pages]
VSE[v_session_overview]
VDQ[v_dq_errors_daily]
end end
BE --> KT1 --> BR --> BE_O --> E --> VE BE --> K1 --> S1 --> MV1 --> O1 --> DE1 --> DM1
LE --> KT2 --> LR --> LE_O --> E LE --> K2 --> S2 --> MV1 --> O2 --> DE1
DE --> KT3 --> DR --> DE_O --> C --> VE DE --> K3 --> S3 --> MV1 --> O3 --> DC1 --> DM1
GE --> KT4 --> GR --> GE_O --> C GE --> K4 --> S4 --> MV1 --> O4 --> DC1
BE_O -.->|ошибки| ERR S1 & S2 & S3 & S4 --> MV2 -.-> OE
E --> VDT & VTP & VUTM & VSE & VDQ DE1 --> DM2 & DM3 & DM4
C --> VDT & VTP & VUTM & VSE & VDQ DC1 --> DM2 & DM3 & DM4
``` ```
### Слои и их назначение ### Слои и их назначение
```mermaid ```mermaid
flowchart LR flowchart TB
subgraph L0["📝 Сырые данные"] subgraph L0["📝 Источники"]
RAW[JSON файлы<br/>1000 строк каждый] RAW["JSON файлы (1000 строк)"]
end end
subgraph L1["STG - Staging"] subgraph L1["📦 STG - Staging"]
STG_T["Таблицы *_raw<br/>MergeTree"] direction LR
KAFKA["Kafka Engine + MV"] KAFKA["Kafka Engine"]
STG_T["*_raw таблицы<br/>(MergeTree)"]
end end
subgraph L2["ODS - Операционный слой"] subgraph L2["🔧 ODS - Операционный слой"]
ODS_T["Типизированные таблицы<br/>ReplacingMergeTree"] direction LR
ODS_T["Типизированные таблицы<br/>(ReplacingMergeTree)"]
DQ["parse_errors<br/>DQ-метрики"] DQ["parse_errors<br/>DQ-метрики"]
end end
subgraph L3["DDS - Детальный слой"] subgraph L3["🎯 DDS - Детальный слой"]
DDS_T["Сущности event + click<br/>Batch SQL"] DDS_T["event + click<br/>(Batch SQL)"]
end end
subgraph L4["DM - Витрины"] subgraph L4["📊 DM - Витрины"]
DM_T["VIEW для BI<br/>Superset/Grafana"] DM_T["VIEW для BI<br/>(Superset/Grafana)"]
end end
RAW -->|kafka-console-producer| KAFKA -->|MV| STG_T RAW -->|kafka-console-producer| KAFKA -->|MV| STG_T -->|MV| ODS_T
STG_T -->|MV| ODS_T ODS_T -->|argMax + JOIN| DDS_T -->|VIEW| DM_T
ODS_T -->|argMax + JOIN| DDS_T ODS_T -.->|ошибки| DQ
DDS_T -->|VIEW| DM_T
ODS_T -.->|ошибки парсинга| DQ
``` ```
--- ---
@@ -159,6 +159,19 @@ CREATE TABLE stg.browser_raw (
| `geo_by_click` | click_id | ReplacingMergeTree(src_ingest_ts) | Гео-данные | | `geo_by_click` | click_id | ReplacingMergeTree(src_ingest_ts) | Гео-данные |
| `*_errors` | — | MergeTree | Строки с битыми ключами | | `*_errors` | — | MergeTree | Строки с битыми ключами |
**Materialized Views для обработки ошибок:**
| MV | Назначение |
|----|-----------|
| `mv_browser_raw_to_ods_errors` | Переносит строки с ошибками в `browser_event_errors` |
| `mv_location_raw_to_ods_errors` | Переносит строки с ошибками в `location_event_errors` |
| `mv_device_raw_to_ods_errors` | Переносит строки с ошибками в `device_by_click_errors` |
| `mv_geo_raw_to_ods_errors` | Переносит строки с ошибками в `geo_by_click_errors` |
**Логика разделения:**
- **Основная таблица**: строки с валидными ключами (`WHERE key IS NOT NULL`)
- **Таблица ошибок**: строки с невалидными ключами (`WHERE key IS NULL`)
**Пример структуры:** **Пример структуры:**
```sql ```sql
CREATE TABLE ods.browser_event ( CREATE TABLE ods.browser_event (
@@ -231,22 +244,57 @@ CREATE TABLE dds.click (
``` ```
**Загрузка (Batch SQL):** **Загрузка (Batch SQL):**
Загрузка `dds.click` с поддержкой partial data (когда device и geo приходят независимо):
```sql ```sql
-- Снапшот ODS через argMax -- UNION всех click_id из device и geo
INSERT INTO dds.click INSERT INTO dds.click
SELECT d.click_id, d.user_domain_id, ..., g.geo_country, ... SELECT
c.click_id,
d.user_domain_id,
d.device_type,
g.geo_country,
g.geo_latitude,
-- ... остальные поля
now64(3) AS dds_update_ts,
arrayFilter(x -> x != '', arrayConcat(
ifNull(d.parse_errors, []),
if(d.click_id IS NULL, ['device_not_found'], []),
if(g.click_id IS NULL, ['geo_not_found'], [])
)) AS ods_parse_errors
FROM ( FROM (
SELECT click_id, argMax(user_domain_id, src_ingest_ts) AS user_domain_id, ... -- Union всех click_id для обработки geo-only и device-only
FROM ods.device_by_click SELECT click_id FROM (
GROUP BY click_id SELECT assumeNotNull(click_id) AS click_id
) d FROM ods.device_by_click WHERE click_id IS NOT NULL
GROUP BY click_id
)
UNION DISTINCT
SELECT click_id FROM (
SELECT assumeNotNull(click_id) AS click_id
FROM ods.geo_by_click WHERE click_id IS NOT NULL
GROUP BY click_id
)
) AS c
LEFT JOIN ( LEFT JOIN (
SELECT click_id, argMax(geo_country, src_ingest_ts) AS geo_country, ... -- Снапшот device
FROM ods.geo_by_click SELECT assumeNotNull(click_id) AS click_id, ...
GROUP BY click_id FROM ods.device_by_click GROUP BY click_id
) g ON g.click_id = d.click_id; ) AS d ON d.click_id = c.click_id
LEFT JOIN (
-- Снапшот geo
SELECT assumeNotNull(click_id) AS click_id, ...
FROM ods.geo_by_click GROUP BY click_id
) AS g ON g.click_id = c.click_id;
``` ```
**Ключевые особенности:**
- **UNION click_id**: собираем все уникальные click_id из обоих источников
- **LEFT JOIN**: обрабатываем случаи когда есть только device или только geo
- **`assumeNotNull`**: типобезопасное преобразование после фильтрации NULL
- **DQ-метрики**: маркируем отсутствующие данные (`device_not_found`, `geo_not_found`)
**Почему batch, а не MV:** **Почему batch, а не MV:**
- **Согласованность**: MV с JOIN даёт eventual consistency (данные приходят в разное время) - **Согласованность**: MV с JOIN даёт eventual consistency (данные приходят в разное время)
- **Контроль**: Batch SQL можно проверить, откатить, перезапустить - **Контроль**: Batch SQL можно проверить, откатить, перезапустить
@@ -280,6 +328,19 @@ FROM dds.event AS e
LEFT JOIN dds.click AS c ON c.click_id = e.click_id; LEFT JOIN dds.click AS c ON c.click_id = e.click_id;
``` ```
**Материализованная таблица DQ:**
```sql
-- Таблица для мониторинга качества (пересоздаётся при каждом batch)
TRUNCATE TABLE dm.dq_summary;
INSERT INTO dm.dq_summary
SELECT today() AS check_date, 'stg' AS layer, ...
FROM ...
```
- `TRUNCATE` предотвращает накопление дубликатов при повторных запусках
- Хранит статистику по всем слоям (stg/ods/dds) для быстрой проверки
**Почему VIEW:** **Почему VIEW:**
- Для демо: достаточно производительности - Для демо: достаточно производительности
- Гибкость: изменения логики не требуют пересоздания таблиц - Гибкость: изменения логики не требуют пересоздания таблиц
@@ -392,30 +453,45 @@ erDiagram
### Сборка DDS-сущностей ### Сборка DDS-сущностей
**event** (browser + location):
```mermaid ```mermaid
flowchart LR flowchart LR
subgraph ODS_IN["ODS (вход)"] subgraph ODS["ODS"]
B[browser_event<br/>event_id + click_id] B["browser_event"]
L[location_event<br/>event_id] L["location_event"]
D[device_by_click<br/>click_id] end
G[geo_by_click<br/>click_id]
subgraph DDS["DDS"]
EV["event"]
end
B -->|JOIN по event_id| EV
L -->|JOIN по event_id| EV
```
**click** (device + geo) с поддержкой partial data:
```mermaid
flowchart LR
subgraph ODS["ODS"]
D["device_by_click"]
G["geo_by_click"]
end end
subgraph BUILD["Batch SQL"] subgraph BUILD["Batch SQL"]
J1["JOIN по event_id"] U["UNION DISTINCT<br/>click_id"]
J2["JOIN по click_id"] J["LEFT JOIN"]
end end
subgraph DDS_OUT["DDS (результат)"] subgraph DDS["DDS"]
EV[event<br/>всё про событие] CL["click"]
CL[click<br/>всё про сессию]
end end
B --> J1 D -->|все click_id| U
L --> J1 --> EV G -->|все click_id| U
B -->|click_id| J2 U --> J
D --> J2 --> CL D -->|данные| J
G --> J2 G -->|данные| J
J --> CL
``` ```
**Важно:** Не все `click_id` из events есть в device/geo. Используем `LEFT JOIN`. **Важно:** Не все `click_id` из events есть в device/geo. Используем `LEFT JOIN`.
@@ -446,6 +522,36 @@ flowchart LR
| **MV + JOIN** | Реалтайм | Eventual consistency, дубли при late arrival | | **MV + JOIN** | Реалтайм | Eventual consistency, дубли при late arrival |
| **Batch (выбрано)** | Согласованность, контроль | Задержка до следующего запуска | | **Batch (выбрано)** | Согласованность, контроль | Задержка до следующего запуска |
### Обработка ошибок в ODS
**Проблема:** Грязные данные с невалидными ключами (NULL event_id/click_id).
**Решение:**
1. **Основная таблица**: только валидные строки (`WHERE key IS NOT NULL`)
2. **Таблица ошибок**: строки с невалидными ключами через отдельные MV
3. **DQ-метрики**: массив `parse_errors` для аудита
```sql
-- Основная таблица
CREATE MV mv_browser_raw_to_ods_browser_event
TO ods.browser_event
SELECT ... FROM stg.browser_raw WHERE event_id IS NOT NULL;
-- Таблица ошибок
CREATE MV mv_browser_raw_to_ods_errors
TO ods.browser_event_errors
SELECT ... FROM stg.browser_raw WHERE event_id IS NULL;
```
### Partial data в DDS
**Проблема:** Device и geo события приходят независимо (не все click_id есть в обоих источниках).
**Решение:**
1. **UNION DISTINCT** всех click_id из обоих источников
2. **LEFT JOIN** для получения данных (обрабатываем device-only и geo-only)
3. **DQ-маркеры**: `device_not_found`, `geo_not_found` в `parse_errors`
--- ---
## Масштабирование ## Масштабирование