ApsaraMQ for Kafka:Create an AnalyticDB sink connector

Step 1: Create the destination resources

Create an AnalyticDB for MySQL or an AnalyticDB for PostgreSQL resource.

• AnalyticDB for MySQL: In the AnalyticDB for MySQL console, create a cluster and a database account, connect to the cluster, and then create a database. For more information, see Create a cluster, Create a database account, Connect to a cluster, and Create a database.

• AnalyticDB for PostgreSQL: In the AnalyticDB for PostgreSQL console, create an instance and a database account, and then log on to the database. For more information, see Create an instance, Create and manage users, and Client connection.

This example uses an AnalyticDB for MySQL database named adb_sink_database and a table named adb_sink_table.

Step 2: Create and enable the AnalyticDB sink connector

1. Log on to the ApsaraMQ for Kafka console. In the Resource Distribution section of the Overview page, select the region where the ApsaraMQ for Kafka instance that you want to manage resides.

2. In the left-side navigation pane, choose Connector Ecosystem Integration > Tasks.

3. On the Tasks page, click Create Task.

   • Configure the task

     1. In the Source step, select Message Queue for Apache Kafka as the Data Provider. Configure the following parameters, and then click Next.

        Parameter	Description	Example
        Region	The region of the source Message Queue for Apache Kafka instance.	China (Beijing)
        Kafka instance	The source Message Queue for Apache Kafka instance.	alikafka_post-cn-jte3****
        Topic	Select the topic for producing Message Queue for Apache Kafka messages.	demo-topic
        Group ID	The consumer group of the source instance. Quick Create: This is the recommended option. A group ID that is named in the GID_EVENTBRIDGE_xxx format is automatically created. Use Existing: Select an independent group ID. Do not use a group ID that is shared with other services to avoid disrupting existing message consumption.	Quick Create
        Consumer offset	The offset from which message consumption starts. Latest offset (latest) Earliest offset (earliest)	Latest offset (latest)
        Network configuration	The network type for routing messages. Basic Network Self-managed Internet	Basic Network
        VPC	Required only if Network configuration is set to Self-managed Internet.	vpc-bp17fapfdj0dwzjkd****
        vSwitch	Required only if Network configuration is set to Self-managed Internet.	vsw-bp1gbjhj53hdjdkg****
        Security group	Required only if Network configuration is set to Self-managed Internet.	alikafka_pre-cn-7mz2****
        Data Format	The encoding format for the message content. For data sources that support binary transmission, select one of the following formats. We recommend Json if you do not have specific encoding requirements. Json: Encodes binary data as a JSON object into the payload using UTF-8. Text: Encodes binary data as a UTF-8 string into the payload. This is the default format. Binary: Encodes binary data as a Base64-encoded string into the payload.	Json
        Messages	Advanced configuration parameter. The maximum number of messages to send per batch. A request is sent only when the number of accumulated messages reaches the specified value. Valid values: 1 to 10,000.	100
        Interval (Unit: Seconds)	Advanced configuration parameter. The interval for invoking a function. The system aggregates messages and sends them to the sink at each interval. Valid values: 0 to 15. A value of 0 specifies that messages are delivered immediately without a delay.	3

     2. In the Filtering step, set the Pattern Content to filter events. For more information, see event pattern.

     3. In the Transformation step, configure data transformation to perform operations such as splitting, mapping, enrichment, and dynamic routing. For more information, see Use Function Compute to clean message data.

     4. In the Sink step, set Service Type to AnalyticDB, configure the following parameters, and then click Save.

        Parameter	Description	Example
        Instance type	Select the database type of your destination instance. AnalyticDB for MySQL AnalyticDB for PostgreSQL	MySQL version
        AnalyticDB instance ID	Select the destination instance.	gp-bp10uo5n536wd****
        Database name	Select the destination database.	adb_sink_database
        Table name	Select the destination data table.	adb_sink_table
        Data Mapping	Use JSONPath expressions to define data extraction rules. When Data Format is set to Json in the Source step, data streamed from ApsaraMQ for Kafka is wrapped in a CloudEvents structure, as shown below: { "data": { "topic": "demo-topic", "partition": 0, "offset": 2, "timestamp": 1739756629123, "headers": { "headers": [], "isReadOnly": false }, "key":"adb-sink-k1", "value": { "userid":"xiaoming", "source":"shanghai" } }, "id": "7702ca16-f944-4b08-***-***-0-2", "source": "acs:alikafka", "specversion": "1.0", "type": "alikafka:Topic:Message", "datacontenttype": "application/json; charset=utf-8", "time": "2025-02-17T01:43:49.123Z", "subject": "acs:alikafka:alikafka_serverless-cn-lf6418u6701:topic:demo-topic", "aliyunaccountid": "1******6789" }	Map each destination table column to a field in the source message using a JSONPath expression. For example, to map the userid field from the message to a table column, use the expression $.data.value.userid.
        Database username	Enter the username for the database account.	user
        Database password	Enter the password for the database account.	******
        Network configuration	VPC: Connect to AnalyticDB through a VPC. Public Network: Connect to AnalyticDB over the public internet.	VPC
        VPC	Select the VPC ID. This parameter is required only when Network configuration is set to VPC.	vpc-bp17fapfdj0dwzjkd****
        vSwitch	Select the vSwitch ID. This parameter is required only when Network configuration is set to VPC. Important After you select a vSwitch, you must add the CIDR block of the vSwitch to the IP address whitelist of the AnalyticDB for MySQL instance. For more information, see Configure an IP address whitelist.	vsw-bp1gbjhj53hdjdkg****
        Security group	Select the security group. This parameter is required only when Network configuration is set to VPC.	test_group

4. Return to the Tasks page. Find the task you created and click Enable in the Actions column.

5. In the Note dialog box, read the message and then click OK.

   The task takes 30 to 60 seconds to start. You can monitor the progress in the Status column on the Tasks page.

Step 3: Verify the AnalyticDB sink connector

1. On the Tasks page, find your task and click the source topic name in the Event Source column.

2. On the topic details page, click Send Test Message.

3. In the Start to Send and Consume Message panel, configure the message body, and then click OK.

   Note

   The message body must be in JSON format. The fields specified in your data mapping rules will be extracted and written to the corresponding columns in the destination table.

   In the Start to Send and Consume Message dialog box, select the Console tab. Set Message key to adb-sink-k1 and Message body to {"userid":"xiaoming","source":"shanghai"}. For Send to a specific partition, select No, and then click OK.

4. On the Tasks page, find your task and click the destination instance name in the Event Target column.

5. On the Basic Information page of the instance, click Log On to Database in the upper-right corner.

6. In the Data Management Service (DMS) console, run the following SQL statement to query all data in the table.

   SELECT * FROM  adb_sink_table;

   The query should return a record with a userid of xiaoming and a source of shanghai. This confirms that the data was successfully written to the destination table.