Модуль 11. MVP-проект: «Аналитика OrderFlow»

Цель модуля: собрать всё, что вы построили руками в курсе, в один работающий и защищаемый стенд. Это развитие брокерского MVP: Order Service остаётся на PostgreSQL (источник правды), аналитика переезжает в ClickHouse, между ними — честный CDC-пайплайн. Шесть спринтов, у каждого — проверяемый checkpoint: не «сделал», а «проверил аварией».

Перед началом: Модули 1–9 (Модуль 10 RAG — не требуется). Инструменты: docker-compose, Go 1.22+, pgx, clickhouse-go, Debezium, Patroni, PgBouncer, Grafana, pgbench, k6. Трудоёмкость: ~16–20 ч (6 спринтов по 2.5–3 ч + защита).

🔥 Инцидент: генеральная репетиция защиты

Стенд OrderFlow собран: PostgreSQL под Patroni, Debezium стримит заказы в Kafka, синк наполняет ClickHouse, Grafana рисует выручку. Шесть спринтов формально закрыты, дашборд отвечает за 180 мс, до защиты — один вечер. Решили прогнать сбой-сценарии — не для галочки, а чтобы понять, что стенд переживёт на защите. Репетиция показала, что закрыто было не всё.

Первый прогон — kill -9 мастера PG. Patroni поднял нового мастера за 12 секунд, приложение пережило переключение, но CDC-слот на новом мастере оказался пуст: события, закоммиченные в те секунды, до Kafka не доехали — в отчёте за день выручка просела на 380 000 ₽. Второй прогон — перезапуск синкера посреди батча: та же выручка посчиталась дважды, за среду вместо 2 100 000 ₽ в отчёте появилось 4 200 000 ₽. При этом дашборд «работал»: ошибок нет, алертов нет, панели бодро рисовали неверные числа.

Этот модуль — шесть спринтов MVP и главное правило: checkpoint — это эксперимент над отказом, а не галочка в чек-листе. Здесь стенд собирают по частям — и каждую проверяют аварией: потеря события на failover, дубль при ретрае, отставание пайплайна под нагрузкой — до того, как это сделает комиссия.

📋 Карта проекта: шесть спринтов

Спринт	Что делаем	Проверка (checkpoint)
S1	PostgreSQL: схема заказов/платежей, pg_stat_statements, pg_partman по events	EXPLAIN показывает партиционный prune; бэкап/restore pgBackRest отработали
S2	HA: Patroni ×3 + PgBouncer + pg_failover_slots	kill -9 мастера → приложение не упало, CDC-слот жив
S3	CDC: Debezium → Kafka (топик orderflow.public.orders)	сообщение в Kafka ≤ 2 сек после коммита
S4	ClickHouse 2×2 (ReplicatedMergeTree + Distributed) + Go-синк с батчами и идемпотентностью	kill синкера → дублей в отчётах нет
S5	Аналитика: MV-агрегаты + Grafana-дашборд (revenue, p99 платежей, топ товаров)	дашборд из CH отвечает < 300ms
S6	Нагрузка: pgbench + k6 → отчёт p99/throughput; намеренный инцидент + разбор	lag < 30 сек при пиковой вставке

Правило проекта: checkpoint — это эксперимент над отказом, а не галочка. Каждый спринт заканчивается скриншотом Grafana или выводом команды в README.

💻 Спринты в деталях

S1. PostgreSQL-фундамент: схема, статистика, партиции

Что строим. Схема заказов и платежей — нормализованная, под OLTP; поток событий — партиционированный по времени. Миграции — goose (Модуль 8.2).

sql
CREATE TABLE orders (
    order_id    bigint GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
    tenant_id   bigint NOT NULL,
    status      smallint NOT NULL DEFAULT 0,
    total       numeric(18,2) NOT NULL DEFAULT 0,
    created_at  timestamptz NOT NULL DEFAULT now(),
    updated_at  timestamptz NOT NULL DEFAULT now()
);
CREATE INDEX idx_orders_tenant_created ON orders (tenant_id, created_at);

CREATE TABLE payments (
    payment_id  bigint GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
    order_id    bigint NOT NULL REFERENCES orders,
    amount      numeric(18,2) NOT NULL,
    pay_latency_ms int NOT NULL,
    paid_at     timestamptz NOT NULL DEFAULT now()
);

-- поток событий: range-партиции по дню (pg_partman)
CREATE TABLE order_events (
    event_id    bigint GENERATED ALWAYS AS IDENTITY,
    tenant_id   bigint NOT NULL,
    order_id    bigint NOT NULL,
    kind        text NOT NULL,
    payload     jsonb NOT NULL DEFAULT '{}',
    created_at  timestamptz NOT NULL DEFAULT now(),
    PRIMARY KEY (event_id, created_at)
) PARTITION BY RANGE (created_at);

