Модуль 7. Моделирование аналитики в ClickHouse

Цель модуля: превратить сырой CDC-поток из Модуля 6 в быстрые дашборды: перестроить мышление с нормальных форм на «широкую таблицу под запросы», освоить AggregatingMergeTree и MV-агрегаты с combinator-функциями, писать дашборд-запросы (группировки, окна, сэмплирование, приблизительные метрики с честной погрешностью) и собрать слои данных — зная, когда они не нужны.

Перед началом: Модуль 4 (MergeTree-семейство, MV, словари) и Модуль 6 (работающий CDC-пайплайн — источник данных для лабораторной). Инструменты: docker-compose курса (ch + поток из Модуля 6), clickhouse-client. Трудоёмкость: ~9 ч (теория 3 ч · практика 5 ч · квиз 1 ч).

🔥 Инцидент: отчёт, который ждали четыре минуты

В аналитике OrderFlow есть дашборд «выручка по дням» — им пользуются и операторы, и продукт. Всё, что он умел, — GROUP BY по всей сырой таблице событий заказов: десятки миллионов строк, полный скан, ни одного агрегата. Четыре минуты на клик. Оператор открывал отчёт, успевал сходить за кофе и возвращался, но добавить фильтр по городу или часу всё равно не мог: каждая новая группировка — ещё минуты ожидания.

Первым делом мы дали узлу больше памяти и подняли max_threads — отчёт ускорился до трёх с половиной минут, и на этом всё. Через месяц сырой таблицы стало вдвое больше, и четыре минуты вернулись. Мы оптимизировали чтение сырого потока вместо того, чтобы не читать его на клик вовсе.

Закончилось тем, что операторы перестали пользоваться отчётом и вернулись к ручным выгрузкам из PostgreSQL — тем самым, ради замены которых вся аналитика и строилась. Дашборд, которым не пользуются, ничем не лучше сломанного, только заметить это сложнее: панель-то зелёная.

Предстоит собрать модель данных под запросы, а не под нормальные формы: широкую таблицу, AggregatingMergeTree с состояниями агрегатов, MV, считающие их на вставке, и слои raw → MV-агрегаты → дашборд — чтобы клик отвечал за миллисекунды.

📖 7.1. Денормализация: из 5НФ — в широкую таблицу под запросы

5НФ в PostgreSQL — про запись без аномалий: одна сущность — одна таблица, правда поддерживается транзакциями. OLAP-таблица живёт под другую метрику: себестоимость одного запроса дашборда. JOIN в ClickHouse — hash-join: правая таблица материализуется в память на каждый запрос; пять таблиц поверх миллиарда строк — гигабайты хеш-таблиц и секунды, которых у дашборда нет. Решение: денормализация на записи — атрибуты измерений «вджойниваются» один раз в момент попадания события в CH.

text
OLTP: 5НФ (PostgreSQL)                OLAP: широкая таблица (ClickHouse)
customers, categories,                order_items_events — одна строка =
orders ── order_items ── products     = позиция заказа с city/title/category
         (5 таблиц, джойны на чтение)   уже «вджойненными» (джойн на записи)

sql
-- Приёмник: широкая таблица позиций заказа (питается CDC-потоком из Модуля 6)
CREATE TABLE order_items_events
(
    event_time    DateTime64(3),
    order_id      UInt64,
    item_id       UInt64,
    customer_id   UInt64,
    customer_city LowCardinality(String),  -- словарные атрибуты — LowCardinality
    product_id    UInt64,
    product_title LowCardinality(String),
    category      LowCardinality(String),
    qty           UInt32,
    price         Decimal(18, 2),
    amount        Decimal(18, 2) MATERIALIZED qty * price,
    _op           LowCardinality(String),  -- c/u/d/r из Debezium-конверта
    version       UInt64                   -- source.lsn
)
ENGINE = ReplacingMergeTree(version)
PARTITION BY toYYYYMM(event_time)
ORDER BY (event_time, order_id, item_id);

Где джойнить атрибуты — по возрастанию связности с CH: в Go-синке (кэш справочника в памяти, обогащение до вставки — наш путь из Модуля 6, полный контроль и юнит-тесты); в MV через словарь (dictGet, Модуль 4.7 — enrichment силами CH); батчем при бэкфилле (разовый INSERT SELECT с джойном).

