Synchronize data from an OceanBase database to RocketMQ

Background information

Message Queue for Apache RocketMQ is a low-latency, high-concurrency, and highly reliable distributed message middleware developed by Alibaba Cloud based on Apache RocketMQ. The data synchronization feature of Data Transmission Service lets you synchronize data in real time between physical tables in an OceanBase database and a RocketMQ data source. This expands your message processing capabilities.

For information about the data formats for the two tenant types when you synchronize data from an OceanBase database to RocketMQ, see Data format for transferring data to a text-based protocol.

Prerequisites

• Data Transmission Service must have the required access permissions for cloud resources. For more information, see Grant permissions to the Data Transmission Service migration role.

• A database user for the data synchronization task must be created in the source OceanBase database, and the required permissions must be granted to the user. For more information, see Create a database user.

Limits

• Data Transmission Service supports RocketMQ V4.x and V5.x instances, including commercial and community editions.

• Only physical tables can be synchronized. Other objects are not supported.

• During data synchronization, Data Transmission Service supports deleting a table and then creating a new one. This means you can run DROP TABLE and then CREATE TABLE on a synchronized table. Data Transmission Service does not support creating a table by renaming it. This means you cannot run RENAME TABLE a TO a_tmp.

• The names of the tables and columns to be synchronized cannot contain Chinese characters.

• Data Transmission Service supports the migration of only objects whose database, table, and column names contain only ASCII characters. These names cannot contain special characters, such as line breaks, spaces, or .|"'`()=;/&\.

• Data Transmission Service does not support using a secondary OceanBase database as the source.

Notes

• If your OceanBase database version is between V4.0.0 and V4.3.x (excluding V4.2.5 BP1) and you select incremental synchronization, configure the STORED property for generated columns. Otherwise, information about the generated columns will not be saved in the incremental logs. This may cause data anomalies during incremental synchronization.

• When a row that is being updated includes a large object (LOB) column:

  • If the LOB column is one of the updated columns, do not rely on its value before the UPDATE or DELETE operation.

    Data types that currently use LOB columns for storage include JSON, GIS, XML, User Defined Type (UDT), and various TEXT types such as LONGTEXT and MEDIUMTEXT.

  • If the LOB column is not one of the updated columns, its value is NULL both before and after the UPDATE or DELETE operation.

• If a task is unexpectedly interrupted and resumes from a breakpoint, duplicate data from the last minute of processing may exist in the RocketMQ instance. Therefore, your downstream system must be able to deduplicate data.

• If the clocks between nodes are not synchronized, or if the clock on your computer is not synchronized with the server clock, the incremental synchronization latency may be inaccurate.

  For example, if a clock is ahead of the standard time, the latency may be a negative value. If a clock is behind the standard time, it may cause a delay.

• If you select only Incremental Synchronization when you create the data sync task, the local incremental logs of the source database must be retained for more than 48 hours.

  If you select Full Synchronization + Incremental Synchronization when you create the data sync task, the local incremental logs of the source database must be retained for at least 7 days. Otherwise, Data Transmission Service may fail to retrieve the incremental logs. This can cause the sync task to fail or lead to data inconsistency between the source and destination.

Supported source and destination instance types

In the following table, the MySQL tenant of OceanBase Database is referred to as OB_MySQL, and the Oracle tenant of OceanBase Database is referred to as OB_Oracle.

Procedure

1. Log on to the OceanBase Management Console and purchase a data sync task.

   For more information, see Purchase a data sync task.

2. In the Data Transmission Service console, navigate to the Data Synchronization page, find the new sync task, and click Configure.

   

   If you want to reuse the configurations of an existing task, click Reuse Configuration. For more information, see Reuse and clear the configurations of a data sync task.

3. On the Select Source and Destination page, configure the parameters.

   Parameter	Description
   Sync Task Name	The name can contain letters, digits, and Chinese characters. The name cannot contain spaces and must be 64 characters or less.
   Source	If you have already created an OceanBase data source, select it from the drop-down list. If you have not, click Create Data Source in the drop-down list and create one in the dialog box on the right. For details about the parameters, see Create an OceanBase data source.
   Destination	If you have already created a RocketMQ data source, select it from the drop-down list. If not, click Create Data Source in the drop-down list and create one in the dialog box that appears on the right. For details about the parameters, see Create a RocketMQ data source.
   Tags (Optional)	Click the text box and select the target tag from the drop-down list. You can also click Manage Tags to create, modify, and delete tags. For more information, see Manage data synchronization tasks using tags.

