Модуль 3. Пайплайн PG → ClickHouse: CDC, лаги и дубли

Цель модуля: проследить путь заказа из WAL PostgreSQL в ClickHouse, разделить три лага пайплайна — слота, консьюмера и end-to-end — и найти застрявшее звено. Воспроизвести «дашборд отстаёт», вернуть догон и объяснить, откуда в аналитике берутся дубли.

Перед началом: WAL и репликация — полигон m2; CDC и Kafka — pg-clickhouse m6; MergeTree — pg-clickhouse m4/m7. Инструменты: Docker Compose v2, psql, curl, kafka-consumer-groups.sh, ClickHouse HTTP, make. Трудоёмкость: ~2 ч (разбор 40 мин · практика 60 мин · квиз 10 мин).

🔥 Инцидент: «дашборд отстаёт, а цифры не сходятся»

08:20, OrderFlow. Дашборд выручки отстаёт на двадцать минут, хотя API отвечает за 200 мс, PostgreSQL и реплика здоровы, ошибок в логах нет. Дежурный открывает сверку:

json
{"pg_orders":200013,"ch_orders":199987,"diff":26}

Ночью останавливали sink «на минуту» и не вернули. Debezium всё это время жив: слот orders_cdc двигается, события едут в Kafka. Но группа sink-ch их не читает — отстаёт последнее звено, а не источник. После make fix-3 sink догнал поток, однако ch_orders обогнал pg_orders: worker перевёл часть заказов в paid, и каждое UPDATE стало в ClickHouse новой строкой. В одном симптоме — два дефекта: отставание и дубли.

📖 3.1. Три лага пайплайна: слот, консьюмер, end-to-end

Пайплайн — три передачи, у каждой своя позиция. Смешивать их — главная ошибка диагностики.

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

На схеме четыре независимые точки измерения: лаг реплики (replay WAL), лаг слота (сколько WAL Debezium ещё не подтвердил), лаг консьюмера (сколько сообщений не прочитал sink) и end-to-end (сколько прошло от created_at до строки в ClickHouse). Растёт всегда ровно одна из них — по ней и определяется виновное звено.

Слот считается в байтах WAL, консьюмер — в сообщениях, end-to-end — в строках и секундах. Они не пропорциональны: остановите sink, и лаг группы вырастет на все события, а лаг слота останется near-zero — Debezium их уже прочитал и подтвердил. Первый вопрос дежурного: «какой именно лаг растёт?»

📖 3.2. Debezium и слот репликации: почему пауза опасна

Коннектор orderflow-connector читает public.orders через логический слот orders_cdc, плагин pgoutput, публикация filtered. Префикс топика cdc даёт cdc.public.orders; на мастере включены wal_level=logical и запас слотов и walsender'ов.

При первом старте Debezium делает снапшот таблицы — на стенде ~200 тысяч строк. Первичная заливка в ClickHouse идёт минуты, и большой лаг консьюмера в этот момент — нормальный догон, а не инцидент. Отличайте холодный старт от застоя: в первом случае лаг монотонно падает, во втором — стоит или растёт.

Слот удерживает WAL до confirmed_flush_lsn: пока Debezium не подтвердил позицию, PostgreSQL не удаляет эти сегменты. Долгая пауза коннектора — растущий pg_wal и риск переполнить диск. Удалить слот «чтобы отпустило» нельзя без последствий: пропадут позиции, и коннектору придётся делать снапшот заново. make debezium пересоздаёт коннектор и ждёт RUNNING, но слот и offset — состояние пайплайна, а не расходник. Устройство CDC — pg-clickhouse m6.

📖 3.3. Sink: батчи, at-least-once и откуда дубли

Sink читает топик cdc.public.orders группой sink-ch. Источник правды — WAL, а не топик orders, куда api публикует события для worker. Батч набирается до 200 событий или 500 мс, затем одним INSERT ... FORMAT JSONEachRow уходит в orderflow.orders_ch.