Честные trade-offs: хранение растёт (атрибуты дублируются), но колоночное сжатие и LowCardinality часто делают широкую таблицу меньше суммы исходников; срез «как было на момент события» — бесплатное свойство (товар переименовали, а старые строки хранят старый title — для аналитики это фича, историческая правда); изменение схемы расходится по таблице — миграции приёмника, Модуль 8.

📖 7.2. AggregatingMergeTree: состояния агрегатов и MV-агрегаты

«Выручка и p99 оплаты за 30 дней» по raw-таблице — скан миллиардов строк на каждый рефреш дашборда. AggregatingMergeTree хранит не строки, а промежуточные состояния агрегатов, которые ассоциативно сливаются при мерже: посчитали по кускам — слили куски, результат равен расчёту по всему объёму. Combinators: -State — при вставке (упаковывает значение в состояние: для uniq — HyperLogLog-скиетч, для квантилей — tdigest), -Merge — при чтении (сливает состояния в результат).

sql
CREATE TABLE orders_minute_summary
(
    minute      DateTime,
    status      LowCardinality(String),
    -- SimpleAggregateFunction: «плоские» агрегаты (sum/min/max/any) —
    -- хранится как обычная колонка, читается БЕЗ -Merge
    revenue     SimpleAggregateFunction(sum, Decimal(18, 2)),
    -- AggregateFunction: состояния (uniq, квантили, argMax) —
    -- читается только через -Merge
    orders_cnt  AggregateFunction(uniq, UInt64),
    pay_ms_p99  AggregateFunction(quantileTDigest(0.99), UInt32)
)
ENGINE = AggregatingMergeTree
ORDER BY (minute, status);

Правило выбора: sum/min/max — берите SimpleAggregateFunction (проще, читается напрямую); uniq, квантили, argMax и прочие состояния — только AggregateFunction + -Merge.

sql
-- Наполнение (в 7.5 — MV/бэкфилл; GROUP BY внутри работает по блоку вставки).
-- «Плоские» агрегаты пишем обычными функциями, состояния — комбинатором -State:
INSERT INTO orders_minute_summary
SELECT toStartOfMinute(event_time) AS minute, status,
       sum(amount)                            AS revenue,
       uniqState(order_id)                    AS orders_cnt,
       quantileTDigestState(0.99)(payment_ms) AS pay_ms_p99
FROM order_items_events
WHERE event_time >= '2026-09-01'
GROUP BY minute, status;

-- Чтение: sum() по SimpleAggregateFunction (она хранит само значение),
-- -Merge + GROUP BY — по состояниям. GROUP BY обязателен даже для одного дня:
-- таблица физически хранит частичные значения по частям данных.
SELECT minute,
       sum(revenue)                           AS revenue,
       uniqMerge(orders_cnt)                  AS orders_cnt,
       quantileTDigestMerge(0.99)(pay_ms_p99) AS pay_p99
FROM orders_minute_summary
WHERE minute >= now() - INTERVAL 30 DAY
GROUP BY minute ORDER BY minute;

⚠️ Смешивать два стиля нельзя, и это самая частая ошибка в DDL витрин: sumState(...) в колонку SimpleAggregateFunction(sum, ...) — ошибка типов (колонка ждёт значение, а не состояние), sumMerge(revenue) по той же колонке — тоже ошибка. Правило простое: что объявили как SimpleAggregateFunction, то пишем sum()/min()/max() и читаем sum()/min()/max(); что объявили как AggregateFunction, то пишем -State и читаем -Merge.

Три ловушки MV-агрегатов:

1. MV не видит историю до своего создания — историю добирают бэкфиллом (INSERT SELECT с -State-функциями), иначе в витрине дыра.
2. MV агрегирует по блоку вставки, финальное слияние — при чтении: не пугайтесь «недосчитанного» агрегата — досчитывает -Merge на чтении.
3. MV поверх ReplacingMergeTree считает физические вставки: дубль версии до мержа попадёт в sumState дважды — деньги завышены, и мерж исходника это уже не исправит. uniqState/minState/maxState/квантильные состояния к дублям устойчивы (значение учитывается один раз), sum — нет. Как это обошли в лаборатории — 7.5.

