DataWorks:Real-time synchronization from a Hologres table to Kafka

1. Select a synchronization task type

1. Log on to the DataWorks console. In the target region, click Data Integration > Data Integration in the left-side navigation pane. Select a workspace from the drop-down list and click Go to Data Integration.

2. In the left-side navigation pane, click Sync Tasks. At the top of the page, click Create synchronization task and configure the following parameters.

   • Data Source And Destination: Hologres→Kafka

   • New Task Name: Customize a name for the synchronization task.

   • Synchronization Mode: single-table real-time.

   • Synchronization Mode Step: Select full synchronization.

2. Configure network and resources

1. In the Configure Network Connections and Resource Groups section, select the Resource Group for the synchronization task. You can also set the Task Resource Usage in CUs.

2. For Source Data Source, select a Hologres data source. For Destination data source, select a Kafka data source. Then, click Test Connection.

3. Once the connectivity tests for the source and destination data sources pass, click Next step.

3. Configure the synchronization link

4. Alerts

To prevent task errors from causing delays in business data synchronization, you can set an alert policy for the synchronization task.

1. Click Alert Settings in the upper-right corner of the page to open the Alert Rule Configurations for Real-time Synchronization Subnode settings page.

2. Click Add Alert Rule to configure an alert rule.

   Note

   The alert rules you define here apply to the real-time synchronization subtasks that this task generates. After you configure the task, you can view and modify the alert rules for these subtasks on the Run and manage real-time synchronization tasks page.

3. Manage alert rules.

   For existing alert rules, you can use the toggle switch to enable or disable them. You can also send alerts to different recipients based on the alert level.

5. Advanced settings

The synchronization task provides several parameters that you can modify as needed.

1. Click advanced settings in the upper-right corner of the page to open the advanced settings page.

2. On the advanced settings page, modify the parameter values as needed.

6. Resource group

You can click Configure Resource Group in the upper-right corner to view and switch the task's current resource group.

7. Execute the synchronization task

1. After completing all settings, click Complete at the bottom of the page.

2. On the Data Integration > Synchronization Task page, find the task you created and click Start in the Operations column.

3. Click the Name/ID of the corresponding task in the Task List to view its detailed execution process.

View the status of the synchronization task

After the synchronization task is created, go to the Tasks page to view all synchronization tasks in the workspace and their basic information.

The task list displays the following columns: Name/ID, Data Source Synchronization Solution (synchronization type, source, and destination), Status, Execution Overview, resource group, and Actions.

• You can Start or Stop a synchronization task in the Actions column. Under More, you can perform other operations, such as Edit and View.

• For started tasks, you can view the basic running status in the Execution Overview and click the corresponding overview area to view the execution details.

A real-time synchronization task from a Hologres table to Kafka consists of the following three steps:

• Structure Migration: includes the creation method of the destination table (existing table or automatic table creation). If you select automatic table creation, the data definition language (DDL) statement for creating the table is displayed.

• Full initialization: If you select Full Sync for the synchronization step of your task, the full initialization progress is displayed here.

• Real-time Data Synchronization: Contains statistics about real-time synchronization, including real-time read/write traffic, dirty data, failovers, and operation logs.

Rerun a task

In special cases, such as when you need to modify synchronized fields or adjust target table information, you can click Rerun in the Operations column of the synchronization task. This action synchronizes the adjusted fields and other changes to the target. The process skips unchanged, previously synchronized tables.

• To run the task again without any changes, click Rerun.

• If you edit the task, click Complete after making your changes. The task's action changes to Apply Updates. Clicking Apply Updates reruns the task with the new configuration.

Canal CDC

Canal CDC is a CDC data format defined by Alibaba Canal.

• Fields and meanings

  Field name	Field value meaning
  id	The value of this field is fixed as 0.
  database	The name of the Hologres database.
  table	The name of the Hologres table.
  pkNames	The primary key column of the Hologres table.
  isDdl	Specifies whether binary logs record DDL changes. The value of this field is fixed as false because synchronization of messages for DDL changes on the Hologres table is not supported.
  type	The type of the DML change. Valid values: INSERT, UPDATE, and DELETE. Note One change operation on the Hologres table generates two records whose type is UPDATE. The records will be written to the Kafka topic. One of the records corresponds to the data content before the change. The other corresponds to the data content after the change. If you want to synchronize full data from the Hologres table to the Kafka topic, the value of the type field is fixed as INSERT.
  es	A 13-digit timestamp in milliseconds. The timestamp indicates the time when the data in the Hologres table is changed. If you want to synchronize full data from the Hologres table to the Kafka topic, the value of the es field is fixed as 0.
  ts	A 13-digit timestamp in milliseconds. The timestamp indicates the time when the binary logs generated for the Hologres table are read by the synchronization task.
  sql	The SQL code that records DDL changes when the binary logs generated for the Hologres table contain the DDL changes. The value of this field is fixed as an empty string because synchronization of messages for DDL changes on the Hologres table is not supported.
  sqlType	The SQL field data types to which the data types of Hologres table fields correspond. Mappings between Hologres data types and valid values of sqlType: bigint: -5 decimal whose scale is not 0: 3 decimal whose scale is 0: -5 boolean: 16 date: 91 float4: 6 float8: 8 integer: 4 smallint: 5 json: 12 text: 12 varchar: 12 timestamp: 93 timestamptz: 93 bigserial: -5 bytea: 12 char: 12 serial: 4 time: 92 int4[]: 12 int8[]: 12 float4[]: 12 float8[]: 12 boolean[]: 12 text[]: 12
  mysqlType	The MySQL field data types to which the data types of Hologres table fields correspond. Mappings between Hologres data types and valid values of mysqlType: bigint: BIGINT int4: INT decimal whose scale is not 0: DECIMAL(xx,xx) decimal whose scale is 0: BIGINT boolean: BOOLEAN date: DATE float4: FLOAT float8: DOUBLE integer: INT smallint: SMALLINT json: TEXT text: TEXT varchar: VARCHAR(xx) timestamp: DATETIME(6) timestamptz: DATETIME(6) bigserial: BIGINT bytea: TEXT char: TEXT serial: INT time: TIME(6) int4[]: TEXT int8[]: TEXT float4[]: TEXT float8[]: TEXT boolean[]: TEXT text[]: TEXT
  data	The data changes in the Hologres table. The field names in the Hologres table are used as keys, and the data changes in the fields are serialized into strings and used as values. Then, the keys and values are organized as JSON-formatted strings. For more information about serialization, see Description of JSON serialization.
  old	Two records whose type is UPDATE are generated for one change operation on the Hologres table and written to the Kafka topic. The records correspond to the data content before and after the change. In the first record, the old field is used to record data content in the Hologres table before the change. The data field is used to record data content for DML changes other than UPDATE.

