Модуль 6. Пайплайн OLTP → OLAP: синхронизация двух миров

Цель модуля: построить пайплайн PostgreSQL → ClickHouse, который не теряет и не дублирует данные: сравнить стратегии синхронизации и осознанно выбрать CDC, собрать цепочку Debezium → Kafka → CH двумя способами (движковая таблица и Go-синк), спроектировать модель вставки под DELETE/UPDATE, мониторить лаг на всёх трёх участках и честно ответить на вопрос «что будет, когда всё сломается».

Перед началом: Модули 1–5. Особенно: логическая репликация и Patroni (3.2), Debezium (3.3), MergeTree-семейство (4.2), MV (4.6), Kafka-движок и clickhouse-go (4.8–4.9). Инструменты: docker-compose курса (профили pg, kafka, ch), Go 1.22+, clickhouse-go v2. Трудоёмкость: ~12 ч (теория 4 ч · практика 7 ч · квиз 1 ч).

🔥 Инцидент: дашборд, которому перестали верить

OrderFlow вырос: заказы и платежи живут в PostgreSQL, аналитика — в ClickHouse, между ними течёт CDC-поток. В пятницу на планёрке сошлись две цифры: дашборд показывал выручку за неделю 12,4 млн, а выгрузка из PostgreSQL — 12,1 млн. Расхождение стабильное, 2–3 % каждую неделю, и тянется оно уже месяц.

Сначала грешили на сам отчёт: фильтры по датам, часовые пояса, отменённые заказы, округления. Запросы переписали, сверили каждый JOIN — расхождение не ушло, только меняло знак. Тогда построили сверку по order_id: в ClickHouse оказалось 34 тысячи лишних строк. Причина нашлась в логах Kafka — коннектор перезапускался в среду, часть батчей приехала повторно, и каждый продублированный заказ честно попал в выручку.

Ещё хуже: за два дня до этого пайплайн отставал на пять часов, и об этом никто не узнал. Ни алерта, ни красной панели в Grafana — лаг слота в PostgreSQL, consumer lag и свежесть данных в ClickHouse не измерял никто. Дашборд показывал позавчерашние цифры как вчерашние, и заметили это случайно.

Дальше — про то, как связать два мира без потерь и дублей: стратегии синхронизации, цепочка Debezium, модели вставки под UPDATE и DELETE, идемпотентность и мониторинг всех трёх участков лага со сверкой PG ↔ CH.

📖 6.1. Стратегии синхронизации: dual-write, CDC, batch ETL, FDW

Два хранилища — и одна правда. PostgreSQL — источник правды (OLTP), ClickHouse — производная проекция (OLAP). Центральный вопрос модуля: как изменения попадают из одного в другой без потерь и без дублей — и что за это платится.

Антипаттерн: dual-write

Самое «очевидное» решение — писать в обе СУБД из прикладного кода:

go
// ❌ АНТИПАТТЕРН. Разбираем, почему — не копируйте в прод.
func (s *OrderService) Create(ctx context.Context, o Order) error {
	if err := s.pg.InsertOrder(ctx, o); err != nil {
		return err // PG не записал — честно отдаём ошибку, тут всё ок
	}
	// PG уже закоммитил. Если здесь упадёт что угодно...
	if err := s.ch.InsertAnalytics(ctx, o); err != nil {
		// ...данные в PG есть, в CH — нет. И ретрай небезопасен:
		// таймаут ≠ отказ, неизвестно, долетел ли INSERT до CH:
		// повтор — риск дубля, не повтор — гарантированная потеря.
		return fmt.Errorf("pg committed, ch failed: %w", err)
	}
	return nil
}

Почему это сломано в корне, а не «иногда сбоит»:

1. Нет атомарности. Два независимых коммита; окно между ними — место аварий: рестарт, деплой, network partition, kill -9.
2. Ретрай небезопасен. Таймаут ≠ «не вставилось»: неопределённый исход делает безопасную ретрай-логику невозможной (дубль или потеря).
3. Компенсация — вторая распределённая транзакция со всеми теми же проблемами.
4. Обходные пути записи. psql-миграция, data-fix, второй сервис — любые записи в PG мимо вашего кода молча не попадают в CH. А они всегда случаются.

