Configure DTS as an event provider - EventBridge - Alibaba Cloud Documentation Center

Prerequisites

You have created a change tracking task in the DTS console that is in the Normal state. For more information, see Manage a change tracking task.
You have created a consumer group for the change tracking task.
You have activated EventBridge and granted the required permissions.

Supported regions

DTS is supported as an event provider in the following regions: China (Hangzhou), China (Shanghai), China (Qingdao), China (Beijing), China (Shenzhen), China (Guangzhou), China (Chengdu), and China (Hong Kong).

Procedure

Important

The EventBridge event stream relays only the DTS data generated from INSERT, DELETE, UPDATE, and DDL operations.

Log on to the EventBridge console. In the left-side navigation pane, click Event Streams.
In the top navigation bar, select a region and click Create Event Stream.

On the Create Event Stream page, enter a Task Name and a Description, configure the following parameters, and then click Save.

Task creation:

In the Source step, select Data Transmission Service (DTS) (Database) for Data Provider, configure the following parameters, and then click Next.

Parameter	Description	Example
Change Tracking Task	Select the name of the change tracking task that you created in the DTS console.	dts8jqe****
Access Method	The access method for the source database instance of the change tracking task. Read-only.	RDS
Instance ID	The ID of the source database instance for the change tracking task. Read-only.	rm-bp18mj3q2dzyb****
Consumer Group	The consumer group that you created in the prerequisites to consume data from the change tracking task. Note Make sure that the consumer group is not running on another client instance. Otherwise, the specified consumer offset may become invalid.	test
Account	The account that you specified when you created the consumer group.	test
Password	The password for the account that you specified when you created the consumer group.	******
Consumer Offset	The timestamp of the first data record to consume. The timestamp must be within the data range of the tracked instance. Note This setting applies only the first time a consumer group runs. On restart, data consumption resumes from the last recorded offset.	2022-06-21 00:00:00
Messages	The maximum number of messages per batch. A request is sent when the batch reaches this size. Valid values: 1 to 10,000.	100
Interval (Unit: Seconds)	The maximum wait time in seconds before sending a batch. The system sends a batch when this interval elapses, even if the batch is not full. Valid values: 0 to 15. A value of 0 means messages are sent immediately.	3

In the Filtering, Transformation, and Sink steps, configure event filtering, transformation rules, and event targets. For information about how to configure event transformation, see Use Function Compute to perform message cleansing.

Task properties

Configure the retry policy and dead-letter queue for the event stream. For more information, see Retry policies and dead-letter queues.

Return to the Event Streams page. Find the event stream that you created and click Enable in the Actions column.

After you enable the event stream, the configuration may take 30 to 60 seconds to take effect. You can view the progress in the Status column on the Event Streams page.

Sample event

The following sample event is generated when you create a change tracking task for a MySQL instance in DTS.