SELECT partman.create_parent(
    p_parent_table := 'public.order_events',
    p_control      := 'created_at',
    p_interval     := '1 day'
);
UPDATE partman.part_config
SET retention = '90 days'
WHERE parent_table = 'public.order_events';

pg_stat_statements включаем первым делом (Модуль 3) — это ваши глаза на профиль на протяжении всего проекта.

Checkpoint S1:

sql
EXPLAIN ANALYZE
SELECT count(*), avg((payload->>'latency')::int)
FROM order_events
WHERE created_at >= now() - interval '3 days';
-- → Subplans Removed: N — prune отсёк ненужные партиции

bash
# бэкап-рутина из Модуля 8.1: полный бэкап + тест восстановления
pgbackrest --stanza=main --type=full backup
pgbackrest --stanza=main --delta --type=time \
  --target='2026-09-15 12:00:00+03' --target-action=promote restore

Проверка зачтена, когда prune виден в EXPLAIN и restore доведён до поднятого кластера с проверкой строк.

S2. HA: Patroni ×3 + PgBouncer + pg_failover_slots

Что строим. Три узла PG под Patroni (DCS — etcd ×3), перед ними PgBouncer в transaction mode, на репликах — pg_failover_slots: без него CDC-слот умирает вместе с мастером (Модуль 3.2, 6.5).

yaml
# patroni.yml (фрагмент, одинаков на трёх узлах, кроме имени и адреса)
bootstrap:
  dcs:
    postgresql:
      parameters:
        wal_level: logical          # CDC
        max_wal_senders: 10
        max_replication_slots: 10
        hot_standby_feedback: on    # дашборды на реплике (Модуль 9.1)
postgresql:
  parameters:
    shared_preload_libraries:
      - pg_failover_slots           # синхронизация logical-слотов на standby

ini
;; pgbouncer.ini (фрагмент)
[databases]
orderflow = host=patroni-lb port=5432 dbname=orderflow

[pgbouncer]
pool_mode = transaction
max_client_conn = 5000
default_pool_size = 20

Приложение знает один адрес (patroni-lb — VIP/балансер записи) и переживает переключения на ретраях: pgx с настроенным retry на transient-ошибки и короткий timeout — вся «магия» отказоустойчивости приложения.

Checkpoint S2:

bash
docker kill -9 patroni-node1        # умирает текущий мастер
# приложение: 1–2 ретрая — и продолжает работать
# Debezium: переподключается — слот найден на новом мастере

# на новом мастере:
psql -c "SELECT slot_name, active FROM pg_replication_slots;"
-- orderflow_slot | f → t (после переподключения Debezium)

Зачтено, когда ни один запрос пользователя не упал с 500, а слот пережил промоцию. Запишите в README, сколько миллисекунд занял failover.

S3. CDC: Debezium → Kafka

Что строим. Kafka + Kafka Connect с Debezium PostgreSQL connector (pgoutput, Модуль 6.2). Топики — orderflow.public.orders, orderflow.public.payments; ключ сообщения — PK таблицы, порядок по сущности сохранён.

json
{
  "name": "orderflow-connector",
  "config": {
    "connector.class": "io.debezium.connector.postgresql.PostgresConnector",
    "database.hostname": "patroni-lb",
    "database.port": "5432",
    "database.user": "debezium",
    "database.dbname": "orderflow",
    "slot.name": "orderflow_slot",
    "publication.autocreate.mode": "filtered",
    "table.include.list": "public.orders,public.payments,public.order_events",
    "topic.prefix": "orderflow",
    "tombstones.on.delete": "false",
    "decimal.handling.mode": "string",
    "time.precision.mode": "adaptive_time_microseconds",
    "heartbeat.interval.ms": "5000",
    "snapshot.mode": "initial"
  }
}

Отличия от черновика, которые стоят инцидента: plugin.name не задаём — в Debezium 2.x+ для PostgreSQL поддерживается только pgoutput, и это значение по умолчанию (свойство осталось, но указывать его незачем); tombstones.on.delete=false — как в Модуле 6.2, иначе в поток полетят пустые надгробия, которые Go-приёмник обязан фильтровать; decimal.handling.mode=string и time.precision.mode=adaptive_time_microseconds — иначе numeric приедет base64-числом, а timestamptz — epoch-числом, и синк молча напишет в витрину 1970 год.

heartbeat.interval.ms — не косметика: без heartbeat слот на тихой БД не двигает restart_lsn, и WAL копится (Модуль 12, вопрос про это обязательно будет).