2PC не спасает: ClickHouse в распределённых транзакциях не участвует, а координатор — ещё одна точка отказа ценой доступности. Вывод, на котором построен модуль: атомарность должен обеспечивать один источник — WAL PostgreSQL. Закоммитили транзакцию — изменения уже «сохранены» для всех подписчиков; дальше вопрос доставки, а не записи.

CDC: изменения из WAL

CDC (Change Data Capture) читает WAL через логическое декодирование (pgoutput): каждое INSERT/UPDATE/DELETE становится событием, Debezium публикует его в Kafka, приёмник материализует в ClickHouse. Ценность — в свойствах, которых нет у альтернатив:

• Полный охват: транзакция в WAL атомарна; сюда попадают и «обходные» записи.
• Порядок per-key: Debezium маршрутизирует по PK → события одной строки в одной партиции → порядок на сущность сохранён.
• Latency — секунды, источник не нагружается запросами (только чтение WAL).
• Прикладной код не трогаем: коннектор — инфраструктура, а не условие в каждом репозитории.

Цена честная: постоянная инфраструктура (Kafka + Connect), слот репликации со своими слабыми местами (6.5), эволюция схемы (Модуль 8).

Batch ETL и FDW: нишевые инструменты

Batch ETL — выгрузка по расписанию (COPY → CSV/Parquet → загрузка): просто, дёшево, консистентный снапшот на момент выгрузки (REPEATABLE READ). Минусы: latency — часы (T+1), нагрузка на PG на время прогона, повторный прогон требует дедупликации. Ниша: историческая отчётность, бэкфиллы, переезды.

FDW (postgres_fdw, clickhouse_fdw) — «чужая» таблица выглядит локальной, данные едут по сети в момент запроса. Ноль синхронизации и всегда актуально — но каждая строка по сети, пушдаун агрегатов ограничен, тяжёлый запрос бьёт по OLTP-хосту. Для разовых ad-hoc сверк — да; для продового пайплайна — нет.

Таблица выбора

Правило курса: CDC — транспорт по умолчанию; batch — для истории и переездов; FDW — для точечных сверок; dual-write — не для этого.

📖 6.2. LAB: цепочка Debezium — PostgreSQL → Kafka → ClickHouse

text
[PG WAL] → (logical decoding, pgoutput) → [slot: orders_cdc]
        → [Debezium connector] → [topic: dbz.public.orders]
        → [приёмник: Kafka-движок CH  ИЛИ  Go-синк] → [orders_ch]

Показать кодКопировать код

На схеме — весь маршрут и три точки, которые обязаны быть под метриками: slot lag (PG не успевает отдавать WAL — слот растёт), consumer lag (Kafka → синк) и freshness (сверка PG ↔ CH). Отдельная ловушка слева: удаление слота — единственная операция, которая теряет данные без возможности догнать; слот снимают только при гарантированно остановленном коннекторе.

Шаг 1. PostgreSQL: логическая репликация

sql
-- postgresql.conf (в docker-compose курса уже настроено):
--   wal_level = logical, max_replication_slots = 10, max_wal_senders = 10
ALTER SYSTEM SET wal_level = 'logical';  -- требует рестарта кластера

CREATE PUBLICATION orders_pub FOR TABLE public.orders, public.payments;

CREATE ROLE debezium WITH LOGIN REPLICATION PASSWORD 'dbz_secret';
GRANT SELECT ON public.orders, public.payments TO debezium;

Слот orders_cdc создаст сам Debezium — руками его не создавайте и не удаляйте «для чистоты»: удаление слота = потеря позиции потока.

Шаг 2. Kafka-топик

bash
kafka-topics.sh --bootstrap-server kafka:9092 --create \
  --topic dbz.public.orders \
  --partitions 6 --replication-factor 3 \
  --config retention.ms=604800000   # 7 суток: запас на лежачий синкер и бэкфилл

Три решения в одной команде: партиции = параллелизм приёмника (Debezium кладёт события одного PK в одну партицию — порядок на заказ сохранён); cleanup.policy=delete, не compact — компактация выкинула бы промежуточные версии, а на репроцессинге нужна вся история; retention с запасом — лежавший сутки синкер по возвращении дочитает всё.

