Synchronize an entire MySQL database to LogHub (SLS) in real time - DataWorks

A real-time synchronization task lets you sync incremental data from an entire MySQL database to LogHub (SLS) in real time. This topic describes how to create a real-time sync task to synchronize an entire MySQL database to LogHub and how to view the task status after it is created.

Prerequisites

You have purchased a Serverless resource group or an exclusive resource group for Data Integration.
You have created a MySQL data source and a LogHub (SLS) data source. For more information, see Data Source Configuration.
You have established network connectivity between the resource group and the data sources. For more information, see Overview of network connection solutions.

Procedure

1. Select the sync task type

Go to the Data Integration page.
Log on to the DataWorks console. In the top navigation bar, select the desired region. In the left-side navigation pane, choose Data Integration > Data Integration. On the page that appears, select the desired workspace from the drop-down list and click Go to Data Integration.
In the left navigation pane, click Synchronization Task. At the top of the page, click New Synchronization Task. In the dialog box that appears, configure the following basic parameters.
- Source Type: MySQL.
- Destination Type: LogHub.
- Task Name: A custom name for the sync task.
- Task Type: Real-time migration of entire database.

2. Configure the network and resources

In the Network and Resource Configuration section, select a Resource Group for the sync task. You can allocate CUs under Task Resource Usage.
Set Source to the MySQL data source and Destination to the LogHub data source. Then, click Test Connectivity.
After the connectivity test passes for both the source and destination data sources, click Next.

3. Select data sources and tables for data synchronization

In this step, select the tables to sync from the Source Table section and click the icon to move them to the Selected Tables section on the right.

SSelect Specified Database and Table:
- In the Database Filtering and Search for Tables fields of the Source Databases and Tables section, enter keywords to filter the databases and tables that you want to sync. Select the required databases and tables, and click the icon to move them to the Selected Tables section.
- In the Selected Tables area, use the Database Filtering and Search for Tables fields to find the databases and tables to exclude from the sync. Select these items, click the icon, and move them to the Source Table area.
Use Regular Expression (table addition and removal during running based on a regular expression supported):
Filter the databases and tables by entering regular expressions in the Database Filtering and Table Filtering fields. Click OK to select the databases and tables to sync.
Note
For example, to filter databases by the prefix a and tables by the prefix order, you can enter a.* in the Database Filtering box and order.* in the Table Filtering box.

4. Map target tables

After you select the tables from which you want to synchronize data, the selected tables are automatically displayed in the Mapping Rules for Destination Tables section. The properties of the destination tables are waiting to be mapped. You must manually define mappings between the source tables and destination tables to determine the data reading and writing relationships. Then, you can click Refresh in the Actions column. You can directly refresh mappings between source tables and destination tables. You can also refresh mappings between source tables and destination tables after you configure settings related to destination tables.

1. Select the destination Logstore

Select a destination Logstore for the sync task from the drop-down list in the Destination Logstore column. To configure multiple tables at once, select them and click Batch Modify > Destination Logstore.

2. Configure DML rules

Data Integration provides default Data Manipulation Language (DML) processing rules. You can also define custom rules for DML commands that are written to the destination table.

To define rules for a single table, click Configure in the DML Rule Configuration column.
Define rules in batch: Select all tables to be synced. At the bottom of the list, choose Batch Modify > DML Rule Configuration.

5. Configure alert rules

To prevent the failure of the synchronization task from causing latency on business data synchronization, you can configure different alert rules for the synchronization task.

In the upper-right corner of the page, click Configure Alert Rule to go to the Alert Rule Configurations for Real-time Synchronization Subnode panel.
In the Configure Alert Rule panel, click Add Alert Rule. In the Add Alert Rule dialog box, configure the parameters to configure an alert rule.
Note
The alert rules that you configure in this step take effect for the real-time synchronization subtask that will be generated by the synchronization task. After the configuration of the synchronization task is complete, you can refer to Run and manage real-time synchronization tasks to go to the Real-time Synchronization Task page and modify alert rules configured for the real-time synchronization subtask.
Manage alert rules.
You can enable or disable alert rules that are created. You can also specify different alert recipients based on the severity levels of alerts.

6. Configure advanced parameters

To perform a fine-grained configuration for the synchronization task, click Customize Advanced Parameters in the top-right corner of the page.

In the upper-right corner of the page, click Configure Advanced Parameters.

The following table describes the key parameters.

Parameter	Description
Whether to expand the collected data	Controls whether to flatten the business data fields from the source table to the top layer of the log. true: Expands the data. You can set Data Expansion Format to Partial or full. false (default): Does not expand the data. The data is encapsulated within the entire JSON field.
Whether to collect DDL event data	Controls whether to collect Data Definition Language (DDL) operation information from the source table, such as `CREATE TABLE` and `ALTER TABLE`, and sync it to SLS as a log entry. true (default): Passes through the information. false: Does not pass through the information.
Null Value Handling Strategy from Source	Configures the value to fill in the destination field when the source table field is NULL. The default value is empty, which means the NULL value is retained.
Data Expansion Format	Defines the structure of the business data (the `data` field) when it is written to SLS. This parameter is visible only when Whether to expand the collected data is set to true. The format that you select directly affects how downstream systems consume the logs. For examples of data expansion formats, see Appendix: Data expansion format examples. Partial expansion (default): Compatible with the Logtail format for MySQL binlog collection. Full expansion: Flattens all business data fields, including those before and after the change, into independent top-level key-value pairs in the data.

7. Configure the resource group

You can click Configure Resource Group in the upper-right corner of the page to view and change the resource groups that are used to run the current synchronization task.

8. Run the sync task