Checkpoint S3. Метрика «коммит → Kafka»: время ts_ms из сообщения Debezium против времени записи в Kafka, гистограмма в Grafana. Зачтено, когда p99 ≤ 2 сек при обычной нагрузке — и вы знаете, где цифра живёт. (Оба значения — часы разных машин: NTP обязателен, иначе гистограмма врёт.)

S4. ClickHouse 2×2 + Go-синк

Что строим. Кластер 2 шарда × 2 реплики + Keeper ×3 (Модуль 5); Go-синк из Kafka в CH — батчи, версии, идемпотентность (Модуль 6.6).

sql
CREATE TABLE analytics.events_local ON CLUSTER orderflow
(
    version        UInt64,             -- source.lsn: версия строки, dedup + порядок
    tenant_id      UInt32,
    order_id       UInt64,
    event_time     DateTime64(3),      -- из updated_at PG, НЕ now() на стороне синка
    kind           LowCardinality(String),
    amount         Decimal(18, 2) DEFAULT 0,
    pay_latency_ms UInt32 DEFAULT 0,   -- обогащение в синке: нужно для p99 в S5
    _deleted       UInt8 DEFAULT 0     -- DELETE из PG = новая версия с флагом
)
ENGINE = ReplicatedReplacingMergeTree(version)
PARTITION BY toYYYYMM(event_time)
ORDER BY (tenant_id, order_id, event_time);

CREATE TABLE analytics.events ON CLUSTER orderflow
AS analytics.events_local
ENGINE = Distributed(orderflow, analytics, events_local, cityHash64(order_id));

Три решения в этом DDL, без которых чекпойнт S4 не сойдётся:

ReplicatedReplacingMergeTree(version), а не ReplicatedMergeTree. Обычный MergeTree ничего не схлопывает: повтор батча после kill -9 останется в данных навсегда, и «дублей нет» окажется неправдой.
Дедупликация Replacing работает по ключу сортировки — по тройке (tenant_id, order_id, event_time). Отсюда требование к event_time: брать время из события в PG, а не time.Now() в синке, иначе при повторе ключ «поедет» и версия не погасит дубль.
version = source.lsn (Модуль 6.4): монотонная позиция в WAL, а не timestamp — две транзакции в одну миллисекунду иначе недетерминированы.

Скелет синка — три инварианта: батч, вставка до коммита оффсетов, версия на каждой строке (полная реализация с метриками, DLQ и тестами — boilerplate/internal/sink):

go
const (
	maxBatchRows = 5_000 // 4.4/5.4: части плодятся от частоты вставок, не от объёма
	flushEvery   = 500 * time.Millisecond
)

func (s *Sink) Run(ctx context.Context) error {
	pending := make(map[RowKey]Row, maxBatchRows) // RowKey = (tenant, order, event_time)
	var inflight []kafka.Message
	ticker := time.NewTicker(flushEvery)
	defer ticker.Stop()

	for {
		m, err := s.reader.FetchMessage(ctx) // читаем, НЕ коммитим
		if err != nil {
			return s.shutdown(ctx, pending, inflight) // ctx отменён — дожимаем хвост
		}

		row, perr := s.parse(ctx, m) // конверт Debezium + обогащение pay_latency_ms
		if perr != nil {
			// flush → DLQ → commit: коммит оффсета poison-сообщения БЕЗ флаша
			// потерял бы уже накопленные строки (разбор — Модуль 6.2)
			if err := s.poison(ctx, m, perr, pending, inflight); err != nil {
				return err
			}
			continue
		}

		if prev, ok := pending[row.Key]; !ok || row.Version >= prev.Version {
			pending[row.Key] = row // внутри батча держим только последнюю версию
		}
		inflight = append(inflight, m)

		if len(pending) >= maxBatchRows || ticked(ticker) {
			// ИНВАРИАНТ: сначала Send (эффект), потом CommitMessages (отметка)
			if err := s.flush(ctx, pending, inflight); err != nil {
				s.metrics.FlushErrors.Inc()
				return err // без коммита оффсетов: Kafka отдаст батч снова
			}
			s.metrics.Flushed.Observe(float64(len(pending)))
			clear(pending)
			inflight = inflight[:0]
		}
	}
}

Повтор после падения синкера между вставкой и коммитом — те же строки с теми же version вставятся ещё раз, но ReplicatedReplacingMergeTree(version) схлопнёт их при мердже; витринные запросы дожимают дедуп через FINAL/argMax (Модуль 6.3). Это и есть exactly-once «по факту».

Checkpoint S4:

