MySQL to Kafka

NineData Data Replication supports full and incremental replication from MySQL-compatible data sources to Kafka.

Overview

Use this workflow to deliver MySQL changes to Kafka as JSON messages for downstream stream processing, analytics, monitoring, and data distribution workloads.

Before you begin

• Add the source and target data sources to NineData. For details, see Add Data Source.

• The supported MySQL source versions range from 5.1 to 9.0.

• The target data source must be Kafka 0.10 or later.

• The source data source must have Binlog enabled, with the following parameters:

  • binlog_format=ROW

  • binlog_row_image=FULL

    tip

    If the source data source is a standby database, enable log_slave_updates to capture the complete Binlog.

Restrictions

• Data replication applies only to user databases. System databases are not replicated. For MySQL data sources, system databases include information_schema, mysql, performance_schema, and sys.

• Before data synchronization, evaluate source and target performance, and run synchronization during off-peak hours when possible. Full data initialization consumes read and write resources on both sides and increases database load.

• Make sure each replicated table has a primary key or unique constraint, and that column names are unique. Otherwise, duplicate rows may be synchronized.

Procedure

1. Sign in to the NineData Console.

2. In the navigation menu, click Replication > Data Replication.

3. On the Replication page, click Create Replication.

4. On the Source & Target tab, configure the source, target, and Kafka delivery settings, and then click Next.

   Parameter	Description
   Name	Enter a meaningful data replication task name for later search and management. Up to 64 characters are supported.
   Source	Select the MySQL data source that contains the data to replicate.
   Target	Select the Kafka data source that receives the replicated data.
   Kafka Topic	Select the Kafka topic that receives the replicated MySQL data.
   Delivery Partition	Select how NineData delivers messages to Kafka partitions. Deliver All to Partition 0: Deliver all messages to partition 0. Deliver to different partition by the Hash value of [databaseName + tableName]: Hash by source database name and table name, and deliver data from the same table to the same partition.
   Type	Select Full or Incremental for the replication task. Full: Replicate all selected source objects and data. This is full data replication. Incremental: After full replication completes, replicate incremental changes based on the source logs. Click the icon to clear operation types that you do not want to replicate during incremental synchronization. Cleared operations are ignored. Note: Expand Target Table Preparation to choose the handling strategy for same-name objects or existing data in the target.
   Incremental Started	Required only when Type is Incremental. From Started: Start incremental replication from the time when the replication task starts. Customized Time: Select a custom time point for incremental replication. Select the time zone that matches your business location. If the time point is earlier than the task start time and DDL operations exist in that period, the task fails.
   Target Table Preparation	Target Table Exists Data Pre-Check Error and Stop Task: Stop the task if the precheck detects existing data in the target topic. Ignore existing target data and append to it.: If the precheck detects existing data in the target topic, ignore that data and append new data. Clear target existing data before write: If the precheck detects existing data in the target topic, delete the existing data before writing new data.

5. On the Objects tab, configure the following parameters, then click Next.

   Parameter	Description
   Replication Objects	Select the objects to replicate. Select Full Instance to replicate all objects in the source database, or select Customized Object, choose objects in the Source Object list, and click > to move them to the Target Object list.
   Blacklist (optional)	Click Add to add a blacklist rule. Select the databases or objects to exclude from replication. Blacklist rules are used to exclude specific databases or objects when Customized Object or Full Instance replication is selected. Left drop-down list: Select the database to add to the blacklist. Right drop-down list: Select one or more objects in the selected database. Leave this field empty to add the entire database to the blacklist. To exclude multiple databases, click Add again to add more rows.

6. On the Mapping tab, configure the columns to replicate to Kafka. By default, NineData replicates all columns in the selected tables.

7. On the Pre-check tab, wait for NineData to complete the precheck. After the precheck passes, click Launch.

   tip

   • If the precheck fails, click Details in the Actions column for the failed item to view the cause. After fixing the issue, click Check Again to run the precheck again.
   • For items whose Result is Warning, fix the issue or ignore the warning when appropriate.

8. On the Launch page, the Launch Successfully message appears, and the replication task starts running. Use these actions to review the task:

   • Click View Details to view the execution status of each replication phase.
   • Click Back to list to return to the Replication task list.

View replication results

1. Sign in to the NineData Console.

2. In the navigation menu, click Replication > Data Replication.