Шаг 3. Debezium-коннектор

json
{
  "name": "pg-orders-source",
  "config": {
    "connector.class": "io.debezium.connector.postgresql.PostgresConnector",
    "database.hostname": "pg-master",
    "database.port": "5432",
    "database.user": "debezium",
    "database.password": "${file:/run/secrets/pg.env:PG_PASSWORD}",
    "database.dbname": "shop",
    "topic.prefix": "dbz",
    "slot.name": "orders_cdc",
    "publication.name": "orders_pub",
    "publication.autocreate.mode": "disabled",
    "table.include.list": "public.orders,public.payments",
    "snapshot.mode": "initial",
    "tombstones.on.delete": "false",
    "decimal.handling.mode": "string",
    "time.precision.mode": "adaptive_time_microseconds",
    "heartbeat.interval.ms": "10000"
  }
}

bash
curl -X POST http://debezium:8083/connectors \
  -H 'Content-Type: application/json' -d @pg-orders-source.json

Разбор полей, которые решают:

• snapshot.mode=initial — при первом старте консистентный снапшот (op: r) с фиксацией LSN, затем стрим с этой позиции: снапшот и стрим сравнимы по LSN — дыр на стыке нет.
• tombstones.on.delete=false — после DELETE летит обычное событие op: d с before-блоком: приёмник проставит флаг удаления; пустые надгробия — лишний кейс для JSON-приёмников.
• decimal.handling.mode=string — NUMERIC едет строкой "1299.00": детерминированный парсинг (дефолт precise кодирует число в base64).
• time.precision.mode=adaptive_time_microseconds — timestamptz едет строкой ISO-8601 ("2026-09-15T12:34:56.789Z"), а не числом миллисекунд. Без этой настройки дефолт connect отдаёт epoch-число, и оба приёмника ниже молча распарсят его в 1970 год — самый противный класс багов: без ошибки, но с неверными данными в витрине.
• heartbeat.interval.ms=10000 — сердцебиение продвигает слот, даже когда изменения идут только в таблицы вне publication (иначе слот раздувает WAL, 6.5).
• Секрет — через secret-провайдер, не в git (Модуль 8.4).

Событие UPDATE в топике: op: u, before — старое состояние, after — новое, source.lsn — позиция в WAL. Запомним LSN: это будет колонка version в ClickHouse (6.4).

Шаг 4a. Приёмник №1: Kafka-движок ClickHouse

Вся цепочка — на SQL, ноль кода: таблица-очередь → MV с парсингом → целевая таблица.

sql
-- 1. Таблица-очередь: сама подписывается на топик
CREATE TABLE orders_queue
(
    raw String
)
ENGINE = Kafka
SETTINGS
    kafka_broker_list = 'kafka:9092',
    kafka_topic_list = 'dbz.public.orders',
    kafka_group_name = 'ch-orders-sink',
    kafka_format = 'JSONEachRow',
    kafka_skip_broken_messages = 10,
    kafka_num_consumers = 1;

-- 2. Целевая таблица: ReplacingMergeTree с версией из LSN
CREATE TABLE orders_ch
(
    order_id   UInt64,
    status     LowCardinality(String),
    amount     Decimal(18, 2),
    updated_at DateTime64(3),
    version    UInt64,
    _deleted   UInt8 DEFAULT 0
)
ENGINE = ReplacingMergeTree(version)
PARTITION BY toYYYYMM(updated_at)
ORDER BY order_id;

-- 3. MV: парсинг Debezium-конверта; coalesce(nullIf(after..), before..)
--    берёт after, а для DELETE (after пуст) — before
CREATE MATERIALIZED VIEW orders_ch_mv TO orders_ch AS
SELECT
    toUInt64OrZero(coalesce(nullIf(JSONExtractString(raw, 'after', 'id'), ''),
        JSONExtractString(raw, 'before', 'id'))) AS order_id,
    coalesce(nullIf(JSONExtractString(raw, 'after', 'status'), ''),
        JSONExtractString(raw, 'before', 'status')) AS status,
    toDecimal64OrZero(coalesce(nullIf(JSONExtractString(raw, 'after', 'amount'), ''),
        JSONExtractString(raw, 'before', 'amount')), 2) AS amount,
    parseDateTime64BestEffort(coalesce(nullIf(JSONExtractString(raw, 'after', 'updated_at'), ''),
        JSONExtractString(raw, 'before', 'updated_at')), 3) AS updated_at,
    JSONExtractUInt64(raw, 'source', 'lsn') AS version,
    toUInt8(JSONExtractString(raw, 'op') = 'd') AS _deleted
