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

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 — включение-выключение виртуальных меток времени.
• RETENTION_PERIOD — время хранения записей. Тип значения — Interval, значение по умолчанию — 24 часа (Interval('PT24H')).
• TOPIC_MIN_ACTIVE_PARTITIONS — количество партиций топика. По умолчанию количество партиций топика равно количеству партиций таблицы.
• 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
);

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

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

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

ALTER TABLE `series` DROP CHANGEFEED `updates_feed`;