bash
docker kill -s KILL sink    # смерть посреди батча (SIGKILL)
docker compose up -d sink

sql
-- 1) сырых строк БОЛЬШЕ, чем ключей: повторы приехали, это норма at-least-once
SELECT count() AS raw_rows,
       uniqExact((tenant_id, order_id, event_time)) AS keys
FROM analytics.events
WHERE kind = 'created';

-- 2) после схлопывания версий — ровно по одной строке на ключ
SELECT count() AS final_rows
FROM analytics.events FINAL
WHERE kind = 'created';
-- final_rows == keys: дублей в актуальном состоянии нет

-- 3) сверка с источником правды (Модуль 6.5)
SELECT count(*) FROM orders WHERE status = 2;   -- в PostgreSQL

Зачтено, когда после серии убийств синкера на живой нагрузке final_rows == keys, сверка с PG сходится, а лаг возвращается в норму. Первый запрос здесь важнее второго: он доказывает, что повторы были и были корректно погашены, а не что «просто ничего не случилось».

S5. Аналитика: MV-агрегаты + Grafana

Что строим. Слой витрин из Модуля 7: MV досчитывает агрегаты при вставке, дашборд читает компактные состояния, а не сырые миллиарды. Три требования, которые легко нарушить: движок витрины — Replicated (иначе в 2×2 она не реплицируется), MV вешается на локальную таблицу и на каждом узле (ON CLUSTER), а деньги считаются после дедупликации по ключу.

sql
-- Слой 1 (схема из Модуля 7.5): сводим версии заказа к одной строке ДО денег
CREATE TABLE analytics.orders_1m ON CLUSTER orderflow
(
    minute    DateTime,
    tenant_id UInt32,
    order_id  UInt64,
    version   SimpleAggregateFunction(max, UInt64),
    status    AggregateFunction(argMax, LowCardinality(String), UInt64),
    amount    SimpleAggregateFunction(max, Decimal(18, 2)),
    pay_ms    SimpleAggregateFunction(max, UInt32),
    deleted   SimpleAggregateFunction(max, UInt8)
)
ENGINE = ReplicatedAggregatingMergeTree(
    '/clickhouse/tables/{shard}/orders_1m', '{replica}')
PARTITION BY toYYYYMM(minute)
ORDER BY (tenant_id, minute, order_id);

-- MV — на events_local, не на Distributed: триггер срабатывает там, куда
-- физически лёг блок, то есть на локальной таблице каждого шарда.
CREATE MATERIALIZED VIEW analytics.mv_orders_1m ON CLUSTER orderflow
TO analytics.orders_1m AS
SELECT
    toStartOfMinute(event_time) AS minute,
    tenant_id,
    order_id,
    max(version)                AS version,   -- max идемпотентен к повторам вставок
    argMaxState(kind, version)  AS status,
    max(amount)                 AS amount,    -- max(дубль, дубль) = значение;
    max(pay_latency_ms)         AS pay_ms,    -- sum здесь задвоил бы деньги (7.2, ловушка 3)
    max(_deleted)               AS deleted
FROM analytics.events_local
GROUP BY minute, tenant_id, order_id;

sql
-- панель дашборда: выручка и p99 платежей за сутки
SELECT
    minute,
    tenant_id,
    count()                       AS orders,   -- по одной строке на заказ
    sum(amount)                   AS revenue,
    quantileTDigest(0.99)(pay_ms) AS pay_p99
FROM (
    SELECT minute, tenant_id, order_id,
           argMaxMerge(status) AS status,      -- AggregateFunction → -Merge
           max(amount)         AS amount,      -- SimpleAggregateFunction → max()
           max(pay_ms)         AS pay_ms,
           max(deleted)        AS deleted
    FROM analytics.orders_1m
    WHERE minute >= now() - INTERVAL 24 HOUR
    GROUP BY minute, tenant_id, order_id
)
WHERE status = 'paid' AND deleted = 0
GROUP BY minute, tenant_id
ORDER BY minute;

Когда p95 этого запроса упрётся в скан слоя 1 (Модуль 7.4 — по метрике, не «для архитектуры»), добавляется слой 2 со -State-колонками, который обновляется по расписанию с водяным знаком; читается он через sumMerge / uniqMerge / quantileTDigestMerge — но не через sumSum: такого комбинатора нет.

Grafana: datasource ClickHouse, три панели — revenue по минутам, p99 платежей, топ товаров; алерты из Модуля 8.3 рядом. Checkpoint S5: запрос дашборда < 300ms на объёме после S6-нагрузки; проверяется clickhouse-benchmark-ом или таймингом в UI.

S6. Нагрузка и намеренный инцидент