4. Click Next. On the Select Synchronization Type page, select a synchronization type for the data synchronization task.

   

   Synchronization types include Full Synchronization and Incremental Synchronization. Incremental Synchronization only supports DML Synchronization (including INSERT, DELETE, and UPDATE). For more information about customizing the configuration, see Customize DDL/DML.

5. Click Next. On the Select Synchronization Objects page, select the objects to synchronize.

   You can select objects to synchronize using two methods: Specify Objects and Matching Rules. This topic describes how to use the Specify Objects method. For more information about how to configure matching rules, see Configure and modify matching rules.

   Note

   If you check DDL Sync in the Select Sync Type step, we recommend selecting sync objects based on matching rules. This ensures that any new objects that match the rules are also synced. If you select sync objects individually, new or renamed objects will not be synced.

   When you synchronize data from an OceanBase database to RocketMQ, you can synchronize multiple tables to multiple topics.

   1. In the Select Sync Objects area, select Specify Objects.

   2. In the selection area, select the objects to be synchronized from the list on the left.

   3. Click >.

   4. In the Map Object To Topic dialog box, search for and select the topic that you want to sync from the Existing Topic drop-down list.

      You can also enter the name of an existing topic and select it from the results.

      

   5. Click OK.

      Data Transmission Service lets you import objects from a text file and perform operations on the destination objects. These operations include changing the topic, setting row filters, and removing a single object or all objects. The structure of the destination objects is Topic > Database > Table.

      Note

      When you use the Matching Rules method to select synchronization objects, the matching rule syntax overrides the renaming capability, and the only available operations are to set filter conditions, select the shard column, and select the columns for synchronization. For more information, see Configure and modify matching rules.

      

      Operation	Description
      Import Objects	In the list on the right side of the selection area, click Import Objects in the upper-right corner. In the dialog box that appears, click OK. Important Importing objects overwrites previous selections. Proceed with caution. In the Import Sync Objects dialog box, import the objects to sync. You can import a CSV file to set row filtering conditions, filter columns, and set sharding columns. For more information, see Download and import sync object configurations. Click Validate. After the validation is successful, click OK.
      Change Topic	DTS lets you change the topic for destination objects. For more information, see Change a topic.
      Settings	DTS supports row filtering using a WHERE clause, and lets you select sharding columns and columns to sync. In the Settings dialog box, you can perform the following operations. In the Row Filter Condition area, enter a standard SQL WHERE clause in the text box to configure row filtering. For more information, see Filter data using SQL conditions. In the Sharding Column drop-down list, select the target sharding column. You can select multiple fields as the sharding column. This parameter is optional. When selecting sharding columns, if there are no special requirements, you can simply select the primary key. If the primary key causes a load imbalance, select a field that is unique and relatively balanced as the sharding column to avoid potential performance issues. The main functions of sharding columns are as follows: Load balancing: If concurrent writes are possible at the destination, the sharding column is used to determine the specific thread for sending a message. Ordering: Because concurrent writes can cause out-of-order issues, DTS ensures that messages with the same sharding column value are received in order. Ordering here refers to the order of changes (the execution order of DML operations on a column). In the Select Columns area, select the columns to sync. For more information, see Column Filtering.
      Remove/Remove All	DTS lets you remove one or more objects that have been temporarily selected for the destination during data mapping. Remove a single sync object In the list on the right side of the selection area, hover over the target object and click the Remove button that appears. Remove all sync objects In the list on the right side of the selection area, click Remove All in the upper-right corner. In the dialog box, click OK to remove all synchronization objects.

