Migrate data from OceanBase Database to Kafka

This topic describes how to use the Data Migration Service to migrate data from OceanBase Database (including MySQL and Oracle compatible modes) to Kafka.

Background information

Kafka is a widely used high-performance distributed stream computing platform. The data migration service supports real-time data synchronization between OceanBase Database in two compatibility modes and Kafka data sources, extending message processing capabilities. It is widely applied in business scenarios such as building real-time data warehouses, data queries, and report distribution.

Kafka network connection tasks include cloud instance Kafka networks and self-managed Kafka networks. The processes for different types of network connection tasks are as follows.

• Cloud instance Kafka network connection task

  

• Self-managed Kafka network connection task

  

Prerequisites

• You have created a cluster and a tenant in the source OceanBase Database. For more information, see Create an instance and Create a tenant.

• You have created a dedicated database user for data migration on both the source and target sides and granted the required permissions. For more information, see User privileges.

Limitations

• Only users with the project role of Project Owner, Project Admin, or Data Services Admin can create a data migration task.

• Limitations on operations on the source database

  Do not perform DDL operations that modify database or table structures during full migration. Otherwise, the data migration task may be interrupted.

• Currently, data migration supports Kafka V0.9, V1.0, and V2.x, and OceanBase Database (including MySQL and Oracle compatibility modes) V2.x, V3.x, and V4.x.

