Удаление в CDC: почему последняя версия строки может не существовать

Поток изменений базы уже подключён к витрине. Создания и обновления обрабатываются одинаково: берём последнюю строку по ключу. Через неделю выясняется, что удалённый клиент всё ещё присутствует в отчёте. Его последнее событие не содержало нового набора полей, поэтому загрузчик его пропустил. Так приём «оставить последнюю версию» превратился в сохранение последней существовавшей версии.

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

Различим удаление и tombstone

У коннектора Debezium для PostgreSQL событие удаления содержит op=d и after=null. Дополнительно может отправляться tombstone: сообщение с тем же ключом и null в значении, используемое при compaction Kafka. Это разные сообщения с разными назначениями. Настройки коннектора и преобразований влияют на то, что дойдёт до потребителя; до написания обработчика посмотрите реальную запись вашего потока.

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

Удалим объект и не дадим старому обновлению воскресить его

Учебный поток ниже уже приведён к общей форме с op, key, version и data. Версия монотонна для одного ключа; в реальном источнике надо выбрать подходящую позицию журнала и правило сравнения. Мы оставляем последнюю принятую версию даже после удаления. Событие версии 2, пришедшее после версии 3, отклоняется.

events = [
    {"key": "A", "version": 1, "op": "upsert", "data": {"name": "Анна"}},
    {"key": "A", "version": 3, "op": "delete", "data": None},
    {"key": "A", "version": 2, "op": "upsert", "data": {"name": "Анна М."}},
    {"key": "B", "version": 1, "op": "upsert", "data": {"name": "Борис"}},
]
state, last = {}, {}
for event in events:
    key, version = event["key"], event["version"]
    if version <= last.get(key, -1):
        continue
    if event["op"] == "delete":
        state.pop(key, None)
    elif event["op"] == "upsert":
        state[key] = event["data"]
    else:
        raise ValueError("unknown operation")
    last[key] = version
print("state", state, "versions", last)
assert "A" not in state and last["A"] == 3
assert set(state) == {"B"}

Анны в текущем состоянии нет, но версия 3 сохранена. Это защищает от старого обновления в пределах принятой модели версий. Если удалить и state, и last, позднее сообщение получит вид нового ключа и восстановит уже удалённый объект. Срок хранения сведений об удалении должен учитывать доступный горизонт повторного чтения и восстановления.

Как применить правило в витрине

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

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

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

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

В курсе вы работаете с Kafka, ключами и схемами сообщений, потоковой обработкой Spark и хранением в lakehouse. Следующий шаг — перенести разобранное правило в end-to-end конвейер и проверить, что обновления и удаления сохраняют согласованное состояние таблицы после повторного запуска. Откройте программу платного курса «Data Engineering Fast‑Track: Kafka → Airflow → Spark».