3. On the Replication page, click the ID of the target replication task to open the Details page. The task detail page contains the following areas.

   

   No.	Area	Description
   1	Synchronization Delay	Shows the synchronization delay between the MySQL source and Kafka target. 0 seconds means the Kafka target has caught up with the MySQL source.
   2	Configure Alerts	After alerts are configured, NineData notifies users when the task fails. For more information, see Introduction to operational monitoring.
   3	More	Pause: Pause a task whose status is Running. Terminate: Terminate a task that is not complete or is still being monitored, such as a task in incremental synchronization. A terminated task cannot be restarted. Proceed with caution. If the synchronization objects contain triggers, select the trigger replication option that matches your task. Delete: Delete the task. Deleted tasks cannot be restored. Proceed with caution.
   4	Full Replication (shown when full replication is included)	Shows full replication progress and details. Click Monitor to view full replication metrics. During full replication, click Flow Control Settings on the monitoring page to limit the write rate to the target data source. The unit is MB/s. Click Log to view the full replication execution log. Click to view the latest information.
   5	Incremental Replication (shown when incremental replication is included)	Shows incremental replication metrics. Click Flow Control Settings to limit the write rate to the target data source. The unit is rows per second. Click Log to view the incremental replication execution log. Click to view the latest information.
   6	Modify Object	Shows the modification history of replicated objects. Click Modify Objects to configure replication objects . Click to view the latest information.
   7	Expand	Shows task details, including Type, Replication Objects, and Started.

Appendix 1: Kafka message format

Data replicated from MySQL to Kafka is stored in JSON format. NineData divides MySQL data into multiple JSON objects, and each JSON object is delivered as a Kafka message.

• During full replication, the number of MySQL rows stored in a single message is determined by the message.max.bytesmessage.max.bytes is the maximum message size allowed in the Kafka cluster. The default value is `1000000` bytes (1 MB). Adjust this value in the Kafka configuration file so that each message can store more MySQL rows. Setting this value too high may reduce Kafka cluster performance because Kafka must allocate memory for each message. parameter.
• During incremental replication, a single message stores one MySQL row.

Each JSON object contains the following fields:

Field name	Field type	Description	Example
serverId	STRING	Data source information for the message, in the format <connection_address:port>.	"serverId":"47.98.224.21:3307"
id	LONG	Record ID of the message. This field is globally incremented and can be used to identify duplicate message consumption.	"Id":156
es	INT	Has different meanings in different task phases: Full replication: the start time of the full replication task, represented as a Unix timestamp. Incremental replication: the time of each event in the Binlog.	"es":1668650385
ts	INT	The time when the message was delivered to Kafka, represented as a Unix timestamp.	"ts":1668651053
isDdl	BOOLEAN	Indicates whether the data is DDL. true: The data is DDL. false: The data is not DDL.	"is_ddl":true
type	STRING	Data type. INIT: Full data replication. INSERT: INSERT operation. DELETE: DELETE operation. UPDATE: UPDATE operation. DDL: DDL operation.	"type":"INIT"
database	STRING	Database to which the data belongs.	"database":"database_name"
table	STRING	Table to which the data belongs. If the object in the DDL statement is not a table, the value is null.	"table":"table_name"
mysqlType	JSON	MySQL data types of the fields, represented as a JSON array.	"mysqlType": {"id": "bigint(20)", "shipping_type": "varchar(50)" }
sqlType	JSON	Reserved field. No action is required.	"sqlType":null
pkNames	ARRAY	Primary key names for the record in the Binlog. For DDL records, the value is null. For INIT or DML records, the value is the primary key name of that record.	"pkNames": ["id", "uid"]
data	ARRAY[JSON]	Data delivered from MySQL to Kafka, stored as a JSON array. Full data replication (type = INIT): Stores the full data delivered from MySQL to Kafka. Incremental data replication: Stores Binlog change details. INSERT: Values inserted into each field. UPDATE: Values after the update. DELETE: Values deleted from each field. DDL: Table structure after the DDL operation.	"data": [{ "name": "someone", "phone": "(737)1234787", "email": "someone@example.com", "address": "somewhere", "country": "china" }]
old	ARRAY[JSON]	Records incremental replication details from MySQL to Kafka. UPDATE: Values before the update. DDL: Table structure before the DDL operation. For other operations, the value is null.	"old": [{ "name": "someone", "phone": "(737)1234787", "email": "someone@example.com", "address": "somewhere", "country": "china" }]
sql	STRING	For incremental DDL operations, records the SQL statement for the operation. For other operations, the value is null.	"sql":"create table sbtest1(id int primary key,name varchar(20))"

Appendix 2: Precheck items

Check item	What NineData checks
Source data source connection check	Checks the source gateway status, instance accessibility, and username and password.
Target data source connection check	Checks the target gateway status, instance accessibility, and username and password.
Source database permission check	Checks whether the source database account has the required permissions.
Source standby log update check	Checks whether log_slave_updates is set to ON when the source database is a standby database.
Source and target version check	Checks whether the source and target versions are compatible.
Source Binlog check	Checks whether Binlog is enabled for the source database.
Source Binlog format check	Checks whether the source database binlog_format is ROW.
Source Binlog row image check	Checks whether the source database binlog_row_image is FULL.
Target permission check	Checks whether the Kafka account has access permissions for the topic.
Target data existence check	Checks whether data already exists in the topic.

Related documents

• Introduction to Data Replication