📖 7.3. Дашборд-запросы: группировки, окна, SAMPLE, приблизительные метрики

GROUP BY ROLLUP (category) даёт промежуточные итоги и общий итог одной строкой (grouping() различает уровни). Оконные функции в ClickHouse считаются после GROUP BY — ранжируем уже посчитанные агрегаты:

sql
-- Топ-10 товаров в каждой категории: окно поверх агрегации
SELECT category, product_title, revenue, rn FROM (
    SELECT category, product_title, sum(amount) AS revenue,
           row_number() OVER (PARTITION BY category ORDER BY sum(amount) DESC) AS rn
    FROM order_items_events
    WHERE event_time >= now() - INTERVAL 7 DAY
    GROUP BY category, product_title
)
WHERE rn <= 10;

-- День к дню: delta выручки
WITH daily AS (
    SELECT toDate(event_time) AS day, sum(amount) AS revenue
    FROM order_items_events
    WHERE event_time >= now() - INTERVAL 14 DAY
    GROUP BY day
)
SELECT day, revenue,
       lagInFrame(revenue) OVER (ORDER BY day ROWS BETWEEN 1 PRECEDING AND CURRENT ROW) AS prev_day,
       round(if(prev_day = 0, NULL, (revenue - prev_day) / prev_day) * 100, 1) AS dod_pct
FROM daily ORDER BY day;

SAMPLE: приблизительно, но детерминированно

Сэмплирование идёт по ключу выборки, который обязан быть частью ключа сортировки — решение принимается при создании таблицы (7.1: проектируйте под сэмплирование заранее):

sql
CREATE TABLE events_sampled
(
    user_id    UInt64,
    event_time DateTime,
    amount     Decimal(18, 2)
)
ENGINE = MergeTree
ORDER BY (user_id, event_time)   -- ключ выборки — в ключе сортировки
SAMPLE BY cityHash64(user_id);

SELECT count(), avg(amount)
FROM events_sampled
SAMPLE 0.01                      -- ~1% строк, детерминированно: те же user_id
WHERE event_time >= today() - 1;

Детерминированность — ключ к честному использованию: сэмплированный график «день к дню» не «прыгает» между рефрешами, сэмпл джойнится с другим сэмплом по тому же ключу. Но это оценка — для точных цифр: полная таблица или слой агрегатов.

Приблизительные count/quantiles: таблица погрешностей

uniqCombined(customer_id) против uniqExact — разница в миллисекундах и гигабайтах на днях данных. Правило одно, и оно про деньги и обещания: всё, что уходит в счёт клиенту, штрафы за SLA или регуляторную отчётность — только exact. Дашборды и тренды — approx: 1% погрешности на тренде «пятница vs четверг» не видна, экономия памяти и времени — на порядки.

📖 7.4. Слои данных: raw → MV-агрегаты → дашборд, и когда слои не нужны

text
[raw: широкие события, ReplacingMergeTree]     ← истина, TTL 3–6 мес.
      │  MV с -State (инкрементально, на вставке)
      ▼
[слой агрегатов: AggregatingMergeTree]         ← минутные/часовые бакеты
      │  SELECT с -Merge (миллисекунды)          + бэкфилл/пересборка из raw
      ▼
[дашборд Grafana / BI]

Свойства, за которые платят: дашборд читает тысячи состояний вместо миллиардов строк (p95 в десятки мс); слой восстановим — это производная raw-слоя, сломали или поменяли агрегат — пересобрали бэкфиллом; TTL разноуровневый — raw живёт месяцы, агрегаты годы, аналитика за прошлый год доступна, когда raw уже в холодном хранилище.

Когда слои не нужны — и это мейнстрим-ответ:

• До ~сотен миллионов строк при хорошем ключе сортировки прямой запрос отвечает за секунды: слой добавит и скорости, и кода, и расхождений.
• Ad-hoc аналитика: слой агрегатов фиксирует измерения — всё, что не вошло в GROUP BY MV, из витрины не достать (город? час? произвольный фильтр? — только raw). Пока срезы неизвестны заранее, слои рано.
• Часто меняющиеся требования: каждая правка витрины = пересборка + бэкфилл.

Правило курса: начинайте без слоёв; добавьте слой, когда p95 дашборда объективно упёрся в скан raw — в метрику, а не «для архитектуры».

