Полная загрузка, инкремент и CDC

Чтобы DWH показывал актуальные данные, нужно доставлять не только новые строки. Старую запись могут исправить или удалить. Есть три распространённых способа узнать о состоянии источника: полный снимок, инкрементальная выборка и CDC. Выбор зависит от объёма, поддержки источника, допустимой задержки и того, нужны ли промежуточные изменения.

Полный снимок

Источник передаёт все строки выбранной области на определённый момент. Для небольшого справочника это простой вариант: можно заменить прежнее текущее состояние или сравнить два снимка. Но полный снимок большой таблицы дорог по сети и обработке. Если между двумя снимками значение меняли дважды, будет видно только конечное состояние.

Отсутствие строки требует осторожности. Если API отдаёт только активные записи, исчезновение может означать закрытие, удаление или изменение фильтра выгрузки. Чтобы трактовать пропажу как удаление, надо знать, что снимок действительно полный для той же области и момента.

Снимок, или snapshot, отвечает «каким был выбранный набор на эту границу». Если сохранять ежедневные снимки, получится история наблюдаемых состояний, но изменения между ними могут потеряться. Это отличается от snapshot табличного формата в озере: там версия метаданных фиксирует состав файлов таблицы, а не обязательно отдельный бизнес-срез на конец дня.

Инкрементальная выборка

Источник отдаёт записи, изменившиеся после предыдущей точки загрузки. Часто для этого используют поле updated_at или возрастающий номер версии. Это сокращает объём чтения, но работает лишь если метка меняется при каждом значимом обновлении. Физическое удаление строки запрос по updated_at не увидит; нужен отдельный признак удаления, журнал или периодическая сверка со снимком.

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

Такую границу часто называют watermark, а сохранённое состояние продолжения — checkpoint. Например, границей может быть пара (updated_at, id), если источник гарантирует её пригодность для последовательной выборки. Зафиксировать новую позицию нужно после успешного применения соответствующей поставки. Если сохранить её до записи результата и упасть, следующий запуск пропустит данные. Одно перекрытие также не спасает, если источник способен изменить старую строку без обновления метки.

Инкрементальный результат можно строить по-разному: дописывать неизменяемые события, обновлять состояние по ключу или заменять затронутые разделы. Выбор зависит от того, могут ли прежние строки изменяться и удаляться. Обновление по ключу должно получать не более одной выбранной версии строки из входа; наличие команды MERGE само не разрешает конфликтующие дубли.

CDC: поток операций изменения

CDC (Change Data Capture, захват изменений данных) передаёт операции над строками: INSERT, UPDATE и DELETE. В PostgreSQL изменения сначала попадают в WAL — журнал предзаписи, нужный самой базе для восстановления и репликации. Логическое декодирование позволяет получить из него последовательность изменений таблиц.

Одна из реализаций — коннектор Debezium для PostgreSQL. Он делает согласованный начальный снимок, а затем читает подтверждённые изменения через механизм логического декодирования и передаёт события в Kafka через Kafka Connect. Kafka не читает WAL самостоятельно: её роль здесь — транспорт и хранение потока событий, а работу с PostgreSQL выполняет коннектор. В другом стеке журнал могут читать и передавать иначе.

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

Начальный снимок даёт состояние при подключении, а не прежнюю историю. Если поле поменяли в прошлом году и старое значение уже исчезло, подключённый сегодня CDC его не восстановит. Для более ранней истории нужен отдельный источник: исторические таблицы, сохранённые события или архивные выгрузки. Срок хранения журнала тоже ограничен; потерю позиции продолжения нельзя исправить обещанием «перечитать всё из Kafka», если нужных событий там уже нет.

CDC позволяет снизить задержку и объём повторного чтения, но не гарантирует готовую историю бизнеса. Получатель должен правильно интерпретировать операции, порядок и границу транзакций. Изменение строки справочника не означает отдельное бизнес-событие; оно лишь говорит о новом состоянии записи.

Повторы, порядок и удаление

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

DELETE может означать удалить текущее состояние из аналитической таблицы, пометить запись неактивной или сохранить исторический факт удаления. Это разные решения. Для персональных данных есть отдельные требования к удалению, которые могут затрагивать и raw-архив. Техническое событие DELETE само не определяет политику хранения.

Дедупликация до записи в целевую таблицу