• Example of Canal JSON-formatted data that corresponds to data changes generated by an INSERT operation in binary logs of the Hologres table

  {
      "id": 0,
      "database": "test",
      "table": "tp_int",
      "pkNames": [
          "id"
      ],
      "isDdl": false,
      "type": "INSERT",
      "es": 1640007049196,
      "ts": 1639633142960,
      "sql": "",
      "sqlType": {
          "bigint": -5,
          "integer": 4,
          "smallint": 5
      },
      "mysqlType": {
          "bigint": "BIGINT",
          "integer": "INT",
          "smallint": "SMALLINT"
      },
      "data": [
          {
              "bigint": "9223372036854775807",
              "integer": "2147483647",
              "smallint": "32767"
          }
      ],
      "old": null
  }

• Example of Canal JSON-formatted data that corresponds to full data that is synchronized from the Hologres table

  {
      "id": 0,
      "database": "test",
      "table": "tp_int",
      "pkNames": [
          "id"
      ],
      "isDdl": false,
      "type": "INSERT",
      "es": 0,
      "ts": 1639633142960,
      "sql": "",
      "sqlType": {
          "bigint": -5,
          "integer": 4,
          "smallint": 5
      },
      "mysqlType": {
          "bigint": "BIGINT",
          "integer": "INT",
          "smallint": "SMALLINT"
      },
      "data": [
          {
              "bigint": "9223372036854775807",
              "integer": "2147483647",
              "smallint": "32767"
          }
      ],
      "old": null
  }

• Example of two Canal JSON-formatted data records that correspond to data changes generated by an UPDATE operation in binary logs of the Hologres table

  // Data content before a change
  {
      "id": 0,
      "database": "test",
      "table": "tp_int",
      "pkNames": [
          "id"
      ],
      "isDdl": false,
      "type": "UPDATE",
      "es": 1640007049196,
      "ts": 1639633142960,
      "sql": "",
      "sqlType": {
          "bigint": -5,
          "integer": 4,
          "smallint": 5
      },
      "mysqlType": {
          "bigint": "BIGINT",
          "integer": "INT",
          "smallint": "SMALLINT"
      },
      "old": [
          {
              "bigint": "0",
              "integer": "0",
              "smallint": "0"
          }
      ],
      "data": null
  }
  // Data content after a change
  {
      "id": 0,
      "database": "test",
      "table": "tp_int",
      "pkNames": [
          "id"
      ],
      "isDdl": false,
      "type": "UPDATE",
      "es": 1640007049196,
      "ts": 1639633142960,
      "sql": "",
      "sqlType": {
          "bigint": -5,
          "integer": 4,
          "smallint": 5
      },
      "mysqlType": {
          "bigint": "BIGINT",
          "integer": "INT",
          "smallint": "SMALLINT"
      },
      "data": [
          {
              "bigint": "9223372036854775807",
              "integer": "2147483647",
              "smallint": "32767"
          }
      ],
      "old": null
  }

• Example of Canal JSON-formatted data that corresponds to data changes generated by a DELETE operation in binary logs of the Hologres table

  {
      "id": 0,
      "database": "test",
      "table": "tp_int",
      "pkNames": [
          "id"
      ],
      "isDdl": false,
      "type": "DELETE",
      "es": 1640007049196,
      "ts": 1639633142960,
      "sql": "",
      "sqlType": {
          "bigint": -5,
          "integer": 4,
          "smallint": 5
      },
      "mysqlType": {
          "bigint": "BIGINT",
          "integer": "INT",
          "smallint": "SMALLINT"
      },
      "data": [
          {
              "bigint": "9223372036854775807",
              "integer": "2147483647",
              "smallint": "32767"
          }
      ],
      "old": null
  }

Json

The JSON format uses Hologres binary log field names as keys and serializes field values into strings. The resulting key-value pairs are written to the Kafka topic as JSON-formatted strings.