Как обмениваться данными между разными системами? CDC-коннектор в РЕД Базе Данных
Привет! Меня зовут Василий, и я разработчик СУБД РЕД База Данных в компании РЕД СОФТ. Сегодня я расскажу про передачу данных из СУБД РЕД База Данных в платформу для обработки потоковых данных Apache Kafka с помощью специального плагина. В этой статье: про создание плагина, его отличия от аналогичных решений для других СУБД, репликацию и возможные кейсы использования.
В продакшене CDC обеспечивает аналитику данными практически в реальном времени и позволяет синхронизировать объекты с минимальной задержкой, например кэши. Технология дает возможность объединять данные из различных источников без дополнительной трансформации, а также проводить миграцию без даунтайма. Кроме того, CDC помогает решать задачи аудита инцидентов и поддерживать согласованность сервисов.
Но для начала немного вводной части. Сегодня данные – основа любого бизнеса. Но они очень редко хранятся в одном месте. Данные обрабатываются в OLTP-системах, анализируются в OLAP, поисковых движках, кэшах, потоках микросервисов.
Основная сложность распределения данных – передать их из одной системы в другую. Обычно такую проблему решают с помощью выгрузки данных в определенный формат и загружают их в целевую систему. Но этот подход имеет серьезный недостаток – данные быстро устаревают. К тому же сохранение большого объема данных требует дополнительного места на носителях информации, что увеличивает финансовые затраты.
Change Data Capture (Захват измененных данных) позволяет решить эту проблему. Это не просто технология, а архитектурная модель, описывающая подход, при котором в базе данных отслеживаются изменения (добавление, изменение и удаление записей) и оперативно передаются в другие системы, обеспечивая надежность и доступность данных. Простыми словами, CDC перехватывает только измененные строки, а не всю базу целиком.
Источником информации является база данных, CDC выступает в качестве отправителя данных, а для аккумуляции и дальнейшей передачи требуется посредник. Таким посредником является Apache Kafka – это платформа с открытым исходным кодом, разработанная для обработки потоков данных в реальном времени. Также Kafka называют брокером сообщений. Взаимодействие между СУБД и целевыми системами, использующими технологию CDC представлено на схеме ниже:
Пользователь вносит изменения в таблицы базы данных, изменения отлавливаются специальным CDC-коннектором, коннектор передает изменения в брокер, брокер обрабатывает данные для дальнейшего хранения и передачи, а целевые системы забирают необходимые данные из брокера.
Работа CDC с Kafka и коннекторами. Особенность CDC-коннектора для РЕД Базы Данных
Как мы выяснили выше, Kafka выступает в роли посредника в передаче данных между CDC-коннекторами и целевыми системами. Kafka берет на себя три задачи:
Сохранность событий и защита от их потери.
Корректный порядок событий в рамках одной партиции или топика, который выступает аналогом таблицы в СУБД.
Параллельное получение изменений несколькими потребителями, не мешающими друг другу.
Связующим звеном между базой данных и Kafka выступают CDC-коннекторы. Это специальные компоненты, которые умеют отлавливать изменения конкретной СУБД, превращать эти изменения в специально структурированные сообщения и отправлять их в Kafka. Обычно коннекторы реализуются на основе фреймворка Kafka Connect, позволяющему подключаться к журналу транзакций базы данных (WAL или Binlog) для просмотра изменений, не создавая дополнительной нагрузки на таблицы запросами или триггерами.
Но вот в чем особенность СУБД РЕД База Данных – в ней отсутствует журнал транзакций в типичном понимании других СУБД. РЕД База Данных организует ввод-вывод таким образом, чтобы данные на диске всегда поддерживались в согласованном состоянии. Это безопасный вариант, который минимизирует риск потери данных. Но как тогда получить изменения записей «на лету»? Ответ простой – репликация! Начиная с РЕД Базы Данных 3 в движке появился механизм встроенной репликации данных. А благодаря интерфейсам репликации возможна реализация плагинов, позволяющих получать изменения таблиц не создавая нагрузку на сервер!
Как это работает представлено на схеме ниже:
Если в какой-либо реплицируемой таблице происходит изменение, внутренний механизм репликации запускает репликационную сессию. Данная сессия будет создавать репликационные транзакции, из которых можно получить изменения в таблицах, и получать изменения генераторов. Транзакции в свою очередь предоставляют методы событий, происходящих для конкретной таблицы – вставка, изменение и удаление. Кроме того, репликационные транзакции могут отлавливать DDL – изменения, меняющие структуру базы данных.
Как работает CDC-коннектор РЕД Базы Данных
Как устроен коннектор Kafka в СУБД РЕД База Данных? Отсутствие механизма журналирования транзакций делает его нетипичным по сравнению с существующими аналогами.
Коннектор написан на Java и использует библиотеки Debezium. Debezium – это платформа с открытым исходным кодом (распространяется под лицензией Apache 2.0), которая решает задачу Change Data Capture, превращает изменения в базах данных в потоки событий Kafka в реальном времени. Debezium предоставляет набор интерфейсов для взаимодействия с СУБД, позволяет управлять жизненным циклом коннектора, обрабатывать события изменений, формировать записи для последующей отправки брокеру Kafka.
Прежде чем сформировать записи для Kafka, необходимо получить данные. Для доступа к данным коннектор реализует следующие интерфейсы объектно-ориентированного API:
Интерфейс репликационной сессии:
public static interface IReplicatedSessionIntf extends IPluginBaseIntf
{
public int VERSION = 4;
public boolean init(IStatus status, IAttachment attachment);
public IReplicatedTransaction startTransaction(IStatus status, ITransaction transaction, long number);
public void cleanupTransaction(IStatus status, long number);
public void setSequence(IStatus status, String name, long value);
}Интерфейс репликационной транзакции:
public static interface IReplicatedTransactionIntf extends IDisposableIntf
{
public int VERSION = 3;
public void prepare(IStatus status);
public void commit(IStatus status);
public void rollback(IStatus status);
public void startSavepoint(IStatus status);
public void releaseSavepoint(IStatus status);
public void rollbackSavepoint(IStatus status);
public void insertRecord(IStatus status, String name, IReplicatedRecord record);
public void updateRecord(IStatus status, String name, IReplicatedRecord orgRecord, IReplicatedRecord newRecord);
public void deleteRecord(IStatus status, String name, IReplicatedRecord record);
public void executeSql(IStatus status, String sql);
public void executeSqlIntl(IStatus status, int charset, String sql);
}Два вышеперечисленных интерфейса позволяют получить информацию о подключении, базе данных, транзакции, таблицах, изменяемых записях, значениях полей и их метаданных.
Через интерфейс IReplicatedSessionIntf плагин получает информацию об имени сервера и базе данных, к которой выполнено подключение, точнее GUID базы и ее полный путь. Также этот интерфейс позволяет получить измененное значение генератора в этой базе данных. Но более важная задача этого интерфейса – создавать экземпляр репликационной транзакции IReplicatedTransactionIntf, которая будет получать доступ к данным, обрабатываемым основной транзакцией на сервере.
При создании экземпляр репликационной транзакции получает объект и номер транзакции на сервере. Номер транзакции пригодится при создании сообщения, отправляемого в Kafka, а через объект транзакции можно получить метаданные таблиц. После создания репликационная транзакция готова к «прослушиванию» изменений, происходящих в транзакции на сервере. «Прослушивание» изменений в таблицах осуществляется методами insertRecord, updateRecord и deleteRecord. Изменения DDL коннектор отслеживает методами executeSql и executeSqlIntl. Они получают на вход имя таблицы, в которой произошли изменения, и интерфейс для доступа к измененной записи (в случае updateRecord - записям) IReplicatedRecord.
IReplicatedRecord позволяет получить информацию количестве полей в изменяемой записи и извлечь данные и метаданные для каждого поля – имя, тип, подтип, точность, длину и кодировку.
После извлечения всех данных и метаданных коннектор РЕД Базы Данных формирует события, которые затем будут преобразованы в сообщения для Kafka.
События изменения данных
Каждое событие изменения данных, которое формирует коннектор РЕД Базы Данных, имеет ключ и значение. Структура ключа и значения зависит от таблицы, из которой исходят события изменения. Коннектор разработан с учетом непрерывных потоков сообщений о событиях. Каждый ключ и значение сообщения состоит из двух частей: схемы и полезной нагрузки. Схема описывает структуру полезной нагрузки, а полезная нагрузка содержит фактические данные. Для каждой измененной таблицы, ключ события изменения формируется из полей в первичном ключе (или уникальном ключевом ограничении) таблицы на момент создания события.
Как и ключ сообщения, значение сообщения о событии изменения имеет раздел схемы и раздел полезной нагрузки. Раздел полезной нагрузки каждого значения события изменения, создаваемого коннектором, имеет структуру конверта со следующими полями:
N | Значение | Описание |
1 | op | Обязательное поле, содержащее строковое значение, описывающее тип операции: CREATE - создание (или вставка), UPDATE - обновление, DELETE - удаление. |
2 | before | Необязательное поле, которое, если присутствует, содержит состояние строки до наступления события. |
3 | after | Необязательное поле, которое, если присутствует, содержит состояние строки после произошедшего события. Структура описывается той же схемой, которая использовалась ранее. |
4 | source | Обязательное поле, содержащее структуру, описывающую исходные метаданные для события, которая содержит такие поля: версия коннектора, имя коннектора, имя сервера, номер транзакции, время (по системным часам в JVM), в которое коннектор обработал событие, имя базы данных, имя таблицы. |
Пример формирования события изменения данных для операции insert into alltypes values (1, 2, 3, 4.21, 5.21, 6.31, 7.89, '08:00', '2016-04-09', '2016-04-10 10:31', 'eleven', 'twelve', 'thirteen', 'fourteen', 'fifteen', 'sixteen', true):
{
"schema": {
"type": "struct",
"fields": [
{
"type": "string",
"optional": false,
"field": "op"
},
{
"type": "int64",
"optional": false,
"field": "ts_ms"
},
{
"type": "struct",
"fields": [
{
"type": "string",
"optional": false,
"field": "version"
},
{
"type": "string",
"optional": false,
"field": "connector"
},
{
"type": "string",
"optional": false,
"field": "name"
},
{
"type": "int64",
"optional": false,
"field": "ts_ms"
},
{
"type": "string",
"optional": true,
"name": "io.debezium.data.Enum",
"version": 1,
"parameters": {
"allowed": "true,last,false,incremental"
},
"default": "false",
"field": "snapshot"
},
{
"type": "string",
"optional": false,
"field": "db"
},
{
"type": "string",
"optional": true,
"field": "sequence"
},
{
"type": "int64",
"optional": true,
"field": "ts_us"
},
{
"type": "int64",
"optional": true,
"field": "ts_ns"
},
{
"type": "int64",
"optional": false,
"field": "tra_number"
},
{
"type": "string",
"optional": false,
"field": "table"
}
],
"optional": false,
"name": "io.debezium.connector.firebird.Source",
"field": "source"
},
{
"type": "struct",
"fields": [
{
"type": "int64",
"optional": true,
"field": "DATECOLUMN"
},
{
"type": "string",
"optional": true,
"field": "VARBINARYCOLUMN"
},
{
"type": "int32",
"optional": true,
"field": "INTEGERCOLUMN"
},
{
"type": "double",
"optional": true,
"field": "DOUBLECOLUMN"
},
{
"type": "int32",
"optional": false,
"field": "SMALLINTCOLUMN"
},
{
"type": "string",
"optional": true,
"field": "VARCHARCOLUMN"
},
{
"type": "int64",
"optional": true,
"name": "io.debezium.time.Timestamp",
"version": 1,
"field": "TIMESTAMPCOLUMN"
},
{
"type": "float",
"optional": true,
"field": "FLOATCOLUMN"
},
{
"type": "boolean",
"optional": true,
"field": "BOOLEANCOLUMN"
},
{
"type": "double",
"optional": true,
"field": "DECIMALCOLUMN"
},
{
"type": "double",
"optional": true,
"field": "NUMERICCOLUMN"
},
{
"type": "int64",
"optional": true,
"field": "TIMECOLUMN"
},
{
"type": "string",
"optional": true,
"field": "BINARYCOLUMN"
},
{
"type": "string",
"optional": true,
"field": "CHARCOLUMN"
},
{
"type": "string",
"optional": true,
"field": "BLOBCOLUMN"
},
{
"type": "int64",
"optional": true,
"field": "BIGINTCOLUMN"
},
{
"type": "string",
"optional": true,
"field": "TEXTBLOBCOLUMN"
}
],
"optional": true,
"name": "reddatabase_topic._var_db_kafka.fdb.ALLTYPES.Value",
"field": "before"
},
{
"type": "struct",
"fields": [
{
"type": "int64",
"optional": true,
"field": "DATECOLUMN"
},
{
"type": "string",
"optional": true,
"field": "VARBINARYCOLUMN"
},
{
"type": "int32",
"optional": true,
"field": "INTEGERCOLUMN"
},
{
"type": "double",
"optional": true,
"field": "DOUBLECOLUMN"
},
{
"type": "int32",
"optional": false,
"field": "SMALLINTCOLUMN"
},
{
"type": "string",
"optional": true,
"field": "VARCHARCOLUMN"
},
{
"type": "int64",
"optional": true,
"name": "io.debezium.time.Timestamp",
"version": 1,
"field": "TIMESTAMPCOLUMN"
},
{
"type": "float",
"optional": true,
"field": "FLOATCOLUMN"
},
{
"type": "boolean",
"optional": true,
"field": "BOOLEANCOLUMN"
},
{
"type": "double",
"optional": true,
"field": "DECIMALCOLUMN"
},
{
"type": "double",
"optional": true,
"field": "NUMERICCOLUMN"
},
{
"type": "int64",
"optional": true,
"field": "TIMECOLUMN"
},
{
"type": "string",
"optional": true,
"field": "BINARYCOLUMN"
},
{
"type": "string",
"optional": true,
"field": "CHARCOLUMN"
},
{
"type": "string",
"optional": true,
"field": "BLOBCOLUMN"
},
{
"type": "int64",
"optional": true,
"field": "BIGINTCOLUMN"
},
{
"type": "string",
"optional": true,
"field": "TEXTBLOBCOLUMN"
}
],
"optional": true,
"name": "reddatabase_topic._var_db_kafka.fdb.ALLTYPES.Value",
"field": "after"
}
],
"optional": false,
"name": "reddatabase_topic._var_db_kafka.fdb.ALLTYPES.Envelope"
},
"payload": {
"op": "CREATE",
"ts_ms": 1788105950981000,
"source": {
"version": "2.1.2",
"connector": "RedDatabase",
"name": "vasiliy-rs",
"ts_ms": 1788105949959653,
"snapshot": "false",
"db": "/var/db/kafka.fdb",
"sequence": null,
"ts_us": 1788105949959653,
"ts_ns": 1788105949959653000,
"tra_number": 195,
"table": "ALLTYPES"
},
"before": null,
"after": {
"DATECOLUMN": 1460160000000,
"VARBINARYCOLUMN": "fifteen",
"INTEGERCOLUMN": 2,
"DOUBLECOLUMN": 6.31,
"SMALLINTCOLUMN": 1,
"VARCHARCOLUMN": "eleven",
"TIMESTAMPCOLUMN": 1460273460000,
"FLOATCOLUMN": 7.89,
"BOOLEANCOLUMN": true,
"DECIMALCOLUMN": 5.21,
"NUMERICCOLUMN": 4.21,
"TIMECOLUMN": 18000000,
"BINARYCOLUMN": "sixteen\u0000\u0000\u0000",
"CHARCOLUMN": "twelve ",
"BLOBCOLUMN": "fourteen",
"BIGINTCOLUMN": 3,
"TEXTBLOBCOLUMN": "thirteen"
}
}
}Пример формирования события изменения данных для операции update alltypes set varcharcolumn = 'updated' where smallintcolumn = 1:
{
"schema": {
"type": "struct",
"fields": [
{
"type": "string",
"optional": false,
"field": "op"
},
{
"type": "int64",
"optional": false,
"field": "ts_ms"
},
{
"type": "struct",
"fields": [
{
"type": "string",
"optional": false,
"field": "version"
},
{
"type": "string",
"optional": false,
"field": "connector"
},
{
"type": "string",
"optional": false,
"field": "name"
},
{
"type": "int64",
"optional": false,
"field": "ts_ms"
},
{
"type": "string",
"optional": true,
"name": "io.debezium.data.Enum",
"version": 1,
"parameters": {
"allowed": "true,last,false,incremental"
},
"default": "false",
"field": "snapshot"
},
{
"type": "string",
"optional": false,
"field": "db"
},
{
"type": "string",
"optional": true,
"field": "sequence"
},
{
"type": "int64",
"optional": true,
"field": "ts_us"
},
{
"type": "int64",
"optional": true,
"field": "ts_ns"
},
{
"type": "int64",
"optional": false,
"field": "tra_number"
},
{
"type": "string",
"optional": false,
"field": "table"
}
],
"optional": false,
"name": "io.debezium.connector.firebird.Source",
"field": "source"
},
{
"type": "struct",
"fields": [
{
"type": "int64",
"optional": true,
"field": "DATECOLUMN"
},
{
"type": "string",
"optional": true,
"field": "VARBINARYCOLUMN"
},
{
"type": "int32",
"optional": true,
"field": "INTEGERCOLUMN"
},
{
"type": "double",
"optional": true,
"field": "DOUBLECOLUMN"
},
{
"type": "int32",
"optional": false,
"field": "SMALLINTCOLUMN"
},
{
"type": "string",
"optional": true,
"field": "VARCHARCOLUMN"
},
{
"type": "int64",
"optional": true,
"name": "io.debezium.time.Timestamp",
"version": 1,
"field": "TIMESTAMPCOLUMN"
},
{
"type": "float",
"optional": true,
"field": "FLOATCOLUMN"
},
{
"type": "boolean",
"optional": true,
"field": "BOOLEANCOLUMN"
},
{
"type": "double",
"optional": true,
"field": "DECIMALCOLUMN"
},
{
"type": "double",
"optional": true,
"field": "NUMERICCOLUMN"
},
{
"type": "int64",
"optional": true,
"field": "TIMECOLUMN"
},
{
"type": "string",
"optional": true,
"field": "BINARYCOLUMN"
},
{
"type": "string",
"optional": true,
"field": "CHARCOLUMN"
},
{
"type": "string",
"optional": true,
"field": "BLOBCOLUMN"
},
{
"type": "int64",
"optional": true,
"field": "BIGINTCOLUMN"
},
{
"type": "string",
"optional": true,
"field": "TEXTBLOBCOLUMN"
}
],
"optional": true,
"name": "reddatabase_topic._var_db_kafka.fdb.ALLTYPES.Value",
"field": "before"
},
{
"type": "struct",
"fields": [
{
"type": "int64",
"optional": true,
"field": "DATECOLUMN"
},
{
"type": "string",
"optional": true,
"field": "VARBINARYCOLUMN"
},
{
"type": "int32",
"optional": true,
"field": "INTEGERCOLUMN"
},
{
"type": "double",
"optional": true,
"field": "DOUBLECOLUMN"
},
{
"type": "int32",
"optional": false,
"field": "SMALLINTCOLUMN"
},
{
"type": "string",
"optional": true,
"field": "VARCHARCOLUMN"
},
{
"type": "int64",
"optional": true,
"name": "io.debezium.time.Timestamp",
"version": 1,
"field": "TIMESTAMPCOLUMN"
},
{
"type": "float",
"optional": true,
"field": "FLOATCOLUMN"
},
{
"type": "boolean",
"optional": true,
"field": "BOOLEANCOLUMN"
},
{
"type": "double",
"optional": true,
"field": "DECIMALCOLUMN"
},
{
"type": "double",
"optional": true,
"field": "NUMERICCOLUMN"
},
{
"type": "int64",
"optional": true,
"field": "TIMECOLUMN"
},
{
"type": "string",
"optional": true,
"field": "BINARYCOLUMN"
},
{
"type": "string",
"optional": true,
"field": "CHARCOLUMN"
},
{
"type": "string",
"optional": true,
"field": "BLOBCOLUMN"
},
{
"type": "int64",
"optional": true,
"field": "BIGINTCOLUMN"
},
{
"type": "string",
"optional": true,
"field": "TEXTBLOBCOLUMN"
}
],
"optional": true,
"name": "reddatabase_topic._var_db_kafka.fdb.ALLTYPES.Value",
"field": "after"
}
],
"optional": false,
"name": "reddatabase_topic._var_db_kafka.fdb.ALLTYPES.Envelope"
},
"payload": {
"op": "UPDATE",
"ts_ms": 1788106339387000,
"source": {
"version": "2.1.2",
"connector": "RedDatabase",
"name": "vasiliy-rs",
"ts_ms": 1788106338368234,
"snapshot": "false",
"db": "/var/db/kafka.fdb",
"sequence": null,
"ts_us": 1788106338368234,
"ts_ns": 1788106338368234000,
"tra_number": 197,
"table": "ALLTYPES"
},
"before": {
"DATECOLUMN": 1460160000000,
"VARBINARYCOLUMN": "fifteen",
"INTEGERCOLUMN": 2,
"DOUBLECOLUMN": 6.31,
"SMALLINTCOLUMN": 1,
"VARCHARCOLUMN": "new value",
"TIMESTAMPCOLUMN": 1460273460000,
"FLOATCOLUMN": 7.89,
"BOOLEANCOLUMN": true,
"DECIMALCOLUMN": 5.21,
"NUMERICCOLUMN": 4.21,
"TIMECOLUMN": 18000000,
"BINARYCOLUMN": "sixteen\u0000\u0000\u0000",
"CHARCOLUMN": "twelve ",
"BLOBCOLUMN": "fourteen",
"BIGINTCOLUMN": 3,
"TEXTBLOBCOLUMN": "thirteen"
},
"after": {
"DATECOLUMN": 1460160000000,
"VARBINARYCOLUMN": "fifteen",
"INTEGERCOLUMN": 2,
"DOUBLECOLUMN": 6.31,
"SMALLINTCOLUMN": 1,
"VARCHARCOLUMN": "updated",
"TIMESTAMPCOLUMN": 1460273460000,
"FLOATCOLUMN": 7.89,
"BOOLEANCOLUMN": true,
"DECIMALCOLUMN": 5.21,
"NUMERICCOLUMN": 4.21,
"TIMECOLUMN": 18000000,
"BINARYCOLUMN": "sixteen\u0000\u0000\u0000",
"CHARCOLUMN": "twelve ",
"BLOBCOLUMN": "fourteen",
"BIGINTCOLUMN": 3,
"TEXTBLOBCOLUMN": "thirteen"
}
}
}Пример формирования события изменения данных для операции update delete from alltypes where smallintcolumn = 1:
{
"schema": {
"type": "struct",
"fields": [
{
"type": "string",
"optional": false,
"field": "op"
},
{
"type": "int64",
"optional": false,
"field": "ts_ms"
},
{
"type": "struct",
"fields": [
{
"type": "string",
"optional": false,
"field": "version"
},
{
"type": "string",
"optional": false,
"field": "connector"
},
{
"type": "string",
"optional": false,
"field": "name"
},
{
"type": "int64",
"optional": false,
"field": "ts_ms"
},
{
"type": "string",
"optional": true,
"name": "io.debezium.data.Enum",
"version": 1,
"parameters": {
"allowed": "true,last,false,incremental"
},
"default": "false",
"field": "snapshot"
},
{
"type": "string",
"optional": false,
"field": "db"
},
{
"type": "string",
"optional": true,
"field": "sequence"
},
{
"type": "int64",
"optional": true,
"field": "ts_us"
},
{
"type": "int64",
"optional": true,
"field": "ts_ns"
},
{
"type": "int64",
"optional": false,
"field": "tra_number"
},
{
"type": "string",
"optional": false,
"field": "table"
}
],
"optional": false,
"name": "io.debezium.connector.firebird.Source",
"field": "source"
},
{
"type": "struct",
"fields": [
{
"type": "int64",
"optional": true,
"field": "DATECOLUMN"
},
{
"type": "string",
"optional": true,
"field": "VARBINARYCOLUMN"
},
{
"type": "int32",
"optional": true,
"field": "INTEGERCOLUMN"
},
{
"type": "double",
"optional": true,
"field": "DOUBLECOLUMN"
},
{
"type": "int32",
"optional": false,
"field": "SMALLINTCOLUMN"
},
{
"type": "string",
"optional": true,
"field": "VARCHARCOLUMN"
},
{
"type": "int64",
"optional": true,
"name": "io.debezium.time.Timestamp",
"version": 1,
"field": "TIMESTAMPCOLUMN"
},
{
"type": "float",
"optional": true,
"field": "FLOATCOLUMN"
},
{
"type": "boolean",
"optional": true,
"field": "BOOLEANCOLUMN"
},
{
"type": "double",
"optional": true,
"field": "DECIMALCOLUMN"
},
{
"type": "double",
"optional": true,
"field": "NUMERICCOLUMN"
},
{
"type": "int64",
"optional": true,
"field": "TIMECOLUMN"
},
{
"type": "string",
"optional": true,
"field": "BINARYCOLUMN"
},
{
"type": "string",
"optional": true,
"field": "CHARCOLUMN"
},
{
"type": "string",
"optional": true,
"field": "BLOBCOLUMN"
},
{
"type": "int64",
"optional": true,
"field": "BIGINTCOLUMN"
},
{
"type": "string",
"optional": true,
"field": "TEXTBLOBCOLUMN"
}
],
"optional": true,
"name": "reddatabase_topic._var_db_kafka.fdb.ALLTYPES.Value",
"field": "before"
},
{
"type": "struct",
"fields": [
{
"type": "int64",
"optional": true,
"field": "DATECOLUMN"
},
{
"type": "string",
"optional": true,
"field": "VARBINARYCOLUMN"
},
{
"type": "int32",
"optional": true,
"field": "INTEGERCOLUMN"
},
{
"type": "double",
"optional": true,
"field": "DOUBLECOLUMN"
},
{
"type": "int32",
"optional": false,
"field": "SMALLINTCOLUMN"
},
{
"type": "string",
"optional": true,
"field": "VARCHARCOLUMN"
},
{
"type": "int64",
"optional": true,
"name": "io.debezium.time.Timestamp",
"version": 1,
"field": "TIMESTAMPCOLUMN"
},
{
"type": "float",
"optional": true,
"field": "FLOATCOLUMN"
},
{
"type": "boolean",
"optional": true,
"field": "BOOLEANCOLUMN"
},
{
"type": "double",
"optional": true,
"field": "DECIMALCOLUMN"
},
{
"type": "double",
"optional": true,
"field": "NUMERICCOLUMN"
},
{
"type": "int64",
"optional": true,
"field": "TIMECOLUMN"
},
{
"type": "string",
"optional": true,
"field": "BINARYCOLUMN"
},
{
"type": "string",
"optional": true,
"field": "CHARCOLUMN"
},
{
"type": "string",
"optional": true,
"field": "BLOBCOLUMN"
},
{
"type": "int64",
"optional": true,
"field": "BIGINTCOLUMN"
},
{
"type": "string",
"optional": true,
"field": "TEXTBLOBCOLUMN"
}
],
"optional": true,
"name": "reddatabase_topic._var_db_kafka.fdb.ALLTYPES.Value",
"field": "after"
}
],
"optional": false,
"name": "reddatabase_topic._var_db_kafka.fdb.ALLTYPES.Envelope"
},
"payload": {
"op": "DELETE",
"ts_ms": 1788106483419000,
"source": {
"version": "2.1.2",
"connector": "RedDatabase",
"name": "vasiliy-rs",
"ts_ms": 1788106482397150,
"snapshot": "false",
"db": "/var/db/kafka.fdb",
"sequence": null,
"ts_us": 1788106482397150,
"ts_ns": 1788106482397150000,
"tra_number": 199,
"table": "ALLTYPES"
},
"before": {
"DATECOLUMN": 1460160000000,
"VARBINARYCOLUMN": "fifteen",
"INTEGERCOLUMN": 2,
"DOUBLECOLUMN": 6.31,
"SMALLINTCOLUMN": 1,
"VARCHARCOLUMN": "updated",
"TIMESTAMPCOLUMN": 1460273460000,
"FLOATCOLUMN": 7.89,
"BOOLEANCOLUMN": true,
"DECIMALCOLUMN": 5.21,
"NUMERICCOLUMN": 4.21,
"TIMECOLUMN": 18000000,
"BINARYCOLUMN": "sixteen\u0000\u0000\u0000",
"CHARCOLUMN": "twelve ",
"BLOBCOLUMN": "fourteen",
"BIGINTCOLUMN": 3,
"TEXTBLOBCOLUMN": "thirteen"
},
"after": null
}
}События изменения схем
Коннектор генерирует события изменения схемы, которые основываются на выполняемых DDL операций на сервере. Коннектор отправляет сообщение каждый раз, когда происходит изменение структуры базы данных. Сообщения, которые коннектор отправляет в тему изменения схемы, содержат полезную нагрузку и также содержат схему о событии изменения. Полезная нагрузка сообщения о событии изменения схемы включает следующие элементы:
N | Значение | Описание |
1 | databaseName | Идентифицирует базу данных, содержащую изменение |
2 | schemaName | Идентифицирует схему, содержащую изменение |
3 | ddl | Текст DDL-запроса |
4 | tableChanges | Массив из одного или нескольких элементов, которые содержат изменения схемы, созданные DDL-запросом |
5 | source | Аналогичен полезной нагрузке события изменения записи |
6 | ts_ms | Время (по системным часам в JVM), в которое коннектор обработал событие |
Пример формирования события изменения данных для операции alter table ft_test add f_int integer:
{
"schema": {
"type": "struct",
"fields": [
{
"type": "struct",
"fields": [
{
"type": "string",
"optional": false,
"field": "version"
},
{
"type": "string",
"optional": false,
"field": "connector"
},
{
"type": "string",
"optional": false,
"field": "name"
},
{
"type": "int64",
"optional": false,
"field": "ts_ms"
},
{
"type": "string",
"optional": true,
"name": "io.debezium.data.Enum",
"version": 1,
"parameters": {
"allowed": "true,last,false,incremental"
},
"default": "false",
"field": "snapshot"
},
{
"type": "string",
"optional": false,
"field": "db"
},
{
"type": "string",
"optional": true,
"field": "sequence"
},
{
"type": "int64",
"optional": true,
"field": "ts_us"
},
{
"type": "int64",
"optional": true,
"field": "ts_ns"
},
{
"type": "int64",
"optional": false,
"field": "tra_number"
},
{
"type": "string",
"optional": false,
"field": "table"
}
],
"optional": false,
"name": "io.debezium.connector.firebird.Source",
"field": "source"
},
{
"type": "int64",
"optional": false,
"field": "ts_ms"
},
{
"type": "string",
"optional": true,
"field": "databaseName"
},
{
"type": "string",
"optional": true,
"field": "schemaName"
},
{
"type": "string",
"optional": true,
"field": "ddl"
},
{
"type": "array",
"items": {
"type": "struct",
"fields": [
{
"type": "string",
"optional": false,
"field": "type"
},
{
"type": "string",
"optional": false,
"field": "id"
},
{
"type": "struct",
"fields": [
{
"type": "string",
"optional": true,
"field": "defaultCharsetName"
},
{
"type": "array",
"items": {
"type": "string",
"optional": false
},
"optional": true,
"field": "primaryKeyColumnNames"
},
{
"type": "array",
"items": {
"type": "struct",
"fields": [
{
"type": "string",
"optional": false,
"field": "name"
},
{
"type": "int32",
"optional": false,
"field": "jdbcType"
},
{
"type": "int32",
"optional": true,
"field": "nativeType"
},
{
"type": "string",
"optional": false,
"field": "typeName"
},
{
"type": "string",
"optional": true,
"field": "typeExpression"
},
{
"type": "string",
"optional": true,
"field": "charsetName"
},
{
"type": "int32",
"optional": true,
"field": "length"
},
{
"type": "int32",
"optional": true,
"field": "scale"
},
{
"type": "int32",
"optional": false,
"field": "position"
},
{
"type": "boolean",
"optional": true,
"field": "optional"
},
{
"type": "boolean",
"optional": true,
"field": "autoIncremented"
},
{
"type": "boolean",
"optional": true,
"field": "generated"
},
{
"type": "string",
"optional": true,
"field": "comment"
},
{
"type": "string",
"optional": true,
"field": "defaultValueExpression"
},
{
"type": "array",
"items": {
"type": "string",
"optional": false
},
"optional": true,
"field": "enumValues"
}
],
"optional": false,
"name": "io.debezium.connector.schema.Column",
"version": 1
},
"optional": false,
"field": "columns"
},
{
"type": "string",
"optional": true,
"field": "comment"
}
],
"optional": true,
"name": "io.debezium.connector.schema.Table",
"version": 1,
"field": "table"
}
],
"optional": false,
"name": "io.debezium.connector.schema.Change",
"version": 1
},
"optional": false,
"field": "tableChanges"
}
],
"optional": false,
"name": "io.debezium.connector.RedDatabase.SchemaChangeValue",
"version": 1
},
"payload": {
"source": {
"version": "2.1.2",
"connector": "RedDatabase",
"name": "vasiliy-rs",
"ts_ms": 1788105252291408,
"snapshot": "false",
"db": "/var/db/kafka.fdb",
"sequence": null,
"ts_us": 1788105252291408,
"ts_ns": 1788105252291408000,
"tra_number": 187,
"table": ""
},
"ts_ms": 1788105253388,
"databaseName": "/var/db/kafka.fdb",
"schemaName": null,
"ddl": "alter table ft_test add f_int integer;",
"tableChanges": [
{
"type": "ALTER",
"id": "\"/var/db/kafka.fdb\".\"FT_TEST\"",
"table": {
"defaultCharsetName": null,
"primaryKeyColumnNames": [],
"columns": [
{
"name": "F_INT",
"jdbcType": 2,
"nativeType": null,
"typeName": "NUMBER",
"typeExpression": "NUMBER",
"charsetName": null,
"length": 38,
"scale": 0,
"position": 1,
"optional": true,
"autoIncremented": false,
"generated": false,
"comment": null,
"defaultValueExpression": null,
"enumValues": null
}
],
"comment": null
}
}
]
}
}Режимы работы коннектора CDC
Ранее было рассказано, как коннектор получает данные из базы и формирует сообщения для Kafka. А как он обрабатывает эти сообщения? Отправляет сразу или ждет завершения транзакции? Ответ – коннектор может работать по-разному в зависимости от требуемого поведения.
Всего CDC-коннектор РЕД Базы Данных поддерживает 3 режима работы:
no_data – отправляет на сервер Kafka только изменения в базе данных, сделанные после включения коннектора, режим no_data используется по умолчанию
initial – отправляет на сервер Kafka cуществующие данные таблиц, после чего продолжает работу в режиме no_data
initial_only – отправляет на сервер Kafka данные, существующие на момент старта коннектора и прекращает отслеживание изменений
Чуть позже я подробнее расскажу про режимы. А для начала разберем, что делает коннектор во время работы транзакций.
При старте транзакции репликации коннектор создает свой собственный журнал, куда будет записывать приходящие изменения. Пока транзакция жива, коннектор будет держать журнал открытым. Как только транзакция завершится коммитом, коннектор сохранит журнал и начнет отсылать из него данные. Если транзакция была откатана, т.е. сделан роллбэк, журнал не сохранится. Журналы коннектора позволяют реализовать повторную отправку, если кластер был недоступен или возникли другие ошибки при первичной отправке сообщений транзакции. А может ли коннектор СРАЗУ отправлять данные транзакции? Может! Причем может параллельно вести журнал, если возникнут какие-то неприятиности при работе с брокером. Также ведение журналов можно отключить, если есть уверенность, что при работе с Kafka не будет проблем.
Теперь вернемся к режимам работы коннектора. Режим работы можно задать в опции snapshot.mode параметров коннектора (об этом будет рассказано далее). Как уже было сказано, режим no_data включен по умолчанию и отслеживает только изменения существующих данных. Первоначалный снимок таблиц не делается.
Если необходимо отправить существующие данные таблиц, то можно включить режим initial. Коннектор считает и отпарвит существующие данные, а затем будет отправлять изменения.
А что если существующая таблица большая? Коннектор может вычитывать данные долгое время, что повлияет на работу сервера РЕД Базы Данных. Для решения этой ситуации был реализован режим initial_only. И самое главное – коннектор работает не на стороне сервера РЕД Базы Данных, а на стороне Kafka Connect! Библиотеку коннектора нужно добавить в список библиотек Kafka Connect и зарегистрировать коннектор. Тогда коннектор считает существующие данные из таблиц, не прерывая работу репликации. Параллельно можно запустить коннектор в режиме no_data на стороне сервера. Это даст возможность получить полную копию таблицы и ее изменений за минимальное время работы.
Схематично режимы работы представлены ниже:
Оранжевыми стрелками показана работа коннектора с фремворком Kafka Connect в режиме initial_only.
Настройка CDC-коннектора РЕД Базы Данных
Так как же запустить коннектор, чтобы отправлять изменения брокеру? Для начала необходимо включить его в файле конфигурации плагинов plugins.conf:
Plugin = Debezium_Connector {
Module = $(dir_plugins)/debezium_connector
Config = Debezium_Connector_config
}
Config = Debezium_Connector_config {
JarDirs = $(this)/debezium
}Шаблон настройки уже существует, можно просто его раскомментировать.
Настройка репликации для работы коннектора
Теперь нужно настроить репликацию. Да, без нее плагин работать не будет. Настройка производится в конфигурационном файле replication.conf. Там в секции database необходимо указать плагин коннектора:
database
{
plugin = Debezium_Connector
}Настройка выше позволит отслеживать изменения плагином во всех базах данных, с которыми будет работать сервер. А что если не требуется отслеживать изменения во всех базах, а только в одной или нескольких? Конечно, это можно настроить! Для этого просто добавляются секции database с указанием пути к базе:
database = /your/db.fdb
{
plugin = Debezium_Connector
}
database = /var/db/rdb.fdb
{
plugin = Debezium_Connector
}Следует учитывать, что в разделе database должен быть указан полный путь к базе данных! Псевдонимы и подстановочные знаки не допускаются!
Итак, настройка репликации в конфигурационном файле replication.conf выполнена. Теперь необходимо для каждой базы данных, указанных в replication.conf, включить репликацию и добавить таблицы для отслеживания изменений в публикации. Для этого в базе данных необходимо выполнить DDL-команды:
ALTER DATABASE ENABLE PUBLICATION;Эта команда включает репликацию в базе. А добавление таблицы в список реплицируемых осуществляется командой:
ALTER DATABASE INCLUDE TABLE <имя таблицы> TO PUBLICATION;Если вы хотите отслеживать изменения во всех таблицах, даже в тех, которые будут созданы в процессе работы, можно выполнить команду:
ALTER DATABASE INCLUDE ALL TO PUBLICATION;Помните, что без добавления таблиц в список реплицируемых коннектор не сможет отслеживать в них изменения!
Теперь можно переходить к настройке самого коннектора.
Конфигурация коннектора CDC
А теперь самое интересное – настройка коннектора! Настройка осуществляется в файле reddatabase.connector.yaml. В этом файле задаются параметры коннектора, а также настройки продюсеров, осуществляющих отправку сообщений в Kafka.
Настройки коннектора:
connector.name – имя коннектора, здесь можно указать любую строку
snapshot.mode – режимы работы коннектора, no_data, initial или initial_only
connector.live.streaming – принимает значения true или false. Если установлено значение true, то данные будут отправляться в Kafka сразу. Если при этом задана опция connector.journal.directory, то данные будут сохраняться ещё и в журналы репликации. В случае ошибки отправки сообщений коннектор попытается отправить изменения из журналов, после успешной отправки журнал этой транзакции удаляется. Если установлено значение false, то изменения будут записываться в журналы репликации и передаваться только после фиксации транзакции на сервере
connector.journal.directory – каталог, в котором коннектор будет сохранять журналы
connector.journal.check.interval.ms – период просмотра наличия журналов транзакций в миллисекундах
kafka.producer.bootstrap.servers – список пар хост/порт, используемых для установления начального соединения с кластером Kafka
kafka.topic.prefix – логическое имя, которое идентифицирует и обеспечивает пространство имен для конкретного сервера РЕД Базы Данных, на котором фиксируются изменения
key.converter – конвертер ключей
value.converter – конвертер значений
И это далеко не все настройки, которые можно задать в этом файле! Полный список можно посмотреть в документации.
Ниже представлен пример конфигурации:
connector.name: reddatabase
connector.journal.directory: /var/reddatabase-connector-journals
connector.journal.check.interval.ms: 0
kafka.producer.bootstrap.servers: kafka:9092
kafka.producer.retries: 3
kafka.producer.retry.backoff.ms: 1000
kafka.producer.max.block.ms: 1000
kafka.producer.timeout.ms: 1000
kafka.producer.transactional.id: reddatabase-transaction-id
kafka.topic.prefix: reddatabase_topic
key.converter: org.apache.kafka.connect.json.JsonConverter
value.converter: org.apache.kafka.connect.json.JsonConverterА можно ли сделать так, чтобы было несколько коннекторов на стороне РЕД Базы Данных и каждый из них писал в определенный сервер Kafka? Конечно можно! Для этого необходимо настроить несколько плагинов репликации, указав каждому свои настройки:
Plugin = Debezium_Connector {
Module = $(dir_plugins)/debezium_connector
Config = Debezium_Connector_config
}
Config = Debezium_Connector_config {
JarDirs = $(this)/debezium
Connector_Config = $(this)/reddatabase.connector.yaml
}
Plugin = Debezium_Connector_Other {
Module = $(dir_plugins)/debezium_connector
RegisterName = Debezium_Connector
Config = Debezium_Connector_Other _config
}
Config = Debezium_Connector_Other _config {
JarDirs = $(this)/debezium
Connector_Config = $(this)/other.reddatabase.connector.yaml
}
Для каждого из экземпляров коннектора используются свои файлы конфигураций.
А вот здесь пример для режима intial_only на стороне Kafka Connect:
{
"name": "kafka-reddatabase-connector",
"config": {
"connector.class": "io.debezium.connector.firebird.KafkaFirebirdConnector",
"database.hostname": "10.81.100.31",
"database.port": "3050",
"database.user": "SYSDBA",
"database.password": "masterkey",
"database.dbname" : "employee",
"topic.prefix": "rdb-topic-prx-",
"table.include.list": "employee",
"batch.size": "1000",
"poll.interval.ms": "5000",
"snapshot.mode": "initial_only",
"key.converter": "org.apache.kafka.connect.json.JsonConverter",
"value.converter": "org.apache.kafka.connect.json.JsonConverter",
"key.converter.schemas.enable": "true",
"value.converter.schemas.enable": "true"
}
}Зарегистрировать и запустить коннектор можно так:
curl -X POST -H "Content-Type: application/json" --data @connector-config/kafka-reddatabase-connector.json http://kafka-connect:8083/connectors | jqЧто в итоге?
Прочитав статью, может сложиться впечатление, что как-то сложно с этим коннектором, есть встроенная базовая репликация, давайте использовать ее! Да, встроенная репликация, может, и будет удобнее и в некоторых моментах быстрее, но работает она только между серверами РЕД Базы Данных, что очень ограничивает число решаемых задач. Использование коннектора CDC позволяет построить гетерогенный конвейер данных между различными системами! Это позволит использовать преимущества различных технологий.
Также благодаря CDC данные можно изменять (ETL), что практически невозможно во встроенной репликации, и видеть всю историю изменений.
Одним из главных преимуществ CDC-коннектора является снижение нагрузки на рабочий сервер РЕД Базы Данных. Независимые потребители работают с одним или несколькими серверами Kafka, в то время как сервер СУБД продолжает свою работу без дополнительных трат на обработку запросов от них.
Оцените, как просто ваши данные начинают двигаться туда, где они действительно нужны, – скачайте РЕД Базу Данных и проверьте коннектор в деле.
KioskNews shows a cleaned-up reading view extracted from the publisher’s page — the original always lives on their site, not ours.