FROM orders_queue
WHERE JSONHas(raw, 'source');  -- отсекаем служебные/битые сообщения

Удаление становится строкой с _deleted = 1, а не отсутствием — почему так, в 6.4.

Шаг 4b. Приёмник №2: Go-синк на clickhouse-go v2

Тот же контракт — руками. Клиент Kafka любой (здесь segmentio/kafka-go); важна семантика: флаш батча в CH, потом коммит оффсетов.

go
// boilerplate/internal/sink. Imports: context, encoding/json, time,
// clickhouse-go/v2, shopspring/decimal, segmentio/kafka-go, prometheus/client_golang.

const insertSQL = `INSERT INTO orders_ch (order_id, status, amount, updated_at, version, _deleted)`

// OrderRow: OrderID, Status, Amount decimal.Decimal, UpdatedAt time.Time,
// Version uint64 (source.lsn: монотонная позиция в WAL), Deleted uint8.
// Sink: reader *kafka.Reader; ch driver.Conn; batchSize (напр. 5000 —
// parts не плодим, 4.4/5.4); dlq func(kafka.Message, error) error — топик + метрика + алерт.

// parseEnvelope(b []byte) (OrderRow, error): json.Unmarshal конверта Debezium
// (after — новое состояние, у DELETE — nil → берём before; source.lsn — версия;
// decimal.handling.mode=string → amount строкой, decimal.NewFromString;
// time.precision.mode=adaptive_time_microseconds → updated_at строкой ISO-8601,
// time.Parse(time.RFC3339Nano, ...)), Deleted=1 при Op=="d".
// Полный код с юнит-тестами на фикстурах сообщений — boilerplate/internal/sink/envelope.go.

func (s *Sink) flush(ctx context.Context, pending map[uint64]OrderRow, inflight []kafka.Message) error {
	if len(pending) == 0 {
		return nil
	}
	b, err := s.ch.PrepareBatch(ctx, insertSQL)
	if err != nil {
		return err
	}
	for _, r := range pending {
		if err := b.Append(r.OrderID, r.Status, r.Amount, r.UpdatedAt, r.Version, r.Deleted); err != nil {
			_ = b.Abort()
			return err
		}
	}
	if err := b.Send(); err != nil {
		return err
	}
	// ИНВАРИАНТ ИДЕМПОТЕНТНОСТИ: сначала эффект (Send), потом отметка (Commit).
	// Падение между ними → Kafka повторит батч → те же версии погасятся
	// в ReplacingMergeTree. Обратный порядок = потеря данных (6.6).
	return s.reader.CommitMessages(ctx, inflight...)
}

func (s *Sink) Run(ctx context.Context) error {
	pending := make(map[uint64]OrderRow, s.batchSize) // in-batch dedup: ключ → последняя версия
	var inflight []kafka.Message
	tick := time.NewTicker(500 * time.Millisecond) // флаш хвоста по таймеру
	defer tick.Stop()

	reset := func() {
		clear(pending)
		inflight = inflight[:0]
	}

	for {
		m, err := s.reader.FetchMessage(ctx) // блокируется; при отмене ctx — ошибка
		if err != nil {
			// graceful shutdown: ctx уже отменён, поэтому дожимаем хвост
			// отдельным контекстом — иначе коммит оффсетов упадёт вместе с ctx
			shutdownCtx, cancel := context.WithTimeout(context.WithoutCancel(ctx), 5*time.Second)
			defer cancel()
			return s.flush(shutdownCtx, pending, inflight)
		}

		row, perr := parseEnvelope(m.Value)
		if perr != nil {
			// Poison-сообщение. Сначала ОБЯЗАТЕЛЬНО флашим накопленное: оффсет m
			// выше, чем у строк в pending, и коммит m «съел» бы их при падении —
			// тот самый commit-до-insert, который мы запрещаем в 6.6.
			if ferr := s.flush(ctx, pending, inflight); ferr != nil {
				return ferr
			}
			reset()

			if derr := s.dlq(m, perr); derr != nil {
				return derr // DLQ недоступен — не теряем сообщение молча
			}
			if cerr := s.reader.CommitMessages(ctx, m); cerr != nil {
				return cerr
			}
			continue
		}

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

		doFlush := len(pending) >= s.batchSize
		if !doFlush {
			select {
			case <-tick.C:
				doFlush = true
			default:
			}
		}
		if doFlush {
			if err := s.flush(ctx, pending, inflight); err != nil {
				return err // без коммита оффсетов: Kafka отдаст батч снова
			}
			reset()
		}
	}
}