📖 7.5. ЛАБОРАТОРИЯ: агрегаты заказа по минутам поверх CDC-потока

Задача: витрина revenue / min / max / p99 платежа по минутам поверх потока заказов из Модуля 6, с бэкфиллом и сверкой с эталоном. Используем расширенную версию orders_ch: добавлены created_at (иммутабельная атрибуция выручки) и payment_ms (длительность оплаты) — колонки едут через коннектор, ALTER TABLE и совместимая миграция — разминка перед Модулем 8.

Шаг 1. Слой 1: бакет по минутам с дедупликацией по ключу

sql
CREATE TABLE orders_minute_agg
(
    minute        DateTime,                -- toStartOfMinute(created_at)
    order_id      UInt64,
    version       SimpleAggregateFunction(max, UInt64),
    status_argmax AggregateFunction(argMax, LowCardinality(String), UInt64),
    amount        SimpleAggregateFunction(max, Decimal(18, 2)),
    payment_ms    SimpleAggregateFunction(max, UInt32)
)
ENGINE = AggregatingMergeTree
PARTITION BY toDate(minute)
ORDER BY (minute, order_id);

-- MV поверх state-таблицы из Модуля 6
CREATE MATERIALIZED VIEW orders_minute_agg_mv TO orders_minute_agg AS
SELECT
    toStartOfMinute(created_at) AS minute,
    order_id,
    max(version)                 AS version,  -- max идемпотентен к дублям вставок
    argMaxState(status, version) AS status_argmax, -- состояние с двумя аргументами
    max(amount)                  AS amount,   -- сумма заказа иммутабельна:
    max(payment_ms)              AS payment_ms -- max(дубль, дубль) = значение
FROM orders_ch
GROUP BY minute, order_id;

Дизайн-решения, спрятанные в DDL: атрибуция по created_at, не по updated_at — иначе статусные обновления двигали бы заказ по минутным бакетам, размножая выручку; max() для иммутабельных значений — max/min/uniq и квантили устойчивы к дублям вставок (ловушка №3 из 7.2), только sum — нет; ключ (minute, order_id) гасит дубли мержем слоя 1.

Шаг 2. Дашборд-запрос по слою 1

Слой 1 хранит частичные значения по частям данных: две версии одного заказа до мержа — это две строки. Поэтому сначала сводим к одной строке на заказ (GROUP BY minute, order_id), и только потом фильтруем по статусу и считаем метрики. Прямой sum(amount) по неслитому слою задвоил бы выручку — ровно ловушка №3 из 7.2:

sql
SELECT
    minute,
    count()                           AS orders,   -- по одной строке на заказ
    sum(amount)                       AS revenue,
    min(amount)                       AS amount_min,
    max(amount)                       AS amount_max,
    quantileTDigest(0.99)(payment_ms) AS pay_ms_p99
FROM (
    SELECT
        minute,
        order_id,
        argMaxMerge(status_argmax) AS status,     -- AggregateFunction → -Merge
        max(amount)                AS amount,     -- SimpleAggregateFunction → max()
        max(payment_ms)            AS payment_ms
    FROM orders_minute_agg
    WHERE minute >= now() - INTERVAL 6 HOUR
    GROUP BY minute, order_id
)
WHERE status = 'paid'      -- фильтр состояния — ПОСЛЕ сведения версий, не в WHERE слоя
GROUP BY minute
ORDER BY minute;

Два момента, на которых спотыкаются: агрегат нельзя класть в WHERE того же уровня (нужен HAVING, а по смыслу — подзапрос, как выше), и по колонкам SimpleAggregateFunction работают обычные max()/sum(), а не -Merge.

Один слой уже даёт все метрики лаборатории — миллисекунды на днях данных. Для p95 < 50 мс на месяцах — слой 2.

Шаг 2b. Тот же запрос из Go

Дашборд-слой читает витрину, а не raw: это контракт API аналитики, а не «SQL-скрипт для Grafana».

go
// boilerplate/internal/analytics/dashboard.go
type MinuteRevenue struct {
	Minute  time.Time       `ch:"minute"`
	Orders  uint64          `ch:"orders"`
	Revenue decimal.Decimal `ch:"revenue"`
	PayP99  float64         `ch:"pay_p99"`
}

