klustrondb CDC User Manual 2
klustrondb CDC User Manual 2
01 Overview
KlustronDB CDC (change data capture) is used to export the data updates from each storage shard (shard) in the KlustronDB distributed database cluster into real-time event streams, with one event stream per shard, for consumption by external components. For example, external components can perform data streaming imports, importing KlustronDB's data updates into third-party database systems, or exporting to event stream analysis systems like Flink. KlustronDB CDC also supports real-time exporting of data update event streams from third-party open-source MySQL, allowing data to be streamed into the KlustronDB cluster.
KlustronDB CDC uses the MySQL binlog dump protocol to connect to MySQL or KlustronDB Klustore instances to capture and transform data updates in real time, then output them in various formats, including JSON objects and SQL statements. For MySQL or KlustronDB Klustore instance (KlustronDB_storage) instances, the KlustronDB CDC component acts like a standby node. Currently, KlustronDB CDC supports two modes: one for exporting data from the KlustronDB cluster, and one for exporting data from open-source MySQL instances.
Starting from Klustron-1.3 version, KlustronDB-cdc supports exporting to Kafka message queues. Starting from Klustron-1.3 version, KlustronDB-cdc also supports parallel exporting of SQL statement flows to other database systems to improve data flow export speed by several times. KlustronDB-CDC starts multiple working threads, with each thread using an independent connection to the target database, then divides the target data tables into several groups assigned to each working thread to achieve parallel export. This requires that each table have a primary key or unique index so that if the export process is interrupted due to various software or hardware failures, Klustron-CDC can correctly continue the export.
02 Function Usage
After reading this article, you can refer to CDC Usage Example to complete the actual data streaming export configuration operations.
2.1 KlustronDB Cluster Export
KlustronDB CDC connects to the KlustronDB metashard based on dump task parameters to obtain the shard parameters of the KlustronDB cluster that need to be dumped. If there are multiple shards, CDC will automatically establish a dump connection for each shard. KlustronDB CDC will select the replica node with the lowest latency among the shard nodes to dump data. If a master-slave switch occurs in the shard during the dump process, such as the current dump node becoming the master node or the dump node failing, CDC will automatically disconnect the current dump connection and reselect another replica node with the lowest latency in the shard to continue dumping.
2.2 Open Source MySQL Export
KlustronDB CDC connects to the MySQL node that needs to be dumped according to the dump task parameters. Currently, for open-source MySQL, CDC will monitor the dump connection status in real time. If the connection is interrupted during the dump, for example due to network issues or the source MySQL killing the connection, CDC will automatically reconnect.
2.3 Event Output and Transformation Plugins
KlustronDB CDC uses a plugin approach to transform and output the captured binlog data. KlustronDB CDC provides API interfaces for developing plugins, allowing users to develop plugins based on the APIs provided by KlustronDB CDC and then attach them to KlustronDB CDC to handle the events generated by CDC. Currently, when KlustronDB CDC is released, it comes with two built-in plugins: event_file and event_sql.
The function of Event_file is: to directly serialize and store CDC output JSON content into a specified file.
The function of Event_sql is: to convert CDC output JSON content into SQL statements and send them directly to the KlustronDB cluster or open-source MySQL.
When KlustronDB CDC synchronizes captured data to the target storage, it can ensure that no data is lost. CDC will back up the synchronization point in real time according to the GTID information of the dump data. If the CDC module exits during the dump process due to various hardware or software failures, it can automatically continue synchronization based on the last synchronization point after restarting. The CDC module supports cluster deployment to achieve high availability.
If you need to implement other conversion plugins, you need to implement the conversion plugins according to the following plugin interface, and then mount them to KlustronDB CDC.
2.4 Data Consistency in Resumable Uploads
KlustronDB CDC is a Raft cluster, with its primary node responsible for event stream processing. If the primary node of KlustronDB CDC unexpectedly exits, the KlustronDB CDC cluster will automatically elect a new primary node to continue working. The new primary node starts executing from the last saved position before the previous primary node exited. The interval for saving positions can be set, with the default being 5 seconds. Therefore, the data tables at the target end must have a primary key to ensure that SQL statements are not executed repeatedly. Otherwise, the new CDC primary node will re-execute all operations that occurred in the last few seconds before the previous primary node exited, which could lead to the related data rows being inserted, deleted, or updated repeatedly. This would result in the target data no longer being consistent with the source database, and it could also cause subsequent data synchronization to fail and be unable to continue.
2.5 Event Output Interface Description
2.5.1 JSON format of binlog events output by KlustronDB CDC
KlustronDB CDC outputs each binlog event as a JSON object with the following attributes.
| Field Name | Description |
|---|---|
| gtid | The GTID of the current event |
| database | Database name |
| table | Table name |
| isDdl | Whether it is a DDL |
| sql | SQL statement executed for DDL |
| event_type | Event name |
| data | If insert, the data for each column being inserted; if delete, the data being deleted; if update, the data after the update |
| old | If insert, empty; if delete, empty; if update, the data before the update |
2.5.2 Binlog Event Types Supported for Output by KlustronDB CDC
| Field Name | Description |
|---|---|
| CREATE_DB | Create database |
| DROP_DB | Drop database |
| CREATE_TABLE | Create table |
| DROP_TABLE | Drop table; this statement supports dropping multiple tables at once. In kunlun_cdc, multiple tables are split into separate DROP TABLE records. |
| CREATE_INDEX | Create index |
| DROP_INDEX | Drop index |
| ALTER_TABLE | Add, drop, and update table columns, etc. |
| RENAME_TABLE | Rename table; this statement supports renaming multiple tables at once. In kunlun_cdc, multiple tables are split into separate RENAME TABLE records. |
| INSERT | Insert data |
| DELETE | Delete data |
| UPDATE | Update data |
For DDL statements, the event's JSON field isDdl=1, and sql records the current DDL statement, for example:
{"event_type":"CREATE_TABLE","db_name":"test","sql":"create table t (a int primary key, b int)","isDdl":"1","table_name":"t","data":"","old":"","gtid":"77cf0403-fe85-11ed-87ad-fc3497a73395:5620"}
For DML statements, the json field isDdl=0, and the specific content is recorded in the data and old fields. For example:
{"event_type":"INSERT","db_name":"test","table_name":"t","sql":"","isDdl":"0","gtid":"77cf0403-fe85-11ed-87ad-fc3497a73395:5621","data":[{"a":"1","b":"1"}], "old":""}
2.6 Custom Plugin Development and Mounting Methods
You need to download the KlustronDB CDC software from the official KlustronDB website, and find the header file dispatch_event.h required for developing plugins in the include subdirectory of the software package.
Inherit the CDispatchEvent class and implement the three virtual function interfaces: init, execute, and close. The usage of each function is explained in detail in dispatch_event.h.
Compile the code into a .so file on the Linux operating system and place it in the plugin directory of the kunlun_cdc installation package.
In the conf directory of the kunlun_cdc installation package, add the name of the newly developed plugin under the plugin_so tag in kunlun_cdc.cnf.
For example, if the plugin so is named event_test.so, then add the following content under the plugin_so tag in the kunlun_cdc.cnf file:
plugin_so = event_file,event_sql,event_test
Special note: Separate plugin names with commas
5、修改完配置文件，需要重启KlustronDB CDC集群所有CDC进程才可以生效
Since the plug-in user develops the plug-in, the plug-in needs to enter parameters, then enter the parameters required for the plug-in under the output_plugins tab when adding the dump task.
03 Deployment Architecture Diagram
04 Configuration and Usage
4.1 Configure via API
4.1.1 Add dump data task (asynchronous interface)
4.1.1.1 Dump data from KlustronDB cluster
- To specify dumping from a specific position, you need to specify the starting dump position of each shard under that cluster. You must specify binlog_file, binlog_pos, and gtid_set.
curl -d '
{
"version":"1.0",
"job_id":"",
"job_type":"add_dump_table",
"timestamp":"1435749309",
"user_name":"kunlun_test",
"paras":{
"meta_db":"172.0.0.1:28001,172.0.0.2:28001,172.0.0.3:28001",
"meta_user":"xxx",
"meta_passwd":"xxxx",
"dump_db_type":"kunlunbase",
"cluster_name":"cluster_xxx_xxx",
"dump_tables":"postgres_$$_public.t1,postgres_$$_public.t2",
"shard_params":[
{
"shard_id":"1",
"dump_hostaddr":"127.0.0.1",
"dump_port":"28801",
"binlog_file":"xxx",
"binlog_pos":"899",
"gtid_set":"xxxx"
},{
"shard_id":"2",
"dump_hostaddr":"127.0.0.2",
"dump_port":"28802",
"binlog_file":"xxx",
"binlog_pos":"899",
"gtid_set":"xxxx"
}
],
"output_plugins":[
{
"plugin_name":"event_file",
"plugin_param":"{\"log_path\":\"../log\"}",
"udf_name":"test1"
},
{
"plugin_name":"event_sql",
"plugin_param":"{\"hostaddr\":\"172.0.0.5\",\"port\":\"24002\",\"user\":\"xxxx\",\"password\":\"xxx\",\"log_path\":\"../log\"}",
"udf_name":"test2"
}
]
}
}' -X POST http://172.0.0.1:18002/kunlun_cdc
- Start dumping from the current time when the dump task is added
curl -d '
{
"version":"1.0",
"job_id":"",
"job_type":"add_dump_table",
"timestamp":"1435749309",
"user_name":"kunlun_test",
"paras":{
"meta_db":"172.0.0.1:28001,172.0.0.2:28001,127.0.0.3:28001",
"meta_user":"xxx",
"meta_passwd":"xxx",
"dump_db_type":"kunlunbase",
"cluster_name":"cluster_xxx_xx",
"dump_tables":"postgres_$$_public.t1,postgres_$$_public.t2",
"output_plugins":[{
"plugin_name":"event_file",
"plugin_param":"{\"log_path\":\"../log\"}",
"udf_name":"test1"
},{
"plugin_name":"event_sql",
"plugin_param":"{\"hostaddr\":\"172.0.0.6\",\"port\":\"24002\",\"user\":\"xxx\",\"password\":\"xxx\",\"log_path\":\"../log\"}",
"udf_name":"test2"
}]
}
}
' -X POST http://172.0.0.1:18002/kunlun_cdc
4.1.1.2 When dumping data from an open-source MySQL cluster, you must specify the exact dump location, that is, set the shard_params parameter. If not specified, adding the task will fail.
curl -d '
{
"version":"1.0",
"job_id":"",
"job_type":"add_dump_table",
"timestamp":"1435749309",
"user_name":"kunlun_test",
"paras":{
"meta_db":"127.0.0.1:28001", --- dump mysql的ip:port
"meta_user":"xxx", --- 连接mysql的账户
"meta_passwd":"xxx", --- 连接mysql的密码
"cluster_name":"mysql",
"dump_db_type":"mysql",
"dump_tables":"test.t1,test.t2",
"shard_params":[{
"binlog_file":"xxx",
"binlog_pos":"899",
"gtid_set":"xxxx"
}],
"output_plugins":[
{
"plugin_name":"event_file",
"plugin_param":"{\"log_path\":\"../log\"}",
"udf_name":"test1"
},
{
"plugin_name":"event_sql",
"plugin_param":"{\"hostaddr\":\"172.0.0.2\",\"port\":\"24002\",\"user\":\"abc\",\"password\":\"abc\",\"log_path\":\"../log\"}",
"udf_name":"test2"
}
]
}
} ' -X POST http://172.0.0.1:18002/kunlun_cdc
4.1.1.3 When dumping data from an open-source MariaDB cluster, you must specify the exact dump location, that is, set the shard_params parameter. If it is not specified, adding the task will fail.
curl -d '
{
"version":"1.0",
"job_id":"",
"job_type":"add_dump_table",
"timestamp":"1435749309",
"user_name":"kunlun_test",
"paras":{
"meta_db":"127.0.0.1:28001", --- dump mariadb的ip:port
"meta_user":"xxx", --- 连接mariadb的账户
"meta_passwd":"xxx", --- 连接mariadb的密码
"cluster_name":"mysql",
"dump_db_type":"mariadb",
"dump_tables":"test.t1,test.t2",
"shard_params":[{
"binlog_file":"xxx",
"binlog_pos":"899",
"gtid_set":"xxxx"
}],
"output_plugins":[
{
"plugin_name":"event_file",
"plugin_param":"{\"log_path\":\"../log\"}",
"udf_name":"test1"
},
{
"plugin_name":"event_sql",
"plugin_param":"{\"hostaddr\":\"172.0.0.2\",\"port\":\"24002\",\"user\":\"abc\",\"password\":\"abc\",\"log_path\":\"../log\"}",
"udf_name":"test2"
}
]
}
} ' -X POST http://172.0.0.1:18002/kunlun_cdc
Special Note: When using add_dump_table, if dump_db_type is not provided, the default value is kunlunbase
4.1.1.4 When dumping data from a Redis cluster, you must specify the exact dump location by setting the shard_params parameter; if it is not specified, adding the task will fail.
curl -d '
{
"version":"1.0",
"job_id":"",
"job_type":"add_dump_table",
"timestamp":"1435749309",
"user_name":"kunlun_test",
"paras":{
"meta_db":"127.0.0.1:28001", --- dump redis的ip:port
"meta_user":"xxx", --- 连接redis的账户,可选
"meta_passwd":"xxx", --- 连接redis的密码，可选
"cluster_name":"mysql",
"dump_db_type":"redis",
"dump_tables":"test.t1,test.t2",
"is_tls":"0", ---- 是否加密连接
"shard_params":[{
"binlog_file":"xxx",
"binlog_pos":"899",
"gtid_set":"xxxx"
}],
"output_plugins":[
{
"plugin_name":"event_file",
"plugin_param":"{\"log_path\":\"../log\"}",
"udf_name":"test1"
},
{
"plugin_name":"redis_sql",
"plugin_param":"{\"hostaddr\":\"172.0.0.2\",\"port\":\"24002\",\"user\":\"abc\",\"password\":\"abc\",\"log_path\":\"../log\"}",
"udf_name":"test2"
}
]
}
} ' -X POST http://172.0.0.1:18002/kunlun_cdc
Special note: meta_user and meta_passwd configure whether to use an account/password to connect to Redis. If the dumped Redis does not have this configured, it can be left empty. is_tls indicates whether the connection is encrypted, with the default being 0, meaning an unencrypted connection.
Under the shard_params tag, binlog_file is the ID of the Redis dump, and binlog_pos is the position of the Redis dump. If binlog_file/binlog_pos is empty, do a full dump of Redis data before performing an incremental dump.
If Redis data is dumped into a relational database, use the redis_sql plugin.
4.1.1.5 When dumping data from kunlunbase/mysql/mariadb, it supports configuring whether to do full data first and then incremental.
curl -d '
{
"version":"1.0",
"job_id":"",
"job_type":"add_dump_table",
"timestamp":"1435749309",
"user_name":"kunlun_test",
"paras":{
"meta_db":"172.0.0.1:28001,172.0.0.2:28001,172.0.0.3:28001",
"meta_user":"xxx",
"meta_passwd":"xxxx",
"dump_db_type":"kunlunbase",
"cluster_name":"cluster_xxx_xxx",
"dump_tables":"postgres_$$_public.t1,postgres_$$_public.t2",
"need_alldump":"1",
"output_plugins":[
{
"plugin_name":"event_file",
"plugin_param":"{\"log_path\":\"../log\"}",
"udf_name":"test1"
},
{
"plugin_name":"event_sql",
"plugin_param":"{\"hostaddr\":\"172.0.0.5\",\"port\":\"24002\",\"user\":\"xxxx\",\"password\":\"xxx\",\"log_path\":\"../log\"}",
"udf_name":"test2"
}
]
}
}' -X POST http://172.0.0.1:18002/kunlun_cdc
Special note: When using add_dump_table, if need_allldump is not passed, the default value is 0, which means no full export will be performed.
4.1.1.6 Support users to manually perform full data synchronization, then CDC initiates incremental synchronization tasks and passes in the table structures that already exist in the current dump task.
curl -d '
{
"version":"1.0",
"job_id":"",
"job_type":"add_dump_table",
"timestamp":"1435749309",
"user_name":"kunlun_test",
"paras":{
"meta_db":"172.0.0.1:28001,172.0.0.2:28001,172.0.0.3:28001",
"meta_user":"xxx",
"meta_passwd":"xxxx",
"dump_db_type":"kunlunbase",
"cluster_name":"cluster_xxx_xxx",
"dump_tables":"postgres_$$_public.t1,postgres_$$_public.t2",
"need_alldump":"0",
"table_structure_file":"xxxx.txt",
"shard_params":[{
"binlog_file":"xxx",
"binlog_pos":"899",
"gtid_set":"xxxx"
}],
"output_plugins":[
{
"plugin_name":"event_file",
"plugin_param":"{\"log_path\":\"../log\"}",
"udf_name":"test1"
},
{
"plugin_name":"event_sql",
"plugin_param":"{\"hostaddr\":\"172.0.0.5\",\"port\":\"24002\",\"user\":\"xxxx\",\"password\":\"xxx\",\"log_path\":\"../log\"}",
"udf_name":"test2"
}
]
}
}' -X POST http://172.0.0.1:18002/kunlun_cdc
Special Note: If the table structure file is not entered, CDC will by default obtain the table structure from the information_schema of the current dump database. Therefore, there must be no changes to the table structure from the completion of the full dump until the CDC dumps to this table. The format of the table structure file is:
CREATE TABLE `db_name`.`table_name1` (
....
);
CREATE TABLE `db_name`.`table_name2` (
....
);
4.1.1.7 Support users to initiate data export from TDengine via CDC
curl -d '
{
"version":"1.0",
"job_id":"",
"job_type":"add_dump_table",
"timestamp":"1435749309",
"user_name":"kunlun_test",
"paras":{
"meta_db":"172.0.0.1:6030",
"meta_user":"xxx",
"meta_passwd":"xxxx",
"dump_db_type":"tdengine",
"cluster_name":"cluster_xxx_xxx",
"dump_tables":"xxx.t*",
"thread_num":"5",
"delay_subscribe_ts":"600",
"shard_params":[{
"binlog_file":"xxx.t1",
"binlog_pos":"ts>=\"2020-08-15 12:00:00.000\"",
"gtid_set":"ts"
},{
"binlog_file":"xxxx.t2",
"binlog_pos":"ts>=\"2020-08-15 12:00:00.000\" and ts < \"2024-01-01 01:00:00.000\"",
"gtid_set":"ts"
}],
"output_plugins":[
{
"plugin_name":"event_file",
"plugin_param":"{\"log_path\":\"../log\"}",
"udf_name":"test1"
},
{
"plugin_name":"event_sql",
"plugin_param":"{\"hostaddr\":\"172.0.0.5\",\"port\":\"24002\",\"user\":\"xxxx\",\"password\":\"xxx\",\"log_path\":\"../log\"}",
"udf_name":"test2"
}
]
}
}' -X POST http://172.0.0.1:18002/kunlun_cdc
Special note: 1. The tables that need to be dumped must have a timestamp field. 2. If there are multiple tables in dump_tables, shard_params needs to specifically indicate the table (database name.table name) in the binlog_file field, the time period of the dump in the binlog_pos field, the timestamp field name of the dump table, and in gtid_set. 3. If the dump is a period of time, the right end must be non-closed. 4. thread_num indicates how many threads are started simultaneously to pull data from TDengine in parallel 5. delay_subscribe_ts indicates the interval in seconds between the latest data point that CDC can fetch and the current time. For example, if set to 600, CDC will fetch data up to the position corresponding to the current time minus 600 seconds and then stop.
4.1.2 Delete dump data task (asynchronous interface)
curl -d '
{
"version":"1.0",
"job_id":"",
"job_type":"del_dump_table",
"timestamp":"1435749309",
"user_name":"kunlun_test",
"paras":{
"meta_db":"172.0.0.1:28001,172.0.0.2:28001,172.0.0.3:28001",
"cluster_name":"cluster_xxxx_xxx",
"dump_db_type":"mysql|mariadb",
"dump_tables":"postgres_$$_public.t1,postgres_$$_public.t2"
}
}
' -X POST http://172.0.0.1:18002/kunlun_cdc
Special Note: A new parameter dump_db_type has been added, with possible values of mysql or mariadb. If this parameter is not provided, it defaults to mysql.
4.1.3 Obtain the Current CDC Cluster Primary Node (Synchronous Interface)
curl -d '
{
"version":"1.0",
"job_id":"",
"job_type":"get_leader",
"timestamp":"1435749309",
"user_name":"kunlun_test"
}
' -X POST http://172.0.0.1:18002/kunlun_cdc
4.1.4 Obtaining the synchronization target plug-ins (synchronization interfaces) supported by the current CDC cluster
curl -d '
{
"version":"1.0",
"job_id":"",
"job_type":"list_support_plugins",
"timestamp":"1435749309",
"user_name":"kunlun_test"
}
' -X POST http://172.0.0.1:18002/kunlun_cdc
4.1.5 Get all dump tasks in the current CDC cluster (synchronous interface)
curl -d '
{
"version":"1.0",
"job_id":"",
"job_type":"list_dump_jobs",
"timestamp":"1435749309",
"user_name":"kunlun_test"
}
' -X POST http://172.0.0.1:18002/kunlun_cdc
4.1.6 Get the synchronization status of a specific dump task (synchronization interface)
curl -d '
{
"version":"1.0",
"job_id":"",
"job_type":"get_job_state",
"timestamp":"1435749309",
"user_name":"kunlun_test",
"paras":{
"meta_db":"172.0.0.1:28001,172.0.0.2:28001,172.0.0.3:28001",
"cluster_name":"cluster_xxx_xxx",
"dump_tables":"postgres_$$_public.t1,postgres_$$_public.t2"
}
}
' -X POST http://172.0.0.1:18002/kunlun_cdc
4.1.7 Interface for Getting Asynchronous Task Status (Synchronous Interface)
curl -d '
{
"version":"1.0",
"job_id":"xxx", --需要查询任务job_id
"job_type":"get_state",
"timestamp":"1435749309",
"user_name":"kunlun_test"
}
' -X POST http://172.0.0.1:18002/kunlun_cdc
4.1.8 Interface for Obtaining CDC Cluster Configuration Information (Synchronous Interface)
curl -d '
{
"version":"1.0",
"job_id":"",
"job_type":"list_cdc_conf",
"timestamp":"1435749309",
"user_name":"kunlun_test"
}
' -X POST http://172.0.0.1:18002/kunlun_cdc
4.2 Configuration via xpanel
4.2.1 Reporting CDC Cluster to xpanel
Click CDC service, add button
Successfully reported to CDC service.
4.2.2 Add CDC Task
Click the CDC task, then the Add button
Set the dump task data source and export from the open-source MySQL.
Set synchronization point data information
Click to confirm and save
Set up data synchronization to the target source, and multiple target sources can be configured.
- Configure data JSON file
Click to confirm and save
- Synchronize configuration to KunlunBase
Click to confirm and save
Click to confirm and issue the task
4.2.3 Delete CDC Task
Find the corresponding business from the xpanel CDC task page
Click delete and enter the verification code to confirm to initiate the delete task
4.2.4 Check CDC Synchronization Status
Find the corresponding business from the xpanel CDC task page
Click on details to view the specific synchronization status
05 Prerequisites for Using KlustronDB CDC
The source MySQL DB to be dumped requires the following configuration
- gtid_mode=ON, otherwise data loss cannot be avoided
- binlog_row_metadata=FULL, otherwise CDC cannot work properly
- binlog_row_image=FULL, it is recommended to set it to FULL.
06 Built-in Plugin Description:
6.1 The event_file plugin directly converts the data captured by CDC into JSON and writes it to a file. When developing custom plugins, you can refer to the output of the event_file plugin.
The input parameter of the event_file plugin is the file into which the JSON content is specifically written. When adding a dump task, the input parameters are specified in the output_plugins field, for example
"plugin_name":"event_file", -- Plugin name "plugin_param":"{"log_path":"xxx","log_name":"xxxx","log_size":"500"}", -- Indicates the file location for writing JSON content, file name, and size of each file "udf_name":"test1" -- Extended field
6.2 event_sql supports converting CDC-captured data into SQL and writing it to the target database. The target database can be a KlustronDB cluster or open-source MySQL. By default, it writes to the KlustronDB cluster; if open-source MySQL is needed, you must add is_kunlun=0 in plugin_param.
event_sql plugin input parameters, when adding a dump task, input parameters in the output_plugins field, for example
"plugin_name":"event_sql", --- Plugin name "plugin_param":"{"hostaddr":"172.0.0.2","port":"24002","user":"abc","password":"abc","log_path":"../log"}", --- Input parameters for event_sql plugin "udf_name":"test2" --- Expansion field
Special note: If you use event_sql to write source data into a KlustronDB node, the plugin_param should be configured with the MySQL port of the KlustronDB Kluscomp instance, not the PostgreSQL port.
6.3 parallel_sql supports exporting CDC captured data to SQL and writing it to the target database in parallel. Currently, it is designed for table-level parallelism, meaning that within the same transaction, multiple different tables can be written to the target database in parallel. The degree of parallelism is determined by user configuration. It also supports database-table mapping. The parallelism parameter is thread_num, which defaults to 5 if not configured.
event_sql plugin input parameters, when adding a dump task, input parameters in the output_plugins field, for example
"plugin_name":"parallel_sql", --- Plugin name "plugin_param":"{"hostaddr":"172.0.0.2","port":"24002","user":"abc","password":"abc","log_path":"../log","remap_rules":"test1.t1=>test2.b1,test1.t2=>test2.b2"}", --- parallel_sql plugin input parameters "udf_name":"test2" --- expanded field
Special note: The plugin parameter plugin_param is in JSON format. When using the API, you need to pay attention to whether the JSON format is valid.
6.4 event_kafka supports converting CDC captured data into JSON and writing it into Kafka.
event_kafka plugin input parameters, input parameters in the output_plugins field when adding a dump task, for example
"plugin_name":"event_kafka", --- Plugin Name
"plugin_param":"{"brokers":"172.0.0.2:9002","log_path":"../log", "assign_topic_type":"channel", "assign_topics":[{"name":"xxx", "kakfa_topic":"xxxx"}]}" --- event_kafka plugin input parameters, users can set the Kafka topic. The way to set the topic is as follows:
The user specifies the topic by setting the assign_topic_type field, supporting three channels: DbTable, Db. If assign_topic_type is not set, it defaults to asking the channel.
6.4.1 channel mode If CDC is currently in global mode, the topic name is specified according to the cluster_name in the added dump task.
For example: when adding a dump task and specifying the cluster_name as test_abc, the corresponding topic name for manual configuration will be abc.
If CDC is currently in shard mode, when exporting data from the KlustronDB cluster, the topic name is specified according to the cluster_name added in the dump task plus the shard name.
For example: when adding a dump task, specify the cluster_name as test_abc, and if there are 3 shards in the current cluster, then when manually configuring the topic, test_abc-shard_1 corresponds to
The topic named abc1, test_abc-shard_2 corresponds to the topic named abc2, test_abc-shard_3 corresponds to the topic named abc3, and multiple shards also specify the same topic name.
6.4.2 Table name mapping or database name mapping If assign_topic_type is a library table pattern mapping, the name format is db_name.table_name. If it is a database name mapping, the name is db_name.
"udf_name":"test2" --- Expansion field
6.5 event_es supports writing CDC captured data into ES event_es plugin input parameters, input parameters in the output_plugins field when adding a dump task, for example
"plugin_name":"event_es", --- Plugin name "plugin_param":"{"es_url":"172.0.0.2:24002","es_index":"xxx","es_version":"v7|v8","log_name":"xxxx","log_path":"../log"}", --- event_es plugin input parameters "udf_name":"test2" --- Scaling field
Special note: The event_es plugin currently only supports ES versions v7 or v8.
6.6 event_mongodb supports writing data captured by CDC into MongoDB event_mongodb plugin input parameters, input parameters in the output_plugins field when adding a dump task, for example
"plugin_name":"event_mongodb", --- Plugin name "plugin_param":"{"mongo_url":"172.0.0.2:24002","mongo_user":"xxx","mongo_passwd":"xxx","log_name":"xxxx","log_path":"../log"}",--- Input parameters for event_mongodb plugin "udf_name":"test2" --- Expansion field
Special note: If the target Mongo does not have an account/password set, there is no need to provide the mongo_user and mongo_passwd fields.
6.7 event_tdengine supports writing CDC-captured data into TDengine event_tdengine plugin input parameters, input parameters in the output_plugins field when adding a dump task, for example
"plugin_name":"event_tdengine", --- Plugin name "plugin_param":"{"hostaddr":"172.0.0.2","port":"24002","user":"xxx","password":"xxx","log_name":"xxxx","log_path":"../log", "remap_rules":"xxx.=>yyy.","sub_tables":[{"dt":"xxx.bbb", "tags":"aa,bb,cc, ...", "gen_regex":"abc_${ccc}"}]}"", --- Input parameters for event_tdengine plugin "udf_name":"test2" --- Scaling field
Special Note:
- Users need to manually create super tables/normal tables in TDengine
- CDC creates subtables under a super table according to the configuration rules. The sub_tables field is configured, where dt is the specific super table name (database name.table name), tags are the specific tags that the super table has, and gen_regex is the format for generating subtable names. For example: test_${aa} indicates that the generated subtable name is test_ combined with the aa field value in tags.
6.8 event_rabbitmq supports writing CDC-captured data into RabbitMQ Input parameters for the event_rabbitmq plugin, for example when adding a dump task, enter the parameters in the output_plugins field
"plugin_name":"event_rabbitmq", --- Plugin name "plugin_param":"{"host":"172.0.0.2:24002","vhost":"xxx","exchange":"xxxx","login_user":"xxx","login_password":"v7|v8","log_name":"xxxx","log_path":"../log"}, --- Input parameters for event_rabbitmq plugin "udf_name":"test2" --- Expansion field
Special Note: Supports specifying routing_keys in the format "assign_routing_keys":[{"channel_name":"xxx", "routing_key":"xxxx"}], where channel_name corresponds to the cluster_name value in the dump task, and routing_key is user-specified. If routing_keys are not supported, the default value in CDC is cdc-rabbitmq-"$cluster_name".
6.9 redis_sql supports importing Redis data into relational databases (MySQL/Postgres, etc.) redis_sql plugin input parameters, when adding a dump task, input parameters in the output_plugins field, for example
"plugin_name":"redis_sql", --- Plugin name "plugin_param":"{"host":"172.0.0.2","port":"12345","user":"xxxx","password":"xxx","database":"xxxx","table_name":"xxxx","log_name":"xxxx","log_path":"../log"}, --- event_rabbitmq plugin input parameters "udf_name":"test2" --- scaling field
Special note: Two tables must be created in the target database. The table names can vary, but the table structure cannot be modified. If the table names change, they need to be specified in the input parameter table_name.
CREATE TABLE `redis_repl_tb` (
`id` int unsigned NOT NULL AUTO_INCREMENT,
`db_id` int unsigned default '0',
`op_record` text,
PRIMARY KEY (`id`)
);
CREATE TABLE `kunlun_cdc_dump_state` (
`job_id` varchar(64) not NULL,
`repl_info` varchar(1024) default '',
PRIMARY KEY(`job_id`)
);
Special Note: Obtain the current CDC mode by calling the api list_cdc_conf interface. The format is as follows:
{"attachment":{"dump_binlog":{"allow_dump_shard_master":"0","dump_shard_max_delay":"1800","query_shard_state_interval":"10","report_cdc_sync_state_interval":"5","pending_binlog_event_num":"1000","dump_mode":"global","binlog_event_queue_len":"4096","reserve_binlog_log_dir":"../data/reserve_dir"},"cdc_ha":{"group_member":"127.0.0.2:18001,127.0.0.1:18001,127.0.0.3:18001","data_dir":"../data/paxosdata","log_dir":"../data/paxoslog"}},"version":"1.0","error_code":"0","error_info":"Ok","status":"Done"}
The content of the dump_mode field
7. KlustronDB CDC Deployment Instructions
- Obtain the kunlun_cdc installation package and extract it to the target directory.
- Go to the conf directory and modify the kunlun_cdc.cnf file
# Copyright (c) 2022 ZettaDB inc. All rights reserved.
# This source code is licensed under Apache 2.0 License,
# combined with Common Clause Condition 1.0, as detailed in the NOTICE file.
[Base_Config]
############################################
# base config
local_ip = 172.0.0.1
http_port = 18002
log_file_path = ../log/kunlun_cdc
log_file_size = 500
基础配置
Local_ip为本机ip地址
http_port为KlustronDB CDC监听到端口
log_file_path和log_file_size 为kunlun_cdc日志配 置
[Binlog_Config]
############################################
# connect cluster shards strategy
allow_dump_shard_master = 0
dump_shard_node_max_delay = 1800
loop_query_shard_state = 10
loop_report_cdc_sync_state = 5
binlog_msg_queue_len = 1024
pending_binlog_event_num = 1000
reserve_binlog_event_dir = ../data/reserve_dir
kunlun_cdc dump binlog相关配置
allow_dump_shard_master是否允许从shard主上dump，默认为0，该配置在从kunlunbase集群dump数据时有效
dump_shard_node_max_delay 为dump的shard 备节点允许的最大延迟，如果该节点延迟大约配置值，则kunlun_cdc自动选择shard其他备节点。该配置在从kunlunbase集群dump数据时有效
loop_query_shard_state 查询dump shard 状态时间间隔
loop_report_cdc_sync_state dump表状态固化时间间隔
pending_binlog_event_num 表示在xa事务中，CDC默认可以缓存多少条binlog event，当大于配置值时，CDC将缓存binlog event消息写入磁盘
reserve_binlog_event_dir 表示CDC可以将缓存binlog event消息写入磁盘位置
[HA_Config]
############################################
# config paxos
ha_group_member = 172.0.0.1:18001,172.0.0.2:18001,172.0.0.3:18001
server_id = 2
paxosdata_dir = ../data/paxosdata
paxoslog_dir = ../data/paxoslog
paxosdata_compress = 0
paxosdata_write_buffer_size = 2
paxosdata_max_write_buffer_number = 2
paxosdata_min_writer_buffer_number_to_merge = 1
paxosdata_max_backgroup_compactions = 6
paxosdata_max_bytes_for_level_base = 64
paxosdata_target_File_size_base = 64
paxosdata_level0_slowdown_writes_trigger = 12
paxosdata_level0_stop_writes_trigger = 16
paxosdata_block_cache_size = 5
paxosdata_block_size = 64
paxosdata_bloom_filter_bits_per_key = 10
paxosdata_block_based_bloom_filter = 0
klustrondb CDC高可用配置
ha_group_member klustrondb CDC集群节点ip:port，个数要求为奇数
server_id为该节点ip在ha_group_member中位置，例如该机器为172.0.0.2，则server_id=2
[Plugin_Config]
############################################
plugin_so = event_file,event_sql
KlustronDB CDC plugin configuration.