// main(): clickhouse.Open + kafka.NewReader с GroupID "ch-orders-sink"
// (оффсеты в Kafka: рестарт продолжит с закоммиченного) + супервизор:
// ошибка → backoff → Run() снова. Повторы Kafka — не авария, а расчётный режим (6.6).

Два места, где этот код ломается в ревью чаще всего, — держите их в голове:

1. DLQ без предварительного флаша. Коммит оффсета poison-сообщения продвигает группу дальше неотправленных строк: при падении процесса они теряются навсегда. Порядок жёсткий: flush → DLQ → commit.
2. clear(pending) до Send. Сбрасывать буфер можно только после успешной вставки; иначе ретрай отправит пустоту, а Kafka считает батч обработанным.

Сравнение приёмников

Вердикт: Kafka-движок — быстрый старт и нормальный вариант для «складывай как есть» потоков. Как только нужны обогащение, нетривиальные трансформации, метрики и тесты — Go-синк. В MVP (Модуль 11, спринт S4) идём Go-синком.

📖 6.3. Модели вставки: ReplacingMergeTree + version vs Collapsing + sign

CDC приносит INSERT/UPDATE/DELETE — а ClickHouse append-only. Задача: закодировать «изменяемое состояние» в потоке только-вставок. Два канонических решения.

Вариант 1: ReplacingMergeTree(version)

Каждое событие — полная строка с монотонной версией; при мерже из версий одного ключа выживает максимальная:

sql
CREATE TABLE orders_ch
(
    order_id   UInt64,
    status     LowCardinality(String),
    amount     Decimal(18, 2),
    updated_at DateTime64(3),
    version    UInt64,      -- source.lsn: монотонный в WAL
    _deleted   UInt8 DEFAULT 0
)
ENGINE = ReplacingMergeTree(version)
PARTITION BY toYYYYMM(updated_at)
ORDER BY order_id;

Ментальная модель: «последняя версия побеждает». Десять UPDATE — десять строк до мержа, одна после. Важно: дедупликация не мгновенна — мержи асинхронны, между вставкой и мержем дубли видны любому SELECT. Это контракт, а не баг: запрос обязан быть готов к дублям (FINAL/argMax, 6.4).

Вариант 2: (Versioned)CollapsingMergeTree + sign

UPDATE кодируется парой событий: отменить старое состояние (sign = -1), утвердить новое (sign = +1). Актуальное состояние — сумма знаков по ключу:

sql
CREATE TABLE orders_collapsed
(
    order_id   UInt64,
    status     LowCardinality(String),
    amount     Decimal(18, 2),
    updated_at DateTime64(3),
    sign       Int8,        -- +1 = состояние, -1 = отмена
    version    UInt64
)
ENGINE = VersionedCollapsingMergeTree(sign, version)
ORDER BY order_id;

Именно VersionedCollapsing: классический CollapsingMergeTree сворачивает пары +1/-1 только при правильном порядке строк, а в CDC-пайплайне порядок не гарантирован (повторы Kafka, ретраи, параллельные вставки) — Versioned-вариант сворачивает по версии и порядку не доверяет. Цена модели — запросы («сколько живых заказов» — уже не count()):

sql
SELECT count() FROM orders_collapsed
WHERE order_id = 42
GROUP BY order_id
HAVING sum(sign) > 0;