Нагрузка. Два генератора: pgbench давит OLTP-часть, k6 — API.

bash
pgbench -c 50 -j 4 -T 600 -P 10 -f bench/order_mix.sql orderflow

javascript
// k6: ramping-vus на POST /orders
import http from 'k6/http';
import { check } from 'k6';

export const options = {
  scenarios: {
    orders: {
      executor: 'ramping-vus',
      startVUs: 10,
      stages: [
        { duration: '2m', target: 200 },
        { duration: '5m', target: 200 },
        { duration: '1m', target: 0 },
      ],
    },
  },
};

export default function () {
  const res = http.post('http://api:8080/orders',
    JSON.stringify({ tenant_id: 1, total: 42.5 }),
    { headers: { 'Content-Type': 'application/json' } });
  check(res, { 'status 201': (r) => r.status === 201 });
}

Инцидент. На пике нагрузки — docker pause sink на 10 минут. Смотрите на графике: растёт consumer lag, растёт слот-лаг, срабатывают алерты из Модуля 8.3. Потом docker unpause — пайплайн доигрывает очередь, дубли схлопываются версиями. Разбор — постмортем в README: таймлайн, где было видно за N минут до паники, что сработало, что нет.

Checkpoint S6: lag < 30 сек при пиковой вставке; отчёт p99/throughput по обоим генераторам; постмортем.

📦 Финальный артефакт

orderflow-analytics/
├── docker-compose.yml        # профили: pg, kafka, ch, obs, app
├── patroni/                  # patroni.yml ×3, etcd
├── pgbouncer/                # pgbouncer.ini
├── debezium/                 # конфиг коннектора
├── clickhouse/               # DDL кластера + MV + витрины
├── services/                 # Go: api, sink (с тестами идемпотентности)
├── bench/                    # pgbench-сценарий, k6-скрипт
├── grafana/                  # дашборды + правила алертов
└── README.md                 # ADR: топология и «почему»

README — это не инструкция «как поднять», а ADR «почему такая топология». Формат каждой записи:

markdown
## ADR-002: CDC через Kafka, а не напрямую в ClickHouse
**Контекст:** латентность аналитики ≤ 30 сек; CH может простаивать на деплоях.
**Решение:** Debezium → Kafka → Go-синк.
**Альтернативы:** Kafka engine в CH (меньше кода, нет replay-контроля);
прямой http-синк (нет буфера на простой CH).
**Цена:** ещё один stateful-компонент (Kafka) и его мониторинг.

Минимум ADR: топология PG (Patroni ×3), топология CH (2×2), CDC-транспорт, модель дедупликации (Replacing + lsn), выбор PgBouncer transaction mode.

✅ Чек-лист код-ревью по спринтам

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

1. Почему после failover Patroni CDC-пайплайн продолжил работать без потери истории?

2. Какие лаги меряются в S6 против «lag < 30 сек при пиковой вставке»? — выберите все подходящие

3. Почему дашборд из S5 отвечает < 300ms, хотя в raw — сотни миллионов строк?

4. Что вы увидите, запросив витрину по ReplacingMergeTree без FINAL сразу после массовой вставки?

5. Почему kill -9 синкера посреди батча не создаёт дублей в отчётах?

6. EXPLAIN на order_events с фильтром по дате показал «Subplans Removed: 87». Что это значит?

7. Зачем в README проекта ADR, а не только инструкция по установке?

8. Как измеряется checkpoint S3 «сообщение в Kafka ≤ 2 сек после коммита»?

9. Зачем PgBouncer в transaction mode перед кластером из трёх узлов?

10. kill -9 мастера при асинхронной репликации: что может потеряться?

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

Собрать стенд end-to-end: PG (Patroni ×3 + PgBouncer) → Debezium → Kafka → Go-синк → ClickHouse 2×2 → Grafana.
Проверить каждый спринт экспериментом над отказом, а не галочкой.
Показать prune в EXPLAIN, живой слот после failover, отсутствие дублей после kill -9.
Обосновать каждую топологию ADR с альтернативами и ценой.
Провести нагрузку pgbench + k6 и назвать bottleneck по данным.
Разобрать собственный инцидент в формате постмортема.

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

Стенд собран и выдержал ваши собственные атаки. Осталось последнее: финальный баттл — 50 вопросов по всему курсу (15 интерактивных + 35 разборов «вопрос → эталон») и защита MVP: чек-лист ревью и типовые вопросы «почему так, а не иначе» с эталонными ответами. Берите стенд с собой — защищать будете его.

← Модуль 10. pgvector и RAG для LLM (опциональный) 🧪Модуль 12. Финал: баттл и защита →