6. Click Next. On the Sync Options page, configure the parameters.

   • Full synchronization

     The following parameters are displayed only if you select Full Synchronization on the Choose Synchronization Types page.

     

     Parameter	Description
     Read Concurrency	This parameter specifies the number of concurrent threads for reading data from the source during full synchronization. The maximum value is 512. A high concurrency may put excessive pressure on the source and affect your business.
     Write Concurrency	This parameter specifies the number of concurrent threads for writing data to the destination during full synchronization. The maximum value is 512. A high concurrency may put excessive pressure on the destination and affect your business.
     Full Synchronization Rate Limit	You can decide whether to enable the full synchronization rate limit as needed. If you enable it, set RPS (the maximum number of data rows that can be synchronized to the destination per second during full synchronization) and BPS (the maximum data volume that can be synchronized to the destination per second during full synchronization). Note The RPS and BPS values set here are only for rate limiting. The actual performance of full synchronization is affected by factors such as the source, destination, and instance specifications.

   • Incremental synchronization

     The following parameters are displayed only if you select Incremental Synchronization on the Choose Synchronization Types page.

     

     Parameter	Description
     Write Concurrency	This parameter specifies the number of concurrent threads for writing data to the destination during incremental synchronization. The maximum value is 512. A high concurrency may put excessive pressure on the destination and affect your business.
     Incremental Synchronization Rate Limit	You can decide whether to enable the incremental synchronization rate limit as needed. If you enable it, set RPS (the maximum number of data rows that can be synchronized to the destination per second during incremental synchronization) and BPS (the maximum data volume that can be synchronized to the destination per second during incremental synchronization). Note The RPS and BPS values set here are only for rate limiting. The actual performance of incremental synchronization is affected by factors such as the source, destination, and instance specifications.
     Incremental Synchronization Start Offset	If you select Full Synchronization when you select the synchronization type, you cannot change this parameter. If you do not select Full Synchronization but select Incremental Synchronization when you select the synchronization type, specify a point in time after which you want to synchronize data. The default value is the current system time. For more information, see Set the start offset for incremental synchronization.

   • Advanced Options

     

     Parameter	Description
     Serialization Method	Specifies the message format for data synchronized to RocketMQ. The supported formats are Default, Canal, DataWorks (supports Version 2.0), SharePlex, DefaultExtendColumnType, Debezium, DebeziumFlatten, and DebeziumSmt. For more information, see Data format description. Important Currently, only MySQL tenants of OceanBase databases support Debezium, DebeziumFlatten, and DebeziumSmt.
     Partitioning Rule	The only supported rule for synchronizing source data to RocketMQ is Hash.Hash means that Data Transmission Service uses a hash algorithm to select a RocketMQ queue (MessageQueue) based on the primary key value or the sharding column value.
     Business System Identifier (Optional)	This parameter is displayed only when Serialization Method is set to DataWorks. It is used to identify the source business system of the data so that you can perform custom processing. The length of this business system identifier is limited to 1 to 20 characters.

   • Destination

     

     Parameter	Description
     Enter Producer Group	A producer group identifies a group of producers that can write data to multiple topics.
     Allow Message Tracing	If you allow message tracing, you can trace a message from its sending by the producer to the Message Queue for Apache RocketMQ server, and then to its processing by the consumer. This provides a complete task information trail, including the time and status of each relevant node. The message trace provides strong data support for troubleshooting issues in a production environment.

7. Click Precheck.

   In the Precheck step, Data Transmission Service verifies the connectivity between the source and the destination. If the precheck fails:

   • You can troubleshoot and resolve the issues, and then run the precheck again.

   • You can also click Skip in the Actions column for a failed precheck item. A dialog box appears and explains the specific impact of skipping this operation. To confirm skipping the item, click OK in the dialog box.

8. After the precheck passes, click Start Task.

   If you do not need to start the task right away, click Save. You can then start the task manually on the Sync Task List page or include it in a batch operation. For more information about batch operations, see Perform batch operations on data sync tasks.

   DTS lets you modify the synchronization objects while a data sync task is running. For more information, see View and modify synchronization objects and their filter conditions. After the data sync task starts, it runs based on the selected synchronization types. For more information, see View synchronization details.

If a data sync task fails, typically due to network connection issues or a slow process startup, you can click Resume on the task list or the product page.

References

• Create an OceanBase data source

• Create a RocketMQ data source

• Run batch operations on data sync tasks

• View data sync task details