• The data migration service only supports migrating objects whose database, table, and column names are ASCII-compliant and do not contain special characters (including .|"'`()=;/& and line breaks).

• Data migration supports physical tables and materialized views. Materialized view synchronization is currently supported only from OceanBase Database in MySQL mode to Kafka. The source database must run V4.3.5.6–V4.4.0 or V4.4.2.4 or later.

• During data migration, if you rename a table within the migration scope at the source and the renamed name is not included in the migration objects, that part of the data will not be synchronized to the target Kafka instance.

• Materialized view synchronization has the following limitations:

  • For full database migration or migration using matching rules, incremental DML changes to materialized views created while the task is running are automatically sent to Kafka. However, the DDL statements that create these views are not sent.

  • Only DML changes to materialized views are synchronized. DDL statements that create materialized views are not synchronized.

  • Synchronizing changes resulting from a COMPLETE refresh of a materialized view to Kafka is not supported. This limitation also applies when a materialized view uses REFRESH FORCE and the refresh falls back to COMPLETE.

• Before using materialized view synchronization, verify that the source OceanBase cluster runs V4.3.5.6–V4.4.0 or V4.4.2.4 or later. If you upgrade an existing cluster to a supported version, complete the following steps for that version before synchronizing materialized views with OMS:

  • For clusters upgraded to V4.3.5.6–V4.4.0:

    1. Submit a support ticket to request access to materialized view synchronization in OMS.

    2. Set the data dictionary dump interval to 1 minute:

       CALL DBMS_DATA_DICT.MODIFY_DUMP_INTERVAL(INTERVAL_EXPR);
       

    3. Wait for the dump to run, then execute the following SQL statement to check whether a new SCN has been generated:

       SELECT * FROM CDB_OB_DATA_DICTIONARY_IN_LOG LIMIT 5;
       

    4. After confirming that a new SCN has been generated, set the data dictionary dump interval to an appropriate value, such as 2 hours.

    5. Create or configure an OMS task to synchronize materialized views.

  • For clusters upgraded to V4.4.2.4 or later:

    1. Manually trigger a data dictionary dump before creating or configuring an OMS task:

       CALL DBMS_DATA_DICT.TRIGGER_DUMP;
       

    2. After the dump completes, create or configure an OMS task to synchronize materialized views.

Considerations

• Clock desynchronization between nodes, or between the client terminal and the server, can lead to inaccurate latency reporting during incremental synchronization.

  For example, if a clock is earlier than standard time, the reported latency may be negative. If a clock is later than standard time, it may cause a positive latency.

• In a data migration task where the source is an OceanBase Database and DDL synchronization is enabled, if a RENAME operation occurs on a table at the source, it is recommended that you restart the task to avoid data loss during incremental synchronization.

• When OceanBase Database is version V4.0.0 to V4.3.x (excluding V4.2.5 BP1) and incremental synchronization is selected, configure the STORED attribute for generated columns. Otherwise, the incremental logs will not contain information for the generated columns, which may cause data anomalies during incremental synchronization.

• When an updated row contains a LOB column:

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

    Currently, data types stored in LOB columns include JSON, GIS, XML, UDT (user-defined type), and various TEXT types such as LONGTEXT and MEDIUMTEXT.

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

• When resuming a task from a checkpoint after an unexpected interruption, some duplicate data (from within the last minute) may exist in the Kafka instance. Therefore, downstream systems must be capable of deduplication.

• When synchronizing data from OceanBase Database to Kafka, if the source fails to execute a unique index creation statement, Kafka will consume both the create DDL and delete DDL statements. If the index creation DDL statement sent downstream fails to execute, ignore this exception.

• If you configure only Incremental Synchronization when creating a data migration task, the local incremental logs in the source database must be retained for more than 48 hours.

  If you configure both Full Migration and Incremental Synchronization when creating a data migration task, the local incremental logs in the source database must be retained for at least 7 days. Otherwise, the data migration task may fail due to inability to obtain incremental logs, potentially leading to data inconsistency between the source and target.

• If you perform incremental synchronization with OceanBase Database as the source and the database table objects contain UDT columns, the incremental synchronization task may fail due to unsupported UDT columns.

Supported source and target instance types

In the following table, the instance types supported by OceanBase MySQL Compatible and OceanBase Oracle Compatible include Dedicated (Transactional) and Self-managed Database.

Procedure

1. Create a data migration task.

   

   1. Log in to the OceanBase Cloud console.

   2. In the left-side navigation pane, click Data Services > Data Migration.

   3. On the Data Migration page, click the Data Migration tab.

   4. On the Data Migration tab, click Create Task in the upper-right corner.

2. In the Edit Task Name text box, enter a custom name for the migration task.

   We recommend that you use a combination of Chinese characters, numbers, and letters. The name cannot contain spaces and must not exceed 64 characters in length.

3. On the Configure Source & Target page, configure the parameters.

   1. In the Source section, configure the parameters.

      If you need to reference an existing data source, you can click Quick Fill next to Source and select the target data source from the drop-down list. After selection, the configurations for the source area will be automatically populated. If you want to save the current configuration as a new data source, click the Save icon in the upper-right corner of the Save section.

      You can also click Manage Data Source in the Quick Fill drop-down list to go to the Data Source page to view and manage data sources. This page provides unified management for different types of data sources. For more information, see the documentation in the Data sources module.

      Parameter	Description
      Cloud Provider	Currently supports Alibaba Cloud, AWS, and Google Cloud.
      Database Type	Select a database type. Select OceanBase MySQL Compatible or OceanBase Oracle Compatible as the source type.
      Instance Type	Select an instance type. Currently, Dedicated (Transactional) and Self-managed Database are supported.
      Region	Select a region. Select the region where the source database is located.
      Connection Type	Select a connection type. Valid values: Endpoint and Public IP. If you select Endpoint as the connection method, you must first add the account ID displayed on the page to the allowlist of your endpoint service to allow connections from this account's endpoint to the endpoint service. For more information, see the Select Private Network Connection section. If you select Public IP as the connection method, you must first add the data source IP address displayed on the page to the allowlist of the OceanBase database instance to ensure connectivity. For more information, see the Select Public Network Connection section. Note This parameter is displayed only when the instance type is set to Self-managed Database. After you select the regions for the source and target, the page will display the data source IP address that needs to be added to the allowlist.
      Connection Information	Select a connection type. This parameter is displayed only when the instance type is set to Self-managed Database. If you select Connection Type as Endpoint, enter the endpoint service name. If you select Connection Type as Public IP, enter the IP address and port number of the database host.
      Instance	Select an instance. The ID or name of the instance where the OceanBase database is located. You can view the ID or name of the target instance on the Cluster List page. Note When the cloud provider is Alibaba Cloud, you can also select an Alibaba Cloud root account instance for cross-account authorization. For more information, see Alibaba Cloud Account Authorization.
      Tenant	Select a tenant. The ID or name of the OceanBase database tenant. You can expand the target instance on the Instances page to view the ID or name of the target tenant under that instance.
      Database Account	The username of the OceanBase database user used for data migration.
      Password	The password of the database user.

      When the instance type is Self-managed Database, if you need to perform schema migration and incremental synchronization, configure the parameters in the Advanced Settings section.

      

      • If you need to select Schema Migration or Incremental Synchronization in the Select Type & Objects step, enable the sys Tenant Account and configure the following parameters.

        Parameter	Description
        Sys Account	The name of the sys user. This user is mainly used to read incremental logs and database object structure information from OceanBase databases. Create it under the sys tenant of the business cluster.
        Password	The password of the sys user.

      • If you need to select Incremental Synchronization in the Select Type & Objects step, enable OBLogProxy and fill in the OBLogProxy connection information.

        Note

        Incremental synchronization is supported only when both the sys tenant account and OBLogProxy are enabled.

        OBLogProxy connection information refers to the incremental log proxy service of OceanBase Database. It provides access and management capabilities for real-time incremental projects as a service, facilitating applications' access to incremental logs of OceanBase Database. It also meets the demand for subscribing to incremental logs under network isolation. The format is OBLogProxy IP:OBLogProxy Port.

   2. In the Target section, configure the parameters.

      If you need to reference a created and saved data source, click Quick Fill on the right side of Target and select the target data source from the drop-down list. After selection, the configurations in the target area will be automatically populated. If you need to save the current configuration as a new data source, click the Save icon in the upper-right corner of the Target section.

      You can also click Manage Data Source in the Quick Fill drop-down list to go to the Data Source page to view and manage data sources. This page provides unified management for different types of data sources. For more information, see Data sources.

      Parameter	Description
      Cloud Provider	Currently supports Alibaba Cloud, AWS, and Google Cloud. You can choose the same cloud provider as the source or perform cross-cloud data migration. Note The cross-cloud data migration feature is not currently enabled by default. To use this feature, contact OceanBase Cloud technical support.
      Database Type	Select Kafka as the target.
      Instance Type	Supports Cloud Instance and Self-managed Kafka.
      Region	Select the region where the target instance is located.
      Connection Type	Includes Endpoint and Public IP. For Alibaba Cloud, AWS, and Google Cloud, you can also select Peering Connection. If you select Endpoint, add the account ID displayed on the page to the allowlist of your Endpoint Service to allow connections from this account to the Endpoint Service. For more information, see Select Private Network Connection. If Cloud Vendor is set to AWS and Require Enable is selected for the Require Accept to Use Endpoint parameter when creating the Endpoint Service, the data migration service prompts you to open the AWS console and perform the Accept endpoint connection request operation for the incoming request the first time it accesses the private connection. If the cloud provider is set to Google Cloud, add the authorized project to Published Services. After authorization, no further manual authorization is required when you test the data source connection. If you select Public IP, add the data source IP address displayed on the page to the Kafka instance's allowlist to ensure connectivity. For more information, see Select Public Network Connection. Note The data source IP address to be added to the allowlist is displayed on the page only after you have selected the regions for both the source and target. If you select Peering Connection, choose an available peering connection from the drop-down list. Add the IP address displayed on the page to the Kafka instance's inbound security group rules or allowlist to ensure connectivity. For more information, see AWS peering connection, Alibaba Cloud peering connection, and Google Cloud peering connection.
      Peering Connection ID	This parameter is displayed only when Connection Type is set to Peering Connection. Select an available peering connection from the drop-down list. The list displays only connections that match the current cloud provider and region and are in the Connected state. It also displays the corresponding VPC ID and CIDR block. If no connection is available, click Create to open a new tab and create a data service peering connection.
      Connection Information	If Connection Type is set to Endpoint, enter the Endpoint Service name and Kafka access point. If Connection Type is set to Public IP, enter the Kafka access point. When Connection Type is set to Peering Connection, enter the Kafka endpoint.

      If the Kafka data source has not completed security authentication configuration, the data migration task using this source or target may fail. The configuration for the security authentication section is as follows.

      1. Click Upload File to upload the trust certificate with the .jks extension.

      2. Based on your business requirements, decide whether to enable SASL. If enabling SASL, configure the following parameters.

         Parameter	Description
         Authentication Method	Supports PLAIN, SCRAM-SHA-256, and SCRAM-SHA-512. The PLAIN authentication method is simpler but does not allow dynamic user changes and configures the username and password in plaintext, which is not secure. SCRAM (Salted Challenge Response Authentication Mechanism) addresses security issues by implementing the traditional username and password authentication mechanism. Kafka supports SCRAM-SHA-256, which can be used together with TLS for security authentication. This authentication method enables dynamic user changes, with user data stored in Zookeeper. Before starting a broker, communication with Zookeeper is required to create a communication user for the broker. However, this method requires plaintext configuration of the username and password. Kafka supports SCRAM-SHA-512, which can be used together with TLS for security authentication.
         Username	The username used for data migration.
         Password	The password for the data migration user.

4. Click Test and Continue.

5. On the Select Type & Objects page, configure the parameters.

   1. In the Migration Type section, select the migration type for the current data migration task.

      Migration Type includes Full Migration and Incremental Synchronization.

      Parameter	Description
      Full Migration	After a full migration task starts, the data migration service migrates the existing data from the source database tables to the corresponding tables in the target database.
      Incremental Synchronization	After an incremental synchronization task starts, it synchronizes the changed data (inserts, modifications, or deletions) from the source database to the corresponding tables in the target database. Incremental Synchronization includes DML Synchronization and DDL Synchronization, which you can customize as needed. For details, see Customize DML/DDL.

   2. In the Select Migration Objects section, configure the method for selecting migration objects.

      You can select migration objects by using either the Specify Objects or Match by Rule method.

      

   3. In the Select Migration Scope section, select the objects to migrate.

      • When you select Specify Objects, data migration supports Table-level and Database-level. Database and table migration allows you to select one or more tables or views from a database as migration objects. Full database migration allows you to select an entire database as the migration object. If you select database and table migration for a database, that database cannot be used for full database migration. Similarly, if you select full database migration for a database, that database cannot be used for database and table migration.

        When migrating data from OceanBase Database to Kafka, you can migrate multiple tables to multiple Topics.

        In the object selection panel, regular tables are listed under Table. If the source database version supports materialized views, Materialized view is also displayed. Expand it to view and select materialized views in the current database.

        1. Select the objects to migrate in the left-side selection area.

        2. Click >.

        3. In the Map Objects to Topic dialog box, search for and select the target Topic in the Existing Topics drop-down list.

        4. Click OK.

        Data migration supports importing objects via text and allows you to perform operations on the target objects, such as changing the Topic, setting row filters, and removing single or all objects.

        Note

        When you select Full Database Migration:

        • The right-side list only displays the database name; it does not support displaying specific objects.

        • If you select the Incremental Synchronization - DDL Synchronization migration type, newly added tables at the source can be synchronized to the target.

        Operation	Description
        Import Objects	In the right-side list of the selection area, click Import Object in the upper-right corner. For details, see Import migration objects.
        Change Topic	Data migration supports changing the Topic for target objects. For details, see Change Topic.
        Settings	Data migration supports filtering rows using WHERE conditions, and allows you to select sharding columns and columns to synchronize. In the Settings dialog box, you can perform the following operations. In the text box of the Row Filters section, enter a WHERE condition to filter rows. For details, see Filter data with SQL conditions. In the Sharding Columns drop-down list, select the target sharding column. You can select multiple fields as sharding columns. This parameter is optional. When selecting sharding columns, if there are no special circumstances, the primary key is the default choice. If there is uneven load on the primary key, select a unique identifier field with relatively balanced load as the sharding column to avoid potential performance issues. The main purposes of sharding columns are as follows: Load balancing: When concurrent writes are possible at the target, sharding columns help determine which specific thread is used to send messages. Orderliness: To address potential out-of-order issues caused by concurrent writes, data migration ensures that messages received by users are ordered when the values of the sharding columns are the same. Here, order refers to the execution order of DML operations on a column. In the Select Columns section, select the columns to synchronize. For details, see Column filtering.
        Remove/Clear All	Data migration supports removing one or more temporarily selected objects to the target during data mapping. Remove a single migration object In the right-side list of the selection area, click the Remove icon next to the target object to remove it. Remove all migration objects In the right-side list of the selection area, click Clear All in the upper-right corner. In the dialog box, click OK to remove all migration objects.

      • To select Match by Rule, see Configure matching rules for database to message queue.

        When the source meets the requirements for materialized view synchronization, matching rules can match both tables and materialized views. When you create a matching rule, both Table and Materialized view are selected by default. You can also select just one object type. Matching results are grouped by object type.

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

   • Full migration

     The following parameters are displayed only if you select Full Migration in the Select Type & Objects step.

     Parameter	Description
     Read Concurrency Configuration	This parameter specifies the number of concurrent reads from the source during the full migration phase. The maximum limit is 512. A high concurrency may cause excessive load on the source and affect business operations.
     Write Concurrency Configuration	This parameter specifies the number of concurrent writes to the target during the full migration phase. The maximum limit is 512. A high concurrency may cause excessive load on the target and affect business operations.
     Limit Full Migration Rate	You can decide whether to enable the full migration rate limit based on your actual needs. If enabled, set the Source Read RPS (the maximum number of rows that can be read from the source per second during full migration), Source Read BPS (the maximum traffic that can be read from the source per second during full migration), Target Write RPS (the maximum number of rows that can be written to the target per second during full migration), and Target Write BPS (the maximum traffic that can be written to the target per second during full migration). Note The RPS and BPS settings here are only for rate limiting. The actual achievable performance during full migration is subject to factors such as the source, target, and instance specifications.

   • Incremental Synchronization

     The following parameters are displayed only if you select Incremental Synchronization in the Select Type & Objects step.

     Parameter	Description
     Write Concurrency Configuration	This parameter specifies the number of concurrent data writes to the target during incremental synchronization. The maximum value is 512. Excessively high concurrency may overwhelm the target and impact your business.
     Incremental Migration Rate Limit	You can decide whether to enable the incremental synchronization rate limit based on your actual needs. If enabled, set the target write RPS (the maximum number of data rows that can be synchronized to the target per second during incremental synchronization) and target write BPS (the maximum traffic that can be synchronized to the target per second during incremental synchronization). Note The RPS and BPS settings here are only for rate limiting. The actual achievable performance during incremental synchronization is subject to factors such as the source, target, and instance specifications.
     Incremental Synchronization Start Timestamp	If you selected Full Migration as the migration type, this parameter is not displayed. If you did not select Full Migration but chose Incremental Synchronization, specify a point in time after which data should be migrated. The default is the current system time. For details, see Set an incremental synchronization timestamp.

   • Advanced Options

     Parameter	Description
     Serialization Method	Controls the message format for migrating data to Kafka. Currently supported methods include Default, Canal, DataWorks (V2.0), SharePlex, DefaultExtendColumnType, Debezium, DebeziumFlatten, DebeziumSmt, and Avro. For details, see Data formats. Note Currently, only the MySQL-compatible mode of OceanBase Database supports Debezium, DebeziumFlatten, DebeziumSmt, and Avro. When DataWorks is selected, DDL synchronization does not support COMMENT ON TABLE or ALTER TABLE…TRUNCATE PARTITION.
     Partitioning Rule	Specifies the rule for synchronizing data from OceanBase Database to a Kafka Topic. Currently supported values are Hash, Table, and One. For delivery and example DDL statements in different scenarios, see the descriptions below the table. Hash indicates that data migration uses a specific hash algorithm to select a partition of the Kafka Topic based on the primary key value or sharding column value. Table indicates that data migration delivers all data from a table to the same partition, using the table name as the hash key. One indicates that JSON messages are delivered to a specific partition under the topic to maintain order.
     Business System Identifier (Optional)	This parameter is displayed only when Serialization Method is set to DataWorks. It identifies the source business system of the data for subsequent custom processing. The length of this identifier must be 1 to 20 characters.

     The following table describes DDL statement delivery in different scenarios.

     Partitioning Rule	DDL Statement Involves Multiple Tables (e.g., RENAME TABLE)	DDL Statement Cannot Identify Related Tables (e.g., DROP INDEX)	DDL Statement Involves a Single Table
     Hash	The DDL statement is delivered to all partitions of the topic where the related tables are located. For example, if a DDL statement involves tables A, B, and C, where A is in Topic 1, B is in Topic 2, and C is not in this task, the DDL statement is delivered to all partitions under Topic 1 and Topic 2.	The DDL statement is delivered to all partitions of all topics in this task. For example, if a DDL statement cannot be recognized by data migration, and this task has three topics, the DDL statement is delivered to all partitions of these three topics.	The DDL statement is delivered to all partitions of the topic to which the table belongs.
     Table	The DDL statement is delivered to the partition where the hash value of the table name is located, within the topic of the related table. For example, if a DDL statement involves tables A, B, and C, where A is in Topic 1, B is in Topic 2, and C is not in this task, the DDL statement is delivered to the partition where the hash value of the related table is located under Topic 1 and Topic 2.	The DDL statement is delivered to all partitions of all topics in this task. For example, if a DDL statement cannot be recognized by data migration, and this task has three topics, the DDL statement is delivered to all partitions of these three topics.	The DDL statement is hashed based on the table name and delivered to a specific partition within the topic to which the table belongs.
     One	The DDL statement is delivered to a fixed partition of the topic where the related table is located. For example, if a DDL statement involves tables A, B, and C, where A is in Topic 1, B is in Topic 2, and C is not in this task, the DDL statement is delivered to a fixed partition under Topic 1 and Topic 2.	The DDL statement is delivered to a fixed partition of a specific topic in all tasks. For example, if a DDL statement cannot be recognized by data migration, and this task has three topics, the DDL statement is delivered to a fixed partition of these three topics.	The DDL statement is delivered to a fixed partition under the topic to which the table belongs.

7. Click Next to have the system perform a pre-check on the data migration task.

   During the Pre-check step, data migration checks the connection to the target Kafka. If the pre-check reports an error:

   • You can troubleshoot and resolve the issue, then re-execute the pre-check until it succeeds.

   • You can also click Skip in the Actions column for the failed pre-check item. A dialog box appears, explaining the specific impact of skipping this operation. After confirming you can skip it, click OK in the dialog box.

8. After the pre-check succeeds, click Purchase to proceed to the Purchase Data Migration Instance page for purchase.

   After successful purchase, you can start the data migration task. For purchase details, see Purchase a data migration instance. If you do not need to purchase a data migration instance immediately, click Save to go to the details page of the data migration task. You can manually purchase it later as needed.

   The data migration service supports modifying migration objects while a data migration task is running. For details, see View and modify migration objects. After a data migration task is started, it executes in sequence according to the selected migration type. For details, see the "View migration details" section in View the details of a data migration task.