Идемпотентность означает, что повторное применение одной поставки даёт тот же результат. Для этого нужно определить, что является повтором: одинаковая позиция события, один бизнес-ключ с той же версией или полностью одинаковые поля. Две строки с одинаковым статусом могут быть разными событиями. А у одного объекта с разными статусами могут быть последовательные версии, которые нельзя бездумно удалить.

В staging сначала разбирают вход и находят повторы. Если поступили разные версии одного ключа, выбирают нужную по подтверждённому порядку источника; при конфликте без правила сохраняют исключение. Затем проверенный набор применяют к цели и фиксируют завершение поставки. Простое DISTINCT убирает только совпадения выбранных колонок и не определяет актуальную версию объекта.

Ограничение UNIQUE может защитить ключ в СУБД, которая его обеспечивает. В PostgreSQL обычная вставка с нарушением такого ограничения завершается ошибкой; обработчик решает, откатить загрузку или применить заранее предусмотренное правило конфликта. ON CONFLICT позволяет явно выбрать действие, но не исправляет неизвестную причину дублей. В ClickHouse ключ MergeTree не даёт такой гарантии уникальности.

Если источник пока не исправить, загрузку отделяют через staging или подготовленное представление с правилом дедупликации. Периодическая очистка уже опубликованной таблицы оставляет промежуток, когда потребитель видит повторы. Поэтому частота очистки зависит от допустимой задержки и доступа к таблице; надёжнее проверять каждую поставку до публикации и одновременно искать причину повторного поступления.

Контроль качества и карантин

Проверки привязывают к контракту набора. Для полноты сверяют все ожидаемые разделы и источники. Кроме типа числа проверяют единицу: сумма в копейках, прочитанная как рубли, остаётся синтаксически корректным числом, но искажает отчёт.

Карантин, или quarantine, — отдельное место для записей, которые не проходят проверку и требуют решения. В нём сохраняют связь с исходной поставкой, причину ошибки, время обнаружения и состояние обработки. Например, запись с неизвестной единицей измерения не стоит превращать в ноль. После исправления запись можно обработать повторно по тому же устойчивому ключу. Доступ и срок хранения исключений задают так же осознанно, как для raw.

В отчёте о качестве полезны набор и период, идентификатор запуска, правило и его версия, ожидаемое и фактическое значение, число или доля ошибок и принятое действие. Например, из 10 000 строк 9 970 принято, 20 оказались подтверждёнными повторами, 10 помещено в карантин: все входные строки объяснены. Порог не выбирают произвольно — допустимость десяти ошибок зависит от того, какие записи потеряны и как это влияет на показатель.

Критическая ошибка, например неполная поставка обязательного источника, может остановить публикацию и сохранить прежнюю проверенную версию. Для допустимых исключений можно публиковать результат с указанным покрытием и уведомить ответственного. Метрики нужно наблюдать по запускам: резкий рост доли карантина или отставания заметен даже при формально успешном завершении задачи. Проверки без заранее определённой реакции только регистрируют ошибку.

Получить изменения и сохранить историю — не одно и то же

Если применять каждый UPDATE поверх прежней строки, получится только текущее состояние, хотя данные доставлялись через CDC. Чтобы ответить «какое значение действовало в прошлом месяце?», нужно хранить версии с датами действия или события, из которых их можно восстановить. И наоборот: историю можно получить сравнением полных снимков, если промежуточные изменения не нужны.

Три полезные даты — время бизнес-события, время изменения записи в источнике и время поступления в DWH. Они могут различаться. Поздно пришедшее событие часто требует пересчёта прошлого периода; при этом факт поздней доставки тоже стоит сохранить для разбора.

При проверке загрузки сравнивают не только число строк. Обновление суммы не меняет количество, но меняет отчёт. Нужны позиции источника, число операций по типам, ошибки применения и контрольные показатели затронутых наборов.

Попробуйте на данных

В учебной базе stg_orders хранит несколько состояний одного заказа. Снимок текущей таблицы orders показывает последнее состояние, но не все переходы. Для одного ключа их можно посмотреть отдельно:

SELECT order_id, status, updated_at, load_date
FROM stg_orders
WHERE order_id = 42
ORDER BY updated_at, loaded_at;

Задания сравнивают снимок и версии, выделяют инкремент по дате загрузки и находят повторно доставленные изменения. Получение изменений само по себе не восстанавливает историю, которая была утрачена до начала наблюдения.