ApsaraMQ for Kafka:Create a Tablestore sink connector

Step 1: Create a Tablestore table

Create a Tablestore table to store data streamed from ApsaraMQ for Kafka. For more information, see Procedure.

Step 2: Create and start the Tablestore sink connector

1. Log on to the ApsaraMQ for Kafka console. On the Overview page, select a region in the Resource Distribution section.

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

3. On the Tasks page, click Create Task.

4. On the Create Task page, set Task Name and Description, configure the following parameters, and then click Save.

   • Create 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	The topic from which messages are consumed.	demo-topic
        Group ID	The consumer group of the source instance. Quick Create (recommended): A group ID in the GID_EVENTBRIDGE_xxx format is automatically created. Use Existing: Select an existing group ID. Do not share this group ID with other services to avoid disrupting 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. 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 per batch. A request is sent when the accumulated message count reaches this value. Valid values: 1 to 10,000.	100
        Interval (Unit: Seconds)	Advanced configuration parameter. The interval, in seconds, at which messages are aggregated and sent to the sink. Valid values: 0 to 15. A value of 0 means messages are delivered immediately.	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 Tablestore and configure the following parameters.

        Parameter	Description	Example
        Instance Name	The name of the destination Tablestore instance.	ots-sink
        Destination Table	The name of the destination Tablestore data table.	ots_sink_table
        Primary Key	Define extraction rules for primary key values and attribute columns by using JsonPath syntax. When the Data Format in the Source is set to JSON, the output data from ApsaraMQ for Kafka is in the following format: { "data": { "topic": "demo-topic", "partition": 0, "offset": 2, "timestamp": 1739756629123, "headers": { "headers": [], "isReadOnly": false }, "key":"ots-sink-k1", "value": "ots-sink-v1" }, "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" }	For example, for a primary key named topic, set the value extraction rule to $.data.topic.
        Attribute Column		For example, for an attribute column named key, set the value extraction rule to $.data.key.
        Operation Mode	The method for writing data to Tablestore. put: If a record with the same primary key already exists, the new data overwrites the existing data. update: When two records share the same primary key, incremental columns are added to the row without deleting existing columns. delete: Deletes the row that matches the specified primary key.	put
        Network configuration	VPC: Use a VPC to deliver Kafka messages to Tablestore. Public Network: Delivers Kafka messages to Tablestore over the public network.	VPC
        VPC	Select the VPC ID. This parameter is required only if Network Configuration is set to VPC.	vpc-bp17fapfdj0dwzjkd****
        vSwitch	Select the vSwitch ID. This parameter is required only if Network Configuration is set to VPC.	vsw-bp1gbjhj53hdjdkg****
        Security group	Select a security group. This parameter is required only when Network Configuration is set to VPC.	test_group

   • Task Properties

     Configure the retry policy for failed event deliveries and the method for handling errors. For more information, see Retry policies and dead-letter queues.

5. Return to the Tasks page. Find your task and click Enable in the Actions column.

6. In the Note dialog box, read the information and click OK.

   After you enable the task, it starts within 30 to 60 seconds. You can monitor its status in the Status column on the Tasks page.

Step 3: Test the Tablestore sink connector

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

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

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

   On the Console tab, set Message Key to ots-sink-k1, Message Content to ots-sink-v1, and Send to Specified Partition to No.

4. Return to the Tasks page, and in the Event Target column for your task, click the destination table name.

5. On the Manage Table page, click the Data Management tab to view the data in the Tablestore table.

   The data table contains the topic (primary key), partition (primary key), offset (primary key), key, and value columns. A record with values such as partition=2, offset=10, key=ots-sink-k1, and value=ots-sink-v1 confirms that the message was successfully written to Tablestore.