Change data capture: как получать изменения из БД без двойной записи

Прямой ответ

Change data capture (CDC) — техника получения потока изменений (INSERT/UPDATE/DELETE) из базы данных с последующей доставкой потребителям. Ключевое отличие от «двойной записи» в приложении: источником правды остаётся сама БД, поэтому изменения не теряются даже если часть кода забыла отправить событие.

Как это работает в PostgreSQL:

1. База ведёт WAL — журнал предзаписи.
2. Логическое декодирование преобразует записи WAL в логические изменения (строки, а не страницы) с помощью output-плагина (pgoutput, decoderbufs, wal2json).
3. Слот репликации (replication slot) фиксирует, до какого места WAL прочитал потребитель, и не даёт базе удалить нужные сегменты.
4. Коннектор (Debezium через Kafka Connect) читает поток, превращает изменения в события и публикует в топики Kafka.

-- публикация для логической репликации
CREATE PUBLICATION shop_pub FOR TABLE orders, order_items;

-- слот, который использует коннектор
SELECT * FROM pg_replication_slots;

-- проверить задержку и объём удерживаемого WAL
SELECT slot_name, active, restart_lsn,
       pg_size_pretty(pg_wal_lsn_diff(pg_current_wal_lsn(), restart_lsn)) AS retained
FROM pg_replication_slots;

Свойства и ограничения:

• Порядок в пределах строки сохраняется, если ключ партиционирования топика — первичный ключ таблицы. Порядок между разными ключами не гарантируется.
• At-least-once — типичная гарантия: при рестарте коннектор может переотправить часть изменений. Потребители обязаны быть идемпотентными — см. идемпотентность API.
• DDL не входит в поток автоматически. Изменения схемы требуют отдельного процесса (реестр схемы, согласованные деплои).
• Удаления. DELETE приходит как событие с ключом; если нужна «мягкая» история, используйте event sourcing-подход или таблицу аудита.

Практика: надёжная настройка конвейера

1. Слоты и место на диске. Неактивный слот удерживает WAL бесконечно — это самый частый способ заполнить диск и уронить базу. Настройте max_slot_wal_keep_size и алерт на размер удерживаемого WAL; как строить алерты — в статье про Alertmanager.

2. Параметр wal_level. Для логического декодирования нужен wal_level = logical; изменение требует перезапуска.

3. Реплика-источник. Чтение CDC с реплики снижает нагрузку на основную базу, но добавляет задержку; механика репликации описана в статье про read-реплики PostgreSQL.

4. Топики и партиционирование. Один топик на таблицу, партиционирование по первичному ключу. Основы топиков и consumer group — в статье про Kafka.

5. Обработка ошибок. Мёртвые письма и повторная обработка обязательны: отравленное событие не должно останавливать поток. Подходы к гарантиям доставки — в материале про надёжность очередей.

6. Мониторинг конвейера. Минимальный набор метрик: lag коннектора (в секундах и в байтах WAL), возраст последнего события, число ошибок десериализации, состояние слота (active = true).

Снапшот и догон потока. Новый коннектор начинает с согласованного снапшота таблиц и только затем переключается на поток WAL. Это важно при добавлении новой таблицы в публикацию: без снапшота потребитель получит только будущие изменения и потеряет исторические строки. Практическое правило — сначала снапшот, затем поток, и явная отметка в потребителе, где заканчивается один режим и начинается другой; иначе возможны дубли, которые придётся разбирать в данных.

Порядок и параллельность потребителей. Если несколько consumer group обрабатывают один поток, каждая получает полную копию изменений, но внутри группы порядок по ключу сохраняется только при корректном числе партиций. Увеличение партиций топика после запуска меняет распределение ключей и может нарушить ожидаемый порядок в существующих потребителях — такие изменения планируют заранее и проверяют на стенде.

Практический сценарий: поток изменений остановился и заполнил диск

Контекст: PostgreSQL 16 + Debezium + Kafka; потребители обновляют поисковый индекс. Утром база начала отдавать ошибки записи.

1. Симптом. Диск с WAL заполнен на 100 %, запись в базу блокируется. В логах — нехватка места в pg_wal.
2. Причина. Kafka-кластер был недоступен ночью, коннектор остановился, слот репликации остался неактивным и продолжил удерживать WAL.
3. Экстренные действия. Проверяем pg_replication_slots: слот active = false, удерживается сотни гигабайт. Поднимаем Kafka, коннектор возобновляет чтение, WAL начинает освобождаться. Если поднять потребителя нельзя — удаляем слот (pg_drop_replication_slot), принимая потерю позиции, и позже выполняем полный снапшот.
4. Проверка целостности. После восстановления сверяем число обработанных событий с количеством изменений в БД за период; при расхождении запускаем переиндексацию из снапшота, а не «догоняем вручную».
5. Предотвращение. Устанавливаем max_slot_wal_keep_size, алерт на active = false дольше N минут, алерт на рост удерживаемого WAL, отдельный диск под WAL и процедуру действий в runbook.

Результат: запись восстановлена, причина устранена конфигурацией и алертами. Вывод: CDC — мощный механизм, но слот репликации превращает сбой потребителя в сбой базы; ограничивать удержание WAL обязательно.

FAQ

Чем CDC отличается от триггеров и «двойной записи»? Двойная запись (в БД и в брокер в одном коде) не атомарна: при сбое между операциями событие теряется или появляется без изменения. Триггеры пишут в таблицу событий атомарно, но нагружают основную БД и требуют очистки. CDC читает готовый журнал, поэтому не требует изменений в приложении и не пропускает записи.

Какие гарантии доставки даёт CDC? Обычно at-least-once: позиция читателя фиксируется после обработки партии, поэтому при сбое возможны повторы. Exactly-once в общем случае требует идемпотентных потребителей или транзакционных механизмов на стороне записи в Kafka.

Можно ли использовать CDC без Kafka? Да: логическую репликацию PostgreSQL можно читать напрямую (например, через pg_recvlogical или специализированные сервисы), но тогда придётся самостоятельно решать задачи доставки, ретраев, масштабирования потребителей и хранения позиции.

Как обрабатывать изменения схемы таблиц? Изменения DDL не попадают в поток как данные. Практика: совместимые изменения (добавление колонки со значением по умолчанию) — вперёд кода, несовместимые — через двойную запись и переключение, с реестром схемы для валидации. Согласованность версий важна и в API — см. версии API и совместимость.

Источники: PostgreSQL docs (logical decoding), Debezium docs (PostgreSQL connector), Kafka docs (Connect) — ссылки в шапке.