After the configuration of the synchronization task is complete, click Complete in the lower part of the page.
In the Synchronization Task section of the Data Integration page, find the created synchronization task and click Start in the Operation column.
Click the Name/ID of the synchronization task in the Tasks section and view the detailed running process of the synchronization task.

Sync task O&M

View the task status

After you create a sync task, you can view the list of created tasks and their basic information on the Sync Task page.

In the Operation column, you can Start or Stop a sync task. Click More to Edit or View the task.
For a running task, you can view its basic status in the Execution Overview section. You can also click an area of the overview to view the execution details.
For a real-time sync task from MySQL to LogHub (SLS), the performance statistics of the real-time synchronization are displayed. These statistics include real-time progress, DDL records, DML records, and alert information.

Rerun the synchronization task

In some special cases, if you add tables to or remove tables from the source, or change the schema or name of a destination table, you can click More in the Operation column of the synchronization task and then click Rerun to rerun the task after the change. During the rerun process, the synchronization task synchronizes data only from the newly added tables to the destination or only from the mapped source table to the destination table whose schema or name is changed.

If you want to rerun the synchronization task without modifying the configuration of the task, click More in the Actions column and then click Rerun to rerun the task to perform full synchronization and incremental synchronization again.
If you want to rerun the synchronization task after you add tables to or remove tables from the task, click Complete after the change. In this case, Apply Updates is displayed in the Actions column of the synchronization task. Click Apply Updates to trigger the system to rerun the synchronization task. During the rerun process, the synchronization task synchronizes data from the newly added tables to the destination. Data in the original tables is not synchronized again.

Appendix: Data expansion format examples

Do not expand

{
  "ExecutionTime": 1761017850000,
  "_db_": "*****",
  "_event_": "row_update",
  "_event_time_": 1761017850,
  "_file_name_": "mysql-bin.*****",
  "_gtid_": "4a21a3ce-ad7a-11f0-a8f3**********",
  "_host_": "rm-*********.mysql.rds.aliyuncs.com",
  "_id_": "176101777********",
  "_offset_": "265*****",
  "_table_": "t_parameter",
  "data": {
    "updateBefore": {
      "_old_id": "3",
      "_old_name": "82174b93-b810-4030-8652-e5c1667d3f72",
      "_old_value": "+@}8-/XC",
      "_old_status": "kBdO",
      "_old_description": "a?!L7{jaH+",
      "_old_create_time": "2023-12-28 19:03:43",
      "_old_create_user": "+Zs",
      "_old_modify_time": "2006-11-26 20:42:31",
      "_old_modify_user": "brTYGI?jLL"
    },
    "updateAfter": {
      "id": "3",
      "name": "82174b93-b810-4030-8652-e5c1667d3f72-tagd",
      "value": "+@}8-/XC",
      "status": "kBdO",
      "description": "a?!L7{jaH+",
      "create_time": "2023-12-28 19:03:43",
      "create_user": "+Zs",
      "modify_time": "2006-11-26 20:42:31",
      "modify_user": "brTYGI?jLL"
    }
  }
}

Partial expansion: Compatible with the Logtail format for MySQL binary log (binlog) collection

{
  "ExecutionTime": 1761017850000,
  "_db_": "*****",
  "_event_": "row_update",
  "_event_time_": 1761017850,
  "_file_name_": "mysql-bin.*****",
  "_gtid_": "4a21a3ce-ad7a-11f0-a8f3**********",
  "_host_": "rm-*********.mysql.rds.aliyuncs.com",
  "_id_": "176101777********",
  "_offset_": "265*****",
  "_table_": "t_parameter",
  "old_data": {
    "_old_id": "1",
    "_old_name": "0e459c1a-c6ce-459b-b374-a161b095c8e9",
    "_old_value": "Hello",
    "_old_status": "b",
    "_old_description": "cw",
    "_old_create_time": "2007-08-06 16:19:03",
    "_old_create_user": "!wW4",
    "_old_modify_time": "2017-04-21 18:21:58",
    "_old_modify_user": "s"
  },
  "data": {
    "id": "1",
    "name": "0e459c1a-c6ce-459b-b374-a161b095c8e9-dsg",
    "value": "Hello",
    "status": "b",
    "description": "cw",
    "create_time": "2007-08-06 16:19:03",
    "create_user": "!wW4",
    "modify_time": "2017-04-21 18:21:58",
    "modify_user": "s"
  }
}

Full expansion: All fields are flattened.

{
  "ExecutionTime": 1761017850000,
  "_db_": "****",
  "_event_": "row_update",
  "_event_time_": 1761017850,
  "_file_name_": "mysql-bin.*****",
  "_gtid_": "4a21a3ce-ad7a-11f0-a8f3**********",
  "_host_": "rm-*********.mysql.rds.aliyuncs.com",
  "_id_": "176101777********",
  "_offset_": "265*****",
  "_table_": "t_parameter",
  "_old_create_time": "2024-09-27 15:27:10",
  "_old_create_user": "o",
  "_old_description": "LZ[1HsTE",
  "_old_id": "6",
  "_old_modify_time": "2008-03-15 08:05:53",
  "_old_modify_user": "/{=>7_d@0Q",
  "_old_name": "cf8a671c-4414-45f5-a22c-62c353a6f1ef",
  "_old_status": "K:HQOX-?gK",
  "_old_value": "23]sn<t",
  "create_time": "2024-09-27 15:27:10",
  "create_user": "o",
  "description": "LZ[1HsTE",
  "id": "6",
  "modify_time": "2008-03-15 08:05:53",
  "modify_user": "/{=>7_d@0Q",
  "name": "cf8a671c-4414-45f5-a22c-62c353a6f1efgsdsa",
  "status": "K:HQOX-?gK",
  "value": "23]sn<t"
}