Добавление или удаление потока изменений

ADD CHANGEFEED <name> WITH (option = value[, ...]) — добавляет поток изменений (changefeed) с указанным именем и параметрами.

Параметры потока изменений

• MODE — режим работы. Указывает, что именно будет записано в поток при каждом изменении данных в таблице:
  • KEYS_ONLY — будут записаны только компоненты первичного ключа и признак изменения.
  • UPDATES — будут записаны значения изменившихся столбцов, получившиеся в результате изменения.
  • NEW_IMAGE — будут записаны значения всех столбцов, получившиеся в результате изменения.
  • OLD_IMAGE — будут записаны значения всех столбцов, предшествующие изменению.
  • NEW_AND_OLD_IMAGES - комбинация режимов NEW_IMAGE и OLD_IMAGE. Будут записаны значения всех столбцов до и в результате изменения.
• FORMAT — формат данных, в котором будут записаны данные:

  • JSON — записывать данные в формате JSON.

  • DEBEZIUM_JSON — записывать данные в JSON-формате, аналогичном Debezium формату.

• VIRTUAL_TIMESTAMPS — включение-выключение виртуальных меток времени.
• BARRIERS_INTERVAL — периодичность выгрузки барьеров. Тип значения — Interval. По умолчанию выключено.
• RETENTION_PERIOD — время хранения записей. Тип значения — Interval, значение по умолчанию — 24 часа (Interval('PT24H')).
• TOPIC_AUTO_PARTITIONING — режим автопартиционирования топика:
  • ENABLED — для потока изменений будет создан автопартиционированный топик. Количество партиций в таком топике увеличивается автоматически по мере роста скорости обновления таблицы. Параметры автопартиционирования топика можно настроить.
  • DISABLED — для потока изменений будет создан топик без автопартиционирования. Это значение по умолчанию.
• TOPIC_MIN_ACTIVE_PARTITIONS — количество партиций топика. По умолчанию количество партиций топика равно количеству партиций таблицы. Для автопартиционированных топиков количество партиций увеличивается по мере роста скорости обновления таблицы. Если при создании ченджфида опция TOPIC_AUTO_PARTITIONING была отключена (DISABLED), то число партиций в топике, связанном с таким ченджфидом, впоследствии изменить нельзя.
• INITIAL_SCAN — включение-выключение первоначального сканирования таблицы. По умолчанию выключено.

Приведенный ниже код добавит поток изменений с именем updates_feed, в который будут выгружаться значения изменившихся столбцов таблицы в формате JSON:

ALTER TABLE `series` ADD CHANGEFEED `updates_feed` WITH (
    FORMAT = 'JSON',
    MODE = 'UPDATES'
);

Записи в таком потоке изменений будут храниться в течение 24 часов (значение по умолчанию). Код из следующего примера создаст поток изменений с хранением записей в течение 12 часов:

ALTER TABLE `series` ADD CHANGEFEED `updates_feed` WITH (
    FORMAT = 'JSON',
    MODE = 'UPDATES',
    RETENTION_PERIOD = Interval('PT12H')
);

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

ALTER TABLE `series` ADD CHANGEFEED `updates_feed` WITH (
    FORMAT = 'JSON',
    MODE = 'UPDATES',
    VIRTUAL_TIMESTAMPS = TRUE
);

Пример создания потока изменений с виртуальными метками времени и барьерами раз в 10 секунд:

ALTER TABLE `series` ADD CHANGEFEED `updates_feed` WITH (
    FORMAT = 'JSON',
    MODE = 'UPDATES',
    VIRTUAL_TIMESTAMPS = TRUE,
    BARRIERS_INTERVAL = Interval('PT10S')
);

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

ALTER TABLE `series` ADD CHANGEFEED `updates_feed` WITH (
    FORMAT = 'JSON',
    MODE = 'UPDATES',
    INITIAL_SCAN = TRUE
);

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

ALTER TABLE `series` ADD CHANGEFEED `updates_feed` WITH (
    FORMAT = 'JSON',
    MODE = 'UPDATES',
    TOPIC_AUTO_PARTITIONING = 'ENABLED',
    TOPIC_MIN_ACTIVE_PARTITIONS = 2
);

DROP CHANGEFEED — удаляет поток изменений с указанным именем. Приведенный ниже код удалит changefeed с именем updates_feed:

ALTER TABLE `series` DROP CHANGEFEED `updates_feed`;