{
  "data": {
    "id": 321****,
    "topicPartition": {
      "hash": 0,
      "partition": 0,
      "topic": "cn_hangzhou_rm_1234****_test_version2"
    },
    "offset": 3218099,
    "sourceTimestamp": 1654847757,
    "operationType": "UPDATE",
    "schema": {
      "recordFields": [
        {
          "fieldName": "id",
          "rawDataTypeNum": 8,
          "isPrimaryKey": true,
          "isUniqueKey": false,
          "fieldPosition": 0
        },
        {
          "fieldName": "topic",
          "rawDataTypeNum": 253,
          "isPrimaryKey": false,
          "isUniqueKey": false,
          "fieldPosition": 1
        }
      ],
      "nameIndex": {
        "id": {
          "fieldName": "id",
          "rawDataTypeNum": 8,
          "isPrimaryKey": true,
          "isUniqueKey": false,
          "fieldPosition": 0
        },
        "topic": {
          "fieldName": "topic",
          "rawDataTypeNum": 253,
          "isPrimaryKey": false,
          "isUniqueKey": false,
          "fieldPosition": 1
        }
      },
      "schemaId": "(hangzhou-test-db,hangzhou-test-db,message_info)",
      "databaseName": "hangzhou--test-db",
      "tableName": "message_info",
      "primaryIndexInfo": {
        "indexType": "PrimaryKey",
        "indexFields": [
          {
            "fieldName": "id",
            "rawDataTypeNum": 8,
            "isPrimaryKey": true,
            "isUniqueKey": false,
            "fieldPosition": 0
          }
        ],
        "cardinality": 0,
        "nullable": true,
        "isFirstUniqueIndex": false
      },
      "uniqueIndexInfo": [],
      "foreignIndexInfo": [],
      "normalIndexInfo": [],
      "databaseInfo": {
        "databaseType": "MySQL",
        "version": "5.7.35-log"
      },
      "totalRows": 0
    },
    "beforeImage": {
      "recordSchema": {
        "recordFields": [
          {
            "fieldName": "id",
            "rawDataTypeNum": 8,
            "isPrimaryKey": true,
            "isUniqueKey": false,
            "fieldPosition": 0
          },
          {
            "fieldName": "topic",
            "rawDataTypeNum": 253,
            "isPrimaryKey": false,
            "isUniqueKey": false,
            "fieldPosition": 1
          }
        ],
        "nameIndex": {
          "id": {
            "fieldName": "id",
            "rawDataTypeNum": 8,
            "isPrimaryKey": true,
            "isUniqueKey": false,
            "fieldPosition": 0
          },
          "topic": {
            "fieldName": "topic",
            "rawDataTypeNum": 253,
            "isPrimaryKey": false,
            "isUniqueKey": false,
            "fieldPosition": 1
          }
        },
        "schemaId": "(hangzhou-test-db,hangzhou-test-db,message_info)",
        "databaseName": "hangzhou-test-db",
        "tableName": "message_info",
        "primaryIndexInfo": {
          "indexType": "PrimaryKey",
          "indexFields": [
            {
              "fieldName": "id",
              "rawDataTypeNum": 8,
              "isPrimaryKey": true,
              "isUniqueKey": false,
              "fieldPosition": 0
            }
          ],
          "cardinality": 0,
          "nullable": true,
          "isFirstUniqueIndex": false
        },
        "uniqueIndexInfo": [],
        "foreignIndexInfo": [],
        "normalIndexInfo": [],
        "databaseInfo": {
          "databaseType": "MySQL",
          "version": "5.7.35-log"
        },
        "totalRows": 0
      },
      "values": [
        {
          "data": 115
        },
        {
          "data": {
            "hb": [
              104,
              101,
              108,
              108,
              111
            ],
            "offset": 0,
            "isReadOnly": false,
            "bigEndian": true,
            "nativeByteOrder": false,
            "mark": -1,
            "position": 0,
            "limit": 9,
            "capacity": 9,
            "address": 0
          },
          "charset": "utf8mb4"
        }
      ],
      "size": 45
    },
    "afterImage": {
      "recordSchema": {
        "recordFields": [
          {
            "fieldName": "id",
            "rawDataTypeNum": 8,
            "isPrimaryKey": true,
            "isUniqueKey": false,
            "fieldPosition": 0
          },
          {
            "fieldName": "topic",
            "rawDataTypeNum": 253,
            "isPrimaryKey": false,
            "isUniqueKey": false,
            "fieldPosition": 1
          }
        ],
        "nameIndex": {
          "id": {
            "fieldName": "id",
            "rawDataTypeNum": 8,
            "isPrimaryKey": true,
            "isUniqueKey": false,
            "fieldPosition": 0
          },
          "topic": {
            "fieldName": "topic",
            "rawDataTypeNum": 253,
            "isPrimaryKey": false,
            "isUniqueKey": false,
            "fieldPosition": 1
          }
        },
        "schemaId": "(hangzhou-test-db,hangzhou-test-db,message_info)",
        "databaseName": "hangzhou-test-db",
        "tableName": "message_info",
        "primaryIndexInfo": {
          "indexType": "PrimaryKey",
          "indexFields": [
            {
              "fieldName": "id",
              "rawDataTypeNum": 8,
              "isPrimaryKey": true,
              "isUniqueKey": false,
              "fieldPosition": 0
            }
          ],
          "cardinality": 0,
          "nullable": true,
          "isFirstUniqueIndex": false
        },
        "uniqueIndexInfo": [],
        "foreignIndexInfo": [],
        "normalIndexInfo": [],
        "databaseInfo": {
          "databaseType": "MySQL",
          "version": "5.7.35-log"
        },
        "totalRows": 0
      },
      "values": [
        {
          "data": 115
        },
        {
          "data": {
            "hb": [
              98,
              121,
              101
            ],
            "offset": 0,
            "isReadOnly": false,
            "bigEndian": true,
            "nativeByteOrder": false,
            "mark": -1,
            "position": 0,
            "limit": 11,
            "capacity": 11,
            "address": 0
          },
          "charset": "utf8mb4"
        }
      ],
      "size": 47
    }
  },
  "id": "12f701a43741d404fa9a7be89d9acae0-321****",
  "source": "DTSstreamDemo",
  "specversion": "1.0",
  "type": "dts:ConsumeMessage",
  "datacontenttype": "application/json; charset=utf-8",
  "time": "2022-06-10T07:55:57Z",
  "subject": "acs:dts:cn-hangzhou:12345****:kk123abc60g782/dtsabcdet1ro"
}

For information about the CloudEvents specification parameters, see Event structure.

The data field contains the following parameters.

Parameter	Type	Description
`id`	String	The ID of the DTS data record.
`topicPartition`	Array	The partition information of the topic.
`hash`	String	The underlying storage parameter of DTS.
`partition`	String	The partition of the topic.
`topic`	String	The name of the topic.
`offset`	Int	The message storage offset for the DTS data record.
`sourceTimestamp`	Int	The timestamp when the DTS data was generated.
`operationType`	String	The operation type of the DTS data.
`schema`	Array	The database table schema.
`recordFields`	Array	The details of the fields.
`fieldName`	String	The name of the field.
`rawDataTypeNum`	Int	The mapped value of the field type. This value corresponds to the `dataTypeNumber` field in the deserialized incremental data from the change tracking channel. For more information, see Use a Kafka client to consume tracked data.
`isPrimaryKey`	Boolean	Indicates whether the field is a primary key.
`isUniqueKey`	Boolean	Indicates whether the field is a unique key.
`fieldPosition`	String	The position of the field.
`nameIndex`	Array	The named index.
`schemaId`	String	The ID of the database table schema.
`databaseName`	String	The name of the database.
`tableName`	String	The name of the table.
`primaryIndexInfo`	String	The primary key index.
`indexType`	String	The type of the primary key index.
`indexFields`	Array	The fields in the primary key index.
`cardinality`	String	The cardinality of the primary key.
`nullable`	Boolean	Indicates whether the primary key can be null.
`isFirstUniqueIndex`	Boolean	Indicates whether it is the first unique index.
`uniqueIndexInfo`	String	The unique index.
`foreignIndexInfo`	String	The foreign key index.
`normalIndexInfo`	String	The normal index.
`databaseInfo`	Array	The information about the database.
`databaseType`	String	The type of the database.
`version`	String	The version of the database.
`totalRows`	Int	The total number of rows in the table.
`beforeImage`	String	The state of the record fields before the operation.
`values`	String	The values of the record fields.
`size`	Int	The size of the record fields.
`afterImage`	String	The state of the record fields after the operation.