И главная опасность: модель не самовосстанавливается. Потерялась -1 — навсегда висит фантомный +1; задвоилась -1 из-за ретрая — строка исчезла из состояния. Движок это не починит: знаки не несут информации, где правда.

Сравнение и выбор

В CDC-пайплайне выбор почти всегда ReplacingMergeTree + version: модель прощает всё, что делает с потоком Kafka. VersionedCollapsing — ниша: горячие счётчики с тысячами обновлений в секунду, где важна экономия строк до мержа, и команда готова платить дисциплиной знаков.

📖 6.4. DELETE и UPDATE: чего в CH нет и как без них жить

Формально ALTER TABLE ... DELETE и lightweight DELETE FROM существуют — но это мутации: тяжёлые фоновые перезаписи частей, не на каждую строку пайплайна. Пайплайн делает иначе: DELETE из PostgreSQL превращается в новую версию строки с флагом _deleted = 1 (значения — из before-блока), UPDATE — в новую версию с большей версией. Физически ничего не удаляется — логически помечается и вытесняется мержем.

Чтение актуального состояния

sql
-- точечное чтение по ключам: argMax без FINAL
SELECT
    order_id,
    argMax(status,   version) AS status,
    argMax(amount,   version) AS amount,
    argMax(_deleted, version) AS deleted
FROM orders_ch
WHERE order_id IN (42, 43, 44)
GROUP BY order_id
HAVING deleted = 0;

-- отчёт по диапазону: FINAL + фильтр удалённых (см. также сверку в 6.5)
SELECT status, count() AS cnt, sum(amount) AS revenue
FROM orders_ch FINAL
WHERE _deleted = 0 AND updated_at >= now() - INTERVAL 30 DAY
GROUP BY status;

Честно про FINAL: он схлопывает части на лету — дороже обычного SELECT, но параллелится и на сотнях миллионов строк даёт секунды. Дашборды — на агрегатах (Модуль 7), FINAL — на сверках; упёрлись — argMax по диапазону партиций.

Версия — из pg_lsn, а не из timestamp

• LSN монотонно растёт в WAL: две транзакции в одну миллисекунду дают разные версии. При равных updated_at ReplacingMergeTree оставит произвольную строку — недетерминизм в данных.
• Снапшот Debezium фиксирует консистентный LSN: версии снапшота (op: r) и стрима сравнимы напрямую — дыр и «старое поверх нового» на стыке нет.

Eventual consistency — это контракт, а не извинение

Между COMMIT в PG и видимой строкой в CH: flush WAL → декодирование → Kafka → fetch синкером → вставка → мерж. Секунды в норме, больше под нагрузом. Следствия, которые проговаривают с командой заранее:

• Аналитика никогда не совпадает с OLTP «в моменте» — совпадает с точностью до лага. Задайте SLO на лаг («p99 лага < 30 сек») и мониторьте его (6.5).
• На CH не строят read-your-writes: «создал заказ → увидел в отчёте» гарантировать нельзя. Свежая точка для пользователя — из PG; CH — для трендов и агрегатов.
• Вопрос «а почему в отчётах ещё старый статус?» — признак несогласованного SLO на лаг, а не сломанного пайплайна.

📖 6.5. Мониторинг пайплайна: три лага и сверка

text
[PG] --лаг слота--> [Debezium] --consumer lag--> [синкер] --лаг вставки--> [CH]
      pg_replication_slots       kafka-consumer-groups        сверка PG ↔ CH

Лаг №1: слот на стороне PostgreSQL

sql
SELECT slot_name, active, wal_status,
       pg_size_pretty(pg_wal_lsn_diff(pg_current_wal_lsn(), confirmed_flush_lsn)) AS retained_wal
FROM pg_replication_slots
WHERE slot_type = 'logical';

Интерпретация: active = false — коннектор мёртв (алерт сразу). retained_wal растёт — слот не продвигается, PG удерживает WAL: диск растёт на OLTP-источнике из-за медленного консьюмера — главный способ, которым CDC-пайплайн роняет базу транзакций. wal_status = 'lost' / слот invalidated — сработал max_slot_wal_keep_size (PG13+): диск спасён, но данные за границей удержания потеряны для этого слота навсегда — в CH дыра (как штопать — см. квиз). Алерты: active = false дольше N минут; retained_wal > порога; invalidated-слоты.

