--- metadata: - name: generator content: Diplodoc Platform v5.50.4 alternate: - https://ydb.tech/docs/en/concepts/cdc.md - https://ydb.tech/docs/ru/concepts/cdc.md sourcePath: ru/core/concepts/cdc.md --- > **Documentation Index:** Fetch the complete configuration index at https://ydb.tech/docs/ru/llms.txt # Change Data Capture (CDC) {% note warning %} Поддерживается только для [строковых](https://ydb.tech/docs/ru/concepts/datamodel/table.md#row-oriented-tables) таблиц. Поддержка функциональности для [колоночных](https://ydb.tech/docs/ru/concepts/datamodel/table.md#column-oriented-tables) таблиц находится в разработке. {% endnote %} Change Data Capture (CDC) обеспечивает захват изменений строк строковой таблицы YDB, формирует из них *поток изменений (changefeed)*, записывает в распределенное хранилище и предоставляет доступ к этим записям для дальнейшей обработки. В качестве распределенного хранилища используется [топик](https://ydb.tech/docs/ru/concepts/datamodel/topic.md), который позволяет эффективно хранить лог изменений таблицы. Когда в таблице добавляется, обновляется или удаляется строка, CDC формирует запись о произошедшем изменении с указанием [первичного ключа](https://ydb.tech/docs/ru/concepts/datamodel/table.md) строки и пишет ее в соответствующую данному ключу партицию топика. ## Гарантии {#guarantees} * Записи об изменениях шардированы между партициями топика по первичному ключу. * Каждое изменение доставляется ровно один раз (exactly-once семантика). * Изменения по одному и тому же первичному ключу доставляются в том же порядке, в котором они происходили в таблице в одну и ту же партицию топика. * Запись об изменении доставляется до партиции топика только после фиксации (коммита) соответствующей транзакции в таблице. ## Ограничения {#restrictions} * В потоке изменений поддерживаются записи о следующих видах операций: * Обновление — перезапись значений указанных столбцов. Пример запроса: [UPDATE](https://ydb.tech/docs/ru/yql/reference/syntax/update.md). * Замена — перезапись значений указанных столбцов, значения неуказанных столбцов заменяются на значения по умолчанию. Пример запроса: [REPLACE INTO](https://ydb.tech/docs/ru/yql/reference/syntax/replace_into.md). * Удаление. Пример запроса: [DELETE FROM](https://ydb.tech/docs/ru/yql/reference/syntax/delete.md). Добавление строки является частным случаем обновления или замены, и в потоке изменений запись о добавлении строки будет выглядеть аналогично записи об обновлении или замене, в зависимости от исходного запроса, приведшего к изменению. ## Виртуальные метки времени {#virtual-timestamps} Все изменения в таблицах YDB упорядочены в соответствии с порядком выполнения транзакций. Каждое изменение маркируется виртуальной меткой времени, являющейся кортежем из двух элементов: 1. Глобального времени координатора. 1. Уникального идентификатора транзакции. Используя эти метки, можно упорядочить записи из разных партиций топика относительного друг друга или использовать их для фильтрации (например, чтобы исключить записи о старых изменениях). {% note info %} По умолчанию виртуальные метки времени не выгружаются в поток изменений. Для их включения используйте [соответствующий параметр](https://ydb.tech/docs/ru/yql/reference/syntax/alter_table/changefeed.md) при создании потока. {% endnote %} ## Барьеры {#barriers} Барьеры — служебные записи без данных об изменении или удалении с [виртуальными метками времени](#virtual-timestamps), которые появляются в каждой партиции топика с заданной периодичностью. Барьер — это гарантия, что любое изменение с виртуальной меткой времени меньшей, чем у барьера, было записано в данную партицию топика. Барьеры могут быть использованы для обеспечения строгой упорядоченности и глобальной согласованности данных путём буферизации данных между ними. {% note info %} По умолчанию барьеры не выгружаются в поток изменений. Для настройки периодичности выгрузки барьеров используйте [соответствующий параметр](https://ydb.tech/docs/ru/yql/reference/syntax/alter_table/changefeed.md) при создании потока. {% endnote %} ## Первоначальное сканирование таблицы {#initial-scan} По умолчанию в поток изменений выгружаются записи только о тех строках таблицы, которые изменились после создания потока. Первоначальное сканирование таблицы позволяет выгрузить в поток изменений значения всех строк, существовавших на момент его создания. Сканирование производится в фоновом режиме поверх снапшота таблицы. При этом возможны следующие ситуации: * В таблице меняется значение непросканированной строки. В поток изменений последовательно будут выгружены: запись с исходным значением и запись об изменении. При повторном изменении этой же строки будет выгружена только запись об изменении. * Во время сканирования обнаруживается измененная строка. В поток изменений ничего не выгружается, так как исходное значение уже было выгружено в момент изменения (см. предыдущий пункт). * В таблице меняется значение просканированной строки. В поток изменений выгружается только запись об изменении. Таким образом, гарантируется, что для одной и той же строки (первичного ключа) сначала будет выгружено исходное значение, а затем — запись об изменении. {% note info %} Запись с исходным значением строки будет помечена как запись об [обновлении](#restrictions). При использовании [виртуальных меток времени](#virtual-timestamps) записи маркируются меткой времени снапшота. {% endnote %} В процессе сканирования, в зависимости от частоты обновления данных таблицы, возможен повышенный фон ошибок `OVERLOADED` из-за того, что, помимо записей об изменениях, необходимо доставить также записи с исходными значениями строк. По окончании сканирования поток изменений переходит в нормальный режим работы. {% note warning %} На время первоначального сканирования в таблице приостанавливаются процессы [автоматического партиционирования](https://ydb.tech/docs/ru/concepts/datamodel/table.md#partitioning_row_table) и в топик не выгружаются [барьеры](#barriers). {% endnote %} ## Структура записи {#record-structure} В зависимости от [параметров потока](https://ydb.tech/docs/ru/yql/reference/syntax/alter_table/changefeed.md) структура записи может отличаться. ### JSON-формат {#json-record-structure} Запись в формате [JSON](https://en.wikipedia.org/wiki/JSON) имеет следующую структуру: ```json { "key": [], "update": {}, "reset": {}, "erase": {}, "newImage": {}, "oldImage": {}, "ts": [, ] } ``` * `key` — массив значений компонент первичного ключа. Присутствует всегда. Порядок элементов соответствует порядку столбцов в первичном ключе таблицы. * `update` — признак обновления. Присутствует, если запись соответствует операции обновления. В режиме `UPDATES` так же содержит названия и значения изменившихся столбцов. * `reset` — признак замены. Присутствует, если запись соответствует операции замены. В режиме `UPDATES` так же содержит названия и значения столбцов, для которых задано значение. * `erase` — признак удаления. Присутствует, если запись соответствует операции удаления. * `newImage` — снимок состояния строки, получившегося в результате изменения. Присутствует в режимах `NEW_IMAGE` и `NEW_AND_OLD_IMAGES`. Содержит названия и значения столбцов. * `oldImage` — снимок состояния строки, предшествовавшего изменению. Присутствует в режимах `OLD_IMAGE` и `NEW_AND_OLD_IMAGES`. Содержит названия и значения столбцов. * `ts` — [виртуальная метка времени](#virtual-timestamps). Присутствует, если включена настройка `VIRTUAL_TIMESTAMPS`. Содержит значение глобального времени координатора (`step`) и уникальный идентификатор транзакции (`txId`). Например, запись об обновлении в режиме `UPDATES`: ```json { "key": [1, "one"], "update": { "payload": "lorem ipsum", "date": "2022-02-22" } } ``` Запись об удалении: ```json { "key": [2, "two"], "erase": {} } ``` Запись со снимками строки: ```json { "key": [1, 2, 3], "update": {}, "newImage": { "textColumn": "value1", "intColumn": 101, "boolColumn": true }, "oldImage": { "textColumn": null, "intColumn": 100, "boolColumn": false } } ``` Запись с виртуальными метками времени: ```json { "key": [1], "update": { "created": "2022-12-12T00:00:00.000000Z", "customer": "Name123" }, "ts": [1670792400890, 562949953607163] } ``` Запись с барьером содержит единственное поле `resolved` с виртуальной меткой времени: ```json { "resolved": [1670792500000, 0] } ``` {% note info %} * Одна и та же запись не может содержать поля `update`, `reset` и `erase` одновременно, так как эти поля являются признаками операции (невозможно одновременно обновить и удалить строку таблицы). Но каждая запись содержит одно из этих полей (любая операция является обновлением, заменой или удалением). * В режиме `UPDATES` для операций обновления или замены поля `update` и `reset` выполняют роль не только признака операции, но и содержат названия и значения изменившихся столбцов. * Поля JSON-объекта, содержащие названия и значения столбцов (`newImage`, `oldImage`, `update` и `reset` в режиме `UPDATES`), *не включают* в себя столбцы, являющиеся компонентами первичного ключа. * Если в записи присутствует поле `erase` (то есть запись соответствует операции удаления), то это всегда пустой JSON-объект (`{}`). {% endnote %} ### JSON-формат, совместимый с Debezium {#debezium-json-record-structure} Запись в формате [JSON](https://en.wikipedia.org/wiki/JSON), совместимого с [Debezium](https://debezium.io), имеет следующую структуру: ```json { "payload": { "op": , "before": {}, "after": {}, "source": { "connector": , "version": , "ts_ms": , "step": , "txId": , "snapshot": } } } ``` * `op` — операция, которая была произведена над строкой в таблице: * `c` — вставка. Допустимо только в режиме `NEW_AND_OLD_IMAGES`. * `u` — обновление. * `d` — удаление. * `r` — чтение из [снапшота](#initial-scan). * `before` — снимок состояния строки, предшествовавшего изменению. Присутствует в режимах `OLD_IMAGE` и `NEW_AND_OLD_IMAGES`. Содержит названия и значения столбцов. * `after` — снимок состояния строки, получившегося в результате изменения. Присутствует в режимах `NEW_IMAGE` и `NEW_AND_OLD_IMAGES`. Содержит названия и значения столбцов. * `source` — метаданные записи. * `connector` — название коннектора. Текущее название: `ydb`. * `version` — версия коннектора, используемая для генерации записи. Текущая версия: `1.0.0`. * `ts_ms` — примерное время применения изменения в YDB, в миллисекундах. * `step` — глобальное время координатора. Компонент [виртуальных меток времени](#virtual-timestamps). * `txId` — уникальный идентификатор транзакции. Компонент [виртуальных меток времени](#virtual-timestamps). * `snapshot` — признак чтения из снапшота. При чтении с использованием [Kafka API](https://ydb.tech/docs/ru/reference/kafka-api/index.md) в качестве ключа сообщения указывается Debezium-совместимый первичный ключ измененной строки: ```json { "payload": {} } ``` * `payload` — первичный ключ строки, которая была изменена. Содержит названия и значения столбцов, являющихся компонентами первичного ключа. ## Время хранения записей {#retention-period} По умолчанию записи хранятся в потоке изменений в течение 24 часов с момента отправки. В зависимости от сценариев использования время хранения можно уменьшить или увеличить до 30 дней. {% note warning %} Записи, время хранения которых истекло, удаляются вне зависимости от того, успели их обработать (прочитать) или нет. {% endnote %} Удаление записей до их обработки клиентом приводит к возникновению пропусков [офсетов](https://ydb.tech/docs/ru/concepts/datamodel/topic.md#offset), то есть офсеты последней прочитанной из партиции записи и самой ранней из доступных будут отличаться более, чем на единицу. Для настройки времени хранения записей укажите параметр [RETENTION_PERIOD](https://ydb.tech/docs/ru/yql/reference/syntax/alter_table/changefeed.md) при создании потока изменений. Данные хранятся в течение всего указанного периода независимо от того, были ли они прочитаны потребителями. ## Количество партиций топика {#topic-partitions} По умолчанию количество [партиций топика](https://ydb.tech/docs/ru/concepts/datamodel/topic.md#partitioning) равно количеству партиций таблицы. Количество партиций топика можно переопределить, указав параметр [TOPIC_MIN_ACTIVE_PARTITIONS](https://ydb.tech/docs/ru/yql/reference/syntax/alter_table/changefeed.md) при создании потока изменений. Кроме того, можно создать поток изменений, в котором количество партиций будет увеличиваться автоматически, задав параметр [TOPIC_AUTO_PARTITIONING](https://ydb.tech/docs/ru/yql/reference/syntax/alter_table/changefeed.md) при создании потока изменений. {% note info %} В настоящий момент возможность явного указания числа партиций топика доступна только для таблиц, у которых первый компонент первичного ключа имеет тип `Uint64` или `Uint32`. {% endnote %} ## Создание и удаление потока изменений {#ddl} Поток изменений может быть добавлен к существующей таблице или удален директивами [ADD CHANGEFEED и DROP CHANGEFEED](https://ydb.tech/docs/ru/yql/reference/syntax/alter_table/changefeed.md) операции YQL `ALTER TABLE`. При удалении таблицы добавленный к ней поток изменений также будет удален. ## Получение и изменение параметров топика {#topic-options} Для получения параметров топика можно воспользоваться [SDK](https://ydb.tech/docs/ru/reference/ydb-sdk/topic.md#describe-topic) или [YDB CLI](https://ydb.tech/docs/ru/reference/ydb-cli/commands/scheme-describe.md), передав в аргументах путь до потока изменений, который формируется следующим образом: ```txt путь/до/строковой_таблицы/имя_потока_изменений ``` >Например, если у строковой таблицы `table` в директории `my` есть поток изменений с именем `updates_feed`, то путь к нему будет выглядеть так: > >```text >my/table/updates_feed >``` Параметры топика могут быть изменены с использованием выражения [ALTER TOPIC](https://ydb.tech/docs/ru/yql/reference/syntax/alter-topic.md). Поддерживаемые действия: * [изменение параметров](https://ydb.tech/docs/ru/yql/reference/syntax/alter-topic.md#alter-topic): * `retention_period`; * `retention_storage_mb`; * [управление читателями](https://ydb.tech/docs/ru/yql/reference/syntax/alter-topic.md#consumer). ## Назначение и применение CDC {#best_practices} Об использовании CDC при разработке приложений смотрите в [рекомендациях](https://ydb.tech/docs/ru/dev/cdc.md).