const revenueByMinute = `
SELECT minute, count() AS orders, sum(amount) AS revenue,
       quantileTDigest(0.99)(payment_ms) AS pay_p99
FROM (
    SELECT minute, order_id,
           argMaxMerge(status_argmax) AS status,
           max(amount)                AS amount,
           max(payment_ms)            AS payment_ms
    FROM analytics.orders_minute_agg
    WHERE minute >= ?
    GROUP BY minute, order_id
)
WHERE status = 'paid'
GROUP BY minute
ORDER BY minute`

func (r *Repository) RevenueByMinute(ctx context.Context, from time.Time) ([]MinuteRevenue, error) {
	rows, err := r.ch.Query(ctx, revenueByMinute, from)
	if err != nil {
		return nil, fmt.Errorf("analytics: revenue by minute: %w", err)
	}
	defer rows.Close()

	out := make([]MinuteRevenue, 0, 512)
	for rows.Next() {
		var m MinuteRevenue
		if err := rows.ScanStruct(&m); err != nil { // маппинг по тегам `ch:"..."`
			return nil, fmt.Errorf("analytics: scan row: %w", err)
		}
		out = append(out, m)
	}
	if err := rows.Err(); err != nil {
		return nil, err
	}
	return out, nil
}

Практические детали: Decimal(18,2) приезжает в decimal.Decimal (shopspring) — не во float, иначе деньги «поплывут» на суммировании; таймаут запроса задаётся контекстом, а лимиты сервера — в clickhouse.Settings (max_execution_time, max_memory_usage) при открытии соединения, чтобы один «тяжёлый» дашборд не уронил узел; rows.Close() обязателен — соединение возвращается в пул только после него.

Шаг 3. Слой 2: свёртка с -State (по расписанию, не MV)

sql
CREATE TABLE orders_minute_summary
(
    minute     DateTime,
    status     LowCardinality(String),
    revenue    AggregateFunction(sum, Decimal(18, 2)),
    orders_cnt AggregateFunction(uniq, UInt64),
    pay_ms_p99 AggregateFunction(quantileTDigest(0.99), UInt32)
)
ENGINE = AggregatingMergeTree
ORDER BY (minute, status);

-- Обновление по расписанию (cron-джоба), с водяным знаком. MV здесь опасен
-- для revenue: sum не идемпотентен к редким дублям вставок слоя 1 (сбой
-- синкера из Модуля 6), а слой-2 сложил бы их в сумму ещё раз.
INSERT INTO orders_minute_summary
SELECT minute, argMaxMerge(status_argmax) AS status,
       sumState(amount) AS revenue,
       uniqState(order_id) AS orders_cnt,
       quantileTDigestState(0.99)(payment_ms) AS pay_ms_p99
FROM orders_minute_agg
WHERE minute >= (SELECT max(minute) - INTERVAL 1 MINUTE FROM orders_minute_summary)
GROUP BY minute, status;

-- Чтение слоя 2
SELECT minute,
       sumMerge(revenue) AS revenue,
       uniqMerge(orders_cnt) AS orders_cnt,
       quantileTDigestMerge(0.99)(pay_ms_p99) AS pay_ms_p99
FROM orders_minute_summary
WHERE minute >= now() - INTERVAL 6 HOUR
GROUP BY minute ORDER BY minute;

Шаг 4. Бэкфилл и сверка с эталоном

MV не видит прошлое (ловушка №1) — заливаем историю из state-таблицы:

sql
INSERT INTO orders_minute_agg
SELECT toStartOfMinute(created_at) AS minute, order_id,
       max(version) AS version,
       argMaxState(status, version) AS status_argmax,
       argMax(amount, version)      AS amount,   -- argMax гасит дубли версий сам
       argMax(payment_ms, version)  AS payment_ms
FROM orders_ch
GROUP BY minute, order_id;

-- Эталон: точный расчёт по state-таблице
SELECT toStartOfMinute(created_at) AS minute,
       sum(amount)                     AS revenue,
       uniqExact(order_id)             AS orders,
       quantileExact(0.99)(payment_ms) AS pay_ms_p99_exact
FROM orders_ch FINAL
WHERE _deleted = 0 AND status = 'paid'
GROUP BY minute ORDER BY minute;