Лаг №2: consumer group

bash
kafka-consumer-groups.sh --bootstrap-server kafka:9092 \
  --describe --group ch-orders-sink
# TOPIC              PARTITION  CURRENT-OFFSET  LOG-END-OFFSET  LAG
# dbz.public.orders  0          812345          812401          56

Как в брокерском курсе: lag без контекста — половина правды. Смотрите тренд и возраст старейшего необработанного сообщения (kafka-exporter отдаёт обе метрики). Go-синк экспортирует свои: sink_kafka_lag, sink_flush_duration_seconds, sink_flush_errors_total, sink_dlq_total.

Лаг №3: сверка счётчиков PG ↔ CH

sql
-- PostgreSQL: срез по дням
SELECT date_trunc('day', updated_at)::date AS day,
       count(*) AS rows, sum(amount) AS revenue
FROM orders
WHERE updated_at >= now() - INTERVAL '3 days'
GROUP BY day ORDER BY day;

sql
-- ClickHouse: тот же срез
SELECT toDate(updated_at) AS day,
       count() AS rows, sum(amount) AS revenue
FROM orders_ch FINAL
WHERE _deleted = 0
  AND updated_at >= now() - INTERVAL 3 DAY
GROUP BY day ORDER BY day;

Как читать расхождения: маленькое и сокращающееся со временем — in-flight лаг, норма в пределах SLO; стабильное или растущее — инцидент: дыра (слот), сломанный парсинг, фильтр publication. Drill-down: срез по часам → по диапазонам ключей → точные потерянные ID → точечный бэкфилл. Полная сверка большой таблицы тяжела для обеих СУБД — гигиена: сверяйте свежие дни/партиции по расписанию. Утилита сверки — 30 строк на Go; в MVP она есть.

📖 6.6. Идемпотентность и exactly-once «по факту»

Гарантии по участкам — без самообмана

Kafka-транзакции (exactly-once semantics) работают между Kafka-топиками и не покрывают запись в ClickHouse. End-to-end exactly-once «из коробки» не существует — и не нужен. Собираем exactly-once по эффекту:

at-least-once на каждом хопе + идемпотентный приёмник = ровно один эффект.

Идемпотентность приёмника уже построена: ключ (ORDER BY order_id) + версия (source.lsn), ReplacingMergeTree поглощает повторы. Инвариант синкера, который держит всё: сначала эффект, потом отметка — вставка (Send) до коммита оффсетов. Падение между ними — повтор батча, безвредный из-за версий. Обратный порядок (commit → insert) — потерянный диапазон: оффсеты сдвинулись, данных нет, и ретраем это не чинится.

Тест с kill -9

Инварианты проверяются не верой, а тестом: продюсим поток с обновлениями, убиваем синкер посреди батча, рестартуем, сверяем с эталоном.

go
// sink_k9_test.go (сокращённо; полная версия — в практике)
func TestSink_SurvivesKill9(t *testing.T) {
	env := startEnv(context.Background(), t) // testcontainers: kafka + clickhouse

	const orders = 10_000
	produce(t, env.Kafka, genUpdates(orders, 5)) // 10k заказов × ~5 версий, вперемешку

	sink := startSinkProcess(env) // отдельный процесс — убиваем по-настоящему
	time.Sleep(2 * time.Second)
	sink.Signal(syscall.SIGKILL) // посреди батча, не на границе

	startSinkProcess(env)        // рестарт: дочитает с закоммиченных оффсетов
	waitLagZero(t, env, 30*time.Second)

	// ИНВАРИАНТ 1: актуальное состояние совпадает с эталоном входного потока
	assertStateEqual(t, env.CH, expectedState)

	// ИНВАРИАНТ 2: сырых строк больше, чем заказов (дубли повторов — норма),
	// после схлопывания — ровно orders:
	//   count() > orders И count() FINAL = orders
}

В проде этот же сценарий происходит сам, без теста — а дисциплина остаётся: версионные записи, идемпотентный приёмник, регулярная сверка (6.5) и kill-9-тест в CI. «Exactly-once» — не свойство брокера, а свойство вашей системы.

