Блог

Главная - Вопросы по программированию - Включение новых таблиц с данными в существующий соединитель Debezium

Включение новых таблиц с данными в существующий соединитель Debezium

#mysql #apache-kafka #apache-kafka-streams #debezium

#mysql #apache-kafka #apache-kafka-streams #debezium

Вопрос:

Цель:

Необходимо добавить новую таблицу в существующий соединитель Debezium MySQL версии 1.1.1.Final. Данные из таблицы должны быть заполнены в теме после перезапуска соединителя с обновленными конфигурациями (новые таблицы в таблице.белый список).

Проблема:

Не существует четкой стратегии добавления новой таблицы (таблиц), которая будет добавлена в Kafka с ее данными после того, как был сделан первоначальный снимок (предыдущая версия соединителя работает). Мы ищем что-то похожее на snapshot.select.statement.overrides только для рабочего соединителя. snapshot.select.statement.overrides не работает, когда снимок уже создан и соединитель перезапущен с новыми таблицами.

Возможное решение:

Создайте и запустите новый соединитель для таблиц, которые должны быть инициализированы данными (snapshot.mode=initial). Недостаток: каждый выпуск может включать новый соединитель.

или

  1. Создать соединитель миграции (v2) для таблиц (с конфигурациями snapshot.mode=initial и new database.history.kafka.topic)
  2. Удалить соединитель v2 после завершения моментальной съемки
  3. Удалить database.history.kafka.topic для предыдущего соединителя (v1)
  4. Обновите соединитель (v1) с таблицами из соединителя (v2) — поместите их в таблицу.белый список
  5. Измените snapshot.mode на schema_only_recovery и запустите соединитель (v1)

Вопрос:

Какой способ является официальным? Может быть, я что-то упустил, и есть более простой способ, чем второй?

UPD: согласно комментарию @Matar и некоторому расследованию, похоже, нет необходимости удалять database.history.kafka.topic, потому что в конфигурации по умолчанию он все равно собирает весь DDL из БД.

Комментарии:

1. Я работаю над чем-то похожим на вашу задачу (MSSQL). Я пошел с решением (2). Мне просто интересно, зачем вам создавать новую database.history.kafka.topic? не могу ли я просто использовать этот из V1, поскольку все таблицы принадлежат одной и той же БД.

2. @matar Я думаю, вы правы. Нет необходимости заново создавать тему, особенно если вы используете свой соединитель, который прослушивает все изменения в БД. Мне интересно, что Debezium еще не поддерживает сценарий обновления спустя годы.

Ответ №1:

С последней версией Debezium Server вы можете добавить следующее значение конфигурации

 debezium.snapshot.new.tables=parallel

В случае, если вы используете Debezium, вы можете попробовать это значение конфигурации

 snapshot.new.tables=parallel

Примечание: сервер Debeziyum поддерживает Kinesis, Google Pub sub и Apache Pulsar. Я использую это, и его конфигурация немного отличается. Мне пришлось добавлять «debezium» перед каждым элементом

Как только эта конфигурация будет добавлена, любое добавление в tables.whitelist Для этих дополнительных таблиц Debezium создаст моментальные снимки.