Сверьте построчно с чтением слоя 2: revenue и orders обязаны совпасть точно; p99 — в пределах погрешности tdigest (десятые доли процента). Расхождение revenue = инцидент пайплайна, и вы знаете, где искать: Модуль 6, раздел про сверку.

💻 Практика

1. Широкая таблица. Разверните order_items_events (7.1), наполните CDC-потоком Модуля 6 с обогащением в синке (или dictGet в MV) — сделайте оба варианта и запишите, где проще жить.
2. Лаборатория 7.5 целиком: слой 1 → дашборд-запрос → слой 2 → бэкфилл → сверка. Добейтесь точного совпадения revenue и p99 в пределах ~1%.
3. Ловушка дублей. Повторите kill-9 из Модуля 6 поверх витрины: sum слоя 1 не должен измениться, а наивная «MV-сумма прямо поверх orders_ch» — меняется. Объяснение в 3 предложения — любимый вопрос код-ревью.
4. Окна и сэмпл. Постройте топ-10 товаров и день-к-дню дельту (7.3); добавьте таблицу с SAMPLE BY и сравните сэмплированный расчёт на 1% с полным — зафиксируйте фактическую погрешность.
5. Порог отказа от слоёв. На ~100 млн строк замерьте p95 дашборд-запроса по raw и по слою 1 — это ваши личные цифры для решения «когда слой нужен».

🧩 Quiz "Проверь себя"

Разбор ключевых ситуаций

• Джойны против широкой таблицы. Единица работы, перенесённая из запроса в запись, экономит умноженные на миллионы запросов ресурсы — это и есть смена парадигмы OLTP → OLAP.
• Исторический срез. Денормализация «замораживает» атрибуты на момент события; для финансовой аналитики это не побочный эффект, а требование.
• -State/-Merge и чтение. «Состояния сливаются ассоциативно» — фундамент; сырая AggregateFunction-колонка без -Merge — не число, а мусор: забыт -Merge, а не сломалась витрина.
• Деньги — только exact. Погрешность approx незаметна в тренде и недопустима в счёте; проведите границу в код-ревью один раз — и она перестанет обсуждаться.
• Измерения витрины. Всё, что не в GROUP BY MV, потеряно для витрины навсегда — цена инкрементальности; фиксируйте срезы витрины письменно до её создания, а слой заводите по метрике p95, не по моде.
• sum поверх дублей. uniq/min/max/квантили устойчивы к дублям вставок, sum — нет: поэтому слой 1 в лаборатории дедуплицирует по ключу, а слой 2 для суммы обновляется по расписанию, а не MV.

✅ Итоги модуля

После прохождения ты должен уметь:

• Обосновать денормализацию (hash-join на каждый запрос против атрибутов, «вджойненных» на записи) и назвать три места обогащения: синк, dictGet в MV, бэкфилл.
• Спроектировать агрегатную витрину: SimpleAggregateFunction для sum/min/max, AggregateFunction(-State/-Merge) для uniq/квантилей/argMax.
• Назвать три ловушки MV-агрегатов: нет истории до создания, слияние только на чтении, sum не идемпотентен к дублям вставок поверх ReplacingMergeTree.
• Писать дашборд-запросы: ROLLUP, окна поверх GROUP BY, детерминированный SAMPLE BY; выбирать exact/approx по правилу «деньги и обещания — только exact».
• Решать, нужны ли слои данных: по p95 дашборда и фиксированности срезов, а не «для архитектуры».
• Собрать витрину «агрегаты заказа по минутам» поверх CDC-потока: слой 1 с дедупом по ключу, слой 2 со -State, бэкфилл и сверка с эталоном.

Шпаргалка модуля

🔗 Мостик к Модулю 8

Витрины построены — теперь их надо годами эксплуатировать. Дальше — эксплуатация двух СУБД одновременно: бэкапы и PITR (pgBackRest/WAL-G для PG, clickhouse-backup для витрин), миграции схем в живом CDC-пайплайне (вы уже видели ALTER TABLE в лаборатории — там будет систематика), мониторинг postgres_exporter + clickhouse-exporter, безопасность и Managed-vs-self-hosted. Данные, которые никто не может восстановить, аналитикой не считаются.