💻 Практика

1. Цепочка на месте. Поднимите профили pg, kafka, ch; настройте publication и коннектор (6.2), сделайте INSERT/UPDATE/DELETE в PG и найдите все три события в топике dbz.public.orders.
2. Приёмник №1. Соберите SQL-цепочку orders_queue → MV → orders_ch. Проверьте: UPDATE меняет статус в CH, DELETE ставит _deleted = 1.
3. Приёмник №2. Запустите Go-синк (другая consumer group), сравните лаг и удобство отладки обоих вариантов; запишите, какой взяли бы в прод и почему.
4. Kill -9. Продюсите нагрузку, через 2 секунды убейте синкер, рестарт. Убедитесь: count() ... FINAL после мержа равен эталону, сырой count() — больше. Это тест идемпотентности вживую.
5. Мониторинг и штопка. Остановите коннектор (active=false, retained_wal растёт), затем синкер (лаг растёт) — пропишите пороги алертов из 6.5. Затем удалите слот (симуляция failover-потери), найдите диапазон пропуска сверкой и залейте бэкфиллом (INSERT ... FROM postgresql(...) или batch ETL).
6. FINAL vs argMax. На ≥10 млн строк сравните время SELECT ... FINAL и argMax-запроса по ограниченному диапазону партиций. Цифры пригодятся в Модуле 7.

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

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

• Dual-write с таймаутом. Ретрай безопасен только при идемпотентном приёмнике — в dual-write его нет по построению. Не чинить dual-write, а убрать.
• Failover и потерянный слот. Слот — часть WAL-истории мастера; новый мастер начинает свою ветку WAL, и события между последним confirmed_flush_lsn и failover физически недоступны. Профилактика: pg_failover_slots (Модуль 3.2) и алерт на active=false; лечение: честный бэкфилл диапазона.
• count() без FINAL и commit до insert. Дубли версий до мержа — контракт ReplacingMergeTree (состояние читают через FINAL/argMax). А оффсеты, закоммиченные раньше эффекта, — единственный непоправимый вариант.
• Слот и heartbeat. Слот продвигается только событиями своей publication; heartbeat двигает его без событий. max_slot_wal_keep_size спасает диск молча — пара к нему: алерт на retained_wal + процедура бэкфилла.
• Двойной −1. Знаки не несут информации о «правде»: любую ошибку знаков движок честно схлопнет — в испорченное состояние. Отсюда выбор Replacing.
• Send до Commit / Kafka EOS. Повтор батча с теми же версиями безвреден — это и есть exactly-once по эффекту. EOS действует только внутри Kafka: как только эффект выходит в CH, идемпотентность приёмника — ваша работа.

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

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

• Объяснить, почему dual-write — антипаттерн (неопределённый исход, дубль или потеря при ретрае, обходные пути записи), и обосновать выбор CDC; назвать ниши batch ETL и FDW.
• Собрать цепочку PG (publication + slot) → Debezium → Kafka → CH двумя приёмниками — Kafka-движок и Go-синк — и обосновать выбор между ними.
• Выбрать модель вставки: ReplacingMergeTree + version по умолчанию; VersionedCollapsing — осознанная ниша; объяснить, почему Collapsing не самовосстанавливается.
• Обрабатывать UPDATE/DELETE версиями: _deleted-флаг, version из LSN, чтение через FINAL/argMax.
• Мониторить три лага: слот, consumer group, сверка PG ↔ CH — и отличать in-flight лаг от дыры.
• Реализовать exactly-once по эффекту: at-least-once + идемпотентный приёмник, инвариант «сначала Send, потом Commit», kill-9-тест в CI.

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

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

Пайплайн даёт в ClickHouse чистый версионный поток изменений — но запросы к сырым строкам с FINAL на каждом шаге дашборд не спасут. Дальше — моделирование аналитики: денормализация под запросы (забыть про 5НФ), AggregatingMergeTree и MV-агрегаты, приблизительные метрики с честной погрешностью и слои данных raw → агрегаты → дашборд. Соберём витрину «агрегаты заказов по минутам» прямо поверх CDC-потока этого модуля.