Offset коммитится вручную и только после успешного INSERT. При ошибке ClickHouse sink увеличивает sink_errors_total и не коммитит батч; при перезапуске или ребалансе незакоммиченные сообщения перечитаются — это at-least-once. Отсюда дубли: повторная доставка допустима по контракту.

Второй источник дублей — UPDATE. Worker ставит status = 'paid', Debezium публикует событие с новым after, а sink пишет его ещё одной строкой: orders_ch — обычный MergeTree без замены по ключу. DELETE-события sink пропускает (after = null), поэтому удаления в аналитику не доезжают.

Метрики sink — sink_events_total, sink_errors_total, sink_batch_size — на :9101; Prometheus скрейпит их как job sink. sink_batch_size — размер последнего батча, а не очередь: не принимайте его за лаг. Батчинг и «too many parts» — pg-clickhouse m5.

📖 3.4. Сверка PG и CH: что показывает /stats

GET /stats отдаёт {"pg_orders":N,"ch_orders":M,"diff":N-M}: count(*) по orders на мастере, count() по orders_ch и их разницу. В core-профиле CLICKHOUSE_URL пуст, ключа ch_orders нет, и check-3 падает с подсказкой «make up-full».

Главное ограничение: diff = 0 не доказывает корректность. Это разница количеств, а не содержимого. UPDATE добавляет строки и толкает ch_orders вверх; DELETE в PostgreSQL не удаляет строку из ClickHouse; повторная доставка батча тоже добавляет копии. Поэтому diff может быть нулевым при потерях и дублях одновременно. check-3 это учитывает: запоминает ch_orders, создаёт заказ и до 60 секунд ждёт роста, а не равенства. На успехе он печатает лаг слота orders_cdc, на неудаче — «событие не доехало до ClickHouse за 60с».

📖 3.5. Дедупликация: ReplacingMergeTree и версии (теория)

ReplacingMergeTree(version) хранит версию строки и при мерже оставляет запись с максимальной версией для каждого ключа сортировки. Схема: ORDER BY id, версия — ts_ms или LSN из конверта Debezium. Замена происходит при мерже, а не в момент INSERT; до мержа обе версии лежат в таблице, и запрос без FINAL (или argMax по версии) увидит дубли.

Для удалений нужен либо CollapsingMergeTree с колонкой-признаком sign (+1 при вставке, −1 при удалении), либо ReplacingMergeTree с флагом is_deleted и фильтрацией на чтении. Оба требуют, чтобы sink знал операцию (c/u/d) и версию — то есть разбирал конверт Debezium целиком, а не только after.

В стенде этого нет: orders_ch — append-only MergeTree с ORDER BY (created_at, id). Дубли здесь не баг настройки, а следствие модели. Как строить аналитику поверх CDC — pg-clickhouse m7.

📖 3.6. Компромиссы

• At-least-once vs at-most-once. Sink выбран at-least-once: данные не теряются ценой возможных дублей. Потеря события в аналитике хуже лишней строки.
• Снапшот vs инкремент. Снапшот прост и не требует начального offset, но грузит PostgreSQL и даёт минуты большого лага; инкрементальный старт дешевле, но требует живого слота и сохранённых offset.
• Батч больше vs меньше. Крупный батч реже пишет в ClickHouse и меньше плодит parts, но увеличивает задержку и объём перечитывания при ошибке.
• Дедупликация vs стоимость чтения. ReplacingMergeTree требует версии и FINAL/argMax на чтении; append-only дешевле на запись, но перекладывает разбор дублей на аналитика.
• Один sink vs несколько. Один консьюмер проще для порядка; несколько параллельнее, но порядок сохраняется только в пределах партиции.

💻 Практика

Все команды — из каталога polygon/. Время — ориентир для одного человека.

1. Базовая линия (10 мин). Поднимите full-профиль, зарегистрируйте коннектор, проверьте данные:

bash
make up-full
make debezium        # ждёт RUNNING и печатает статус
make seed
for i in $(seq 1 60); do curl -s localhost:8088/stats; echo; sleep 5; done

Пока идёт снапшот, ch_orders растёт от нуля, diff сокращается, лаг консьюмера большой. Это норма: дождитесь, пока рост замедлится. Не запускайте check-3 на снапшоте: его рост замаскирует проверку.

2. Лаг консьюмера (5 мин). Посмотрите группу sink-ch:

bash
docker compose exec -T kafka /opt/kafka/bin/kafka-consumer-groups.sh \
  --bootstrap-server localhost:9092 --describe --group sink-ch

В колонке LAG — непрочитанные сообщения по партициям cdc.public.orders. Во время снапшота сумма большая, после догона — около нуля.

3. Лаг слота (5 мин). Сравните текущий WAL с подтверждённым Debezium:

bash
docker compose exec -T postgres-master psql -U app -d orderflow -c \
"SELECT slot_name, active,
        pg_wal_lsn_diff(pg_current_wal_lsn(), confirmed_flush_lsn)::bigint AS lag_bytes
   FROM pg_replication_slots WHERE slot_name = 'orders_cdc';"

Ожидаемо: active = t, лаг — килобайты или единицы мегабайт. Это не лаг консьюмера.

4. Сломать (5 мин). Остановите sink и создайте пять заказов:

bash
make fail-3
make check-3         # ~60 с, ожидаемо CHECK FAIL

Пока check-3 ждёт, повторите --describe: LAG вырос как минимум на пять событий (плюс возможные UPDATE от worker). Запрос из шага 3 покажет, что лаг слота почти не изменился. Объясните почему: Debezium жив и подтверждает WAL, застряло звено после него — sink.

5. Метрики sink (5 мин). Порт :9101 не опубликован на хост, смотрите из контейнера или через Prometheus:

bash
docker compose exec -T sink wget -qO- http://localhost:9101/metrics | grep '^sink_'
curl -s 'http://localhost:9092/api/v1/query?query=sink_events_total'

PromQL: rate(sink_events_total[1m]) — пропускная способность, sink_errors_total — ошибки, sink_batch_size — последний батч.

6. Починить (3 мин). Верните sink, проверьте рост и найдите дубли:

bash
make fix-3
make check-3         # CHECK OK и slot lag в байтах
curl -s 'http://localhost:8124/?database=orderflow' \
  --data-binary "SELECT id, count() AS copies FROM orders_ch
                 GROUP BY id HAVING copies > 1 LIMIT 5"

Sink дочитывает с последнего коммита, ClickHouse догоняет. Строки с copies > 1 — не сбой доставки, а append-only модель и UPDATE от worker.

7. Защита (10 мин). Предложите алерт на лаг группы sink-ch (в стенде нет экспортёра Kafka, сигнал придётся строить на /stats или добавить его), мониторинг pg_replication_slots и max_slot_wal_keep_size, идемпотентную запись по id с версией. Объясните, что закроет каждый пункт.

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

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

• Поднять full-профиль, зарегистрировать коннектор и пережить снапшот на ~200k строк.
• Объяснить три лага и назвать инструмент измерения для каждого.
• Показать, что при остановленном sink растёт лаг группы, а не лаг слота.
• Пройти цикл fail-3 → check-3 (red) → fix-3 → check-3 (green).
• Найти в orders_ch дубли по id и объяснить их происхождение.
• Сформулировать защиту: мониторинг слота и лага, дедупликация по версии.

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

Sink догнал поток, но следующий участок того же пути устроен иначе. Worker читает топик orders группой payments, и его лаг — уже не байты слота, а непрочитанные сообщения при живом API. В модуле 4 остановим worker, увидим рост лага группы и разберём prefetch, пул консьюмеров, ack после обработки и poison-сообщения. Навык прежний: измерить позицию и найти застрявшее звено.