ApsaraMQ for Kafka:Create a Tablestore sink connector

Notes

• You can only export data from a source topic in an ApsaraMQ for Kafka instance to Tablestore within the same region. For information about connector limits, see Usage limits.

• When you create a connector, ApsaraMQ for Kafka automatically creates a service-linked role.

  • If a service-linked role does not exist, ApsaraMQ for Kafka automatically creates one, enabling ApsaraMQ for Kafka to export data to Tablestore.

  • If a service-linked role already exists, ApsaraMQ for Kafka does not create a new one.

  For more information about service-linked roles, see Service-linked roles.

Procedure

This topic describes how to use a Tablestore sink connector to export data from a source topic in an ApsaraMQ for Kafka instance to Tablestore.

1. Optional: Create the topics and group required for the Tablestore sink connector.

   If you do not need to customize the topics and group, you can skip this step and select Auto in the next step.

   Important

   Some topics required by the Tablestore sink connector must use the Local storage engine. For ApsaraMQ for Kafka instances of major version 0.10.2, you cannot manually create topics that use the Local storage engine. These topics can only be created automatically.

   1. Create the topics for the Tablestore sink connector

   2. Create the group for the Tablestore sink connector

2. Create and deploy the Tablestore sink connector

3. Verify the result

   1. Send a test message

   2. View data in the table

Create topics for the Tablestore sink connector

In the ApsaraMQ for Kafka console, manually create the five topics required for the Tablestore sink connector: a task offset topic, a task configuration topic, a task status topic, a dead-letter queue topic, and an error data topic. The required number of partitions and the storage engine vary for each topic. For more information, see List of source service parameters.

1. Log on to the ApsaraMQ for Kafka console.

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

   Important

   You must create topics in the region where your Elastic Compute Service (ECS) instance is deployed. A topic cannot be used across regions. For example, if the producers and consumers of messages run on an ECS instance that is deployed in the China (Beijing) region, the topic must also be created in the China (Beijing) region.

3. On the Instances page, click the name of the instance that you want to manage.

4. In the left-side navigation pane, click Topics.

5. On the Topics page, click Create Topic.

6. In the Create Topic panel, specify the properties of the topic and click OK.

   Parameter	Description	Example
   Name	The topic name.	demo
   Description	The topic description.	demo test
   Partitions	The number of partitions in the topic.	12
   Storage Engine Note You can specify the storage engine type only if you use a non-serverless Professional Edition instance. For other types of instances, Cloud Storage is selected by default.	The type of the storage engine that is used to store messages in the topic. ApsaraMQ for Kafka supports the following types of storage engines: Cloud Storage: If you select this value, the system uses Alibaba Cloud disks for the topic and stores data in three replicas in distributed mode. This storage engine features low latency, high performance, long durability, and high reliability. If you set the Instance Edition parameter to Standard (High Write) when you created the instance, you can set this parameter only to Cloud Storage. Local Storage: If you select this value, the system uses the in-sync replicas (ISR) algorithm of open source Apache Kafka and stores data in three replicas in distributed mode.	Cloud Storage
   Message Type	The message type of the topic. Valid values: Normal Message: By default, messages that have the same key are stored in the same partition in the order in which the messages are sent. If a broker in the cluster fails, the order of messages that are stored in the partitions may not be preserved. If you set the Storage Engine parameter to Cloud Storage, this parameter is automatically set to Normal Message. Partitionally Ordered Message: By default, messages that have the same key are stored in the same partition in the order in which the messages are sent. If a broker in the cluster fails, messages are still stored in the partitions in the order in which the messages are sent. Messages in some partitions cannot be sent until the partitions are restored. If you set the Storage Engine parameter to Local Storage, this parameter is automatically set to Partitionally Ordered Message.	Normal Message
   Log Cleanup Policy	The log cleanup policy that is used by the topic. If you set the Storage Engine parameter to Local Storage, you must configure the Log Cleanup Policy parameter. You can set the Storage Engine parameter to Local Storage only if you use an ApsaraMQ for Kafka Professional Edition instance. ApsaraMQ for Kafka provides the following log cleanup policies: Delete: the default log cleanup policy. If sufficient storage space is available in the system, messages are retained based on the maximum retention period. After the storage usage exceeds 85%, the system deletes the earliest stored messages to ensure service availability. Compact: the log compaction policy that is used in Apache Kafka. Log compaction ensures that the latest values are retained for messages that have the same key. This policy is suitable for scenarios such as restoring a failed system or reloading the cache after a system restarts. For example, when you use Kafka Connect or Confluent Schema Registry, you must store the information about the system status and configurations in a log-compacted topic. Important You can use log-compacted topics only in specific cloud-native components, such as Kafka Connect and Confluent Schema Registry. For more information, see aliware-kafka-demos.	Compact
   Tag	The tags that you want to attach to the topic.	demo

   After a topic is created, you can view the topic on the Topics page.

Create a group for Tablestore sink connector

You can manually create a group for the Tablestore sink connector data synchronization task in the ApsaraMQ for Kafka console. The name of this group must be connect-Task Name. For more information, see Source service parameters.

1. Log on to the ApsaraMQ for Kafka console.

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

3. On the Instances page, click the name of the instance that you want to manage.

4. In the left-side navigation pane, click Groups.

5. On the Groups page, click Create Group.

6. In the Create Group panel, enter a group name in the Group ID field and a group description in the Description field, attach tags to the group, and then click OK.

   After you create a consumer group, you can view the consumer group on the Groups page.

Create and deploy a Tablestore sink connector

Create and deploy a Tablestore sink connector to synchronize data from an ApsaraMQ for Kafka instance to a Tablestore table.

1. Log on to the ApsaraMQ for Kafka console.

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

3. In the left-side navigation pane, click Connectors.

4. On the Connectors page, select the instance to which the connector belongs from the Select Instance drop-down list and click Create Connector.

5. In the Create Connector wizard, complete the following steps.

   1. On the Configure Basic Information tab, configure the following parameters and then click Next.

      Parameter	Description	Example
      Name	The name of the connector. The name must meet the following requirements: The name can be up to 48 characters long and can contain only digits, lowercase letters, and hyphens (-). It cannot start with a hyphen (-). The name must be unique within an ApsaraMQ for Kafka instance. The connector's data synchronization task uses a consumer group named connect-task-name. If you do not create this consumer group manually, the system creates it automatically.	kafka-ts-sink
      Instance	By default, the instance name and instance ID are displayed.	demo alikafka_post-cn-st21p8vj****

   2. On the Configure Source Service tab, set Data Source to ApsaraMQ for Kafka, configure the following parameters, and then click Next.

      Note

      If you have already created the required topic and group, select Manual for resource creation and enter their information. Otherwise, select Auto.

      Table 1. Parameters for the source service

      Parameter	Description	Example
      Data Source Topic	The source topic for data synchronization.	ts-test-input
      Consumer Thread Concurrency	The number of concurrent consumer threads for the data source topic. The default value is 6. Valid values: 1 2 3 6 12	6
      Consumer Offset	The starting position for message consumption. Valid values: Earliest Offset: Consumes messages from the earliest offset. Latest Offset: Consumes messages from the latest offset.	Earliest Offset
      VPC ID	The ID of the virtual private cloud (VPC) where the data synchronization task runs. This parameter is displayed after you click Configure Runtime Environment. This parameter defaults to the VPC of the ApsaraMQ for Kafka instance and requires no configuration.	vpc-bp1xpdnd3l***
      vSwitch ID	The ID of the vSwitch where the data synchronization task runs. This parameter is displayed after you click Configure Runtime Environment. The vSwitch must be in the same VPC as the ApsaraMQ for Kafka instance. This parameter defaults to the vSwitch specified when the ApsaraMQ for Kafka instance was deployed.	vsw-bp1d2jgg81***
      Failure Handling Policy	The policy for handling message-sending failures in a topic partition. This parameter is displayed after you click Configure Runtime Environment. Valid values: Continue Subscription: Continues the subscription to the topic partition where the error occurred and prints error logs. Stop Subscription: Stops the subscription to the topic partition where the error occurred and prints error logs. Note For more information about how to view logs, see Manage connectors. For more information about how to find solutions based on error codes, see Error codes.	Continue Subscription
      Resource Creation Method	The method for creating the required topics and group. This parameter is displayed after you click Configure Runtime Environment. Auto Manual	Auto
      Connector Consumer Group	The consumer group used by the data synchronization task. This parameter is displayed after you click Configure Runtime Environment. The name of the group must be in the format connect-task-name.	connect-cluster-kafka-ots-sink
      Task Offset Topic	The topic that stores consumer offsets. This parameter is displayed after you click Configure Runtime Environment. Topic: The topic name must start with connect-offset. Partitions: The number of partitions for the topic must be greater than 1. Storage engine: The storage engine for the topic must be Local Storage. cleanup.policy: The log cleanup policy for the topic must be compact.	connect-offset-kafka-ots-sink
      Task Configuration Topic	The topic that stores task configurations. This parameter is displayed after you click Configure Runtime Environment. Topic: The topic name must start with connect-config. Partitions: The number of partitions for the topic must be 1. Storage engine: The storage engine for the topic must be Local Storage. cleanup.policy: The log cleanup policy for the topic must be compact.	connect-config-kafka-ots-sink
      Task Status Topic	The topic that stores task status. This parameter is displayed after you click Configure Runtime Environment. Topic: The topic name must start with connect-status. Partitions: The recommended number of partitions is 6. Storage engine: The storage engine for the topic must be Local Storage. cleanup.policy: The log cleanup policy for the topic must be compact.	connect-status-kafka-ots-sink
      Dead-letter Queue Topic	The topic used to store error data from the Kafka Connect framework. This parameter is displayed after you click Configure Runtime Environment. To save topic resources, you can use the same topic for the dead-letter queue and the Error Data Topic. Topic: The topic name must start with connect-error. Partitions: The recommended number of partitions is 6. Storage engine: The storage engine for the topic can be Local Storage or Cloud Storage.	connect-error-kafka-ots-sink
      Error Data Topic	The topic used to store error data from the sink. This parameter is displayed after you click Configure Runtime Environment. To save topic resources, you can use the same topic for the Dead-letter Queue Topic and the error data topic. Topic: The topic name must start with connect-error. Partitions: The recommended number of partitions is 6. Storage engine: The storage engine for the topic can be Local Storage or Cloud Storage.	connect-error-kafka-ots-sink

   3. On the Configure Destination Service tab, set Destination Service to Tablestore, configure the following parameters, and then click Create.

      Parameter	Description	Example
      Instance Name	The name of the Tablestore instance.	k00eny67****
      Automatically Create Destination Table	Specifies whether to automatically create a table in Tablestore. Yes: A table is automatically created in Tablestore to store synchronized data based on the configured table name. No: An existing table is used to store synchronized data.	Yes
      Destination Table Name	The name of the table that stores the synchronized data. If you set Automatically Create Destination Table to No, the table name must be the same as the name of an existing table in the Tablestore instance.	kafka_table
      Tablestore	The type of the table that stores the synchronized data. Wide Column Model TimeSeries Model	Wide Column Model
      Message Key Format	The input format of the message key. Valid values are String and JSON. The default value is JSON. This parameter is displayed only when Tablestore is set to Wide Column Model. String: The message key is parsed directly as a string. JSON: The message key must be in JSON format.	String
      Message Value Format	The input format of the message value. Valid values are String and JSON. The default value is JSON. This parameter is displayed only when Tablestore is set to Wide Column Model. String: The message value is parsed directly as a string. JSON: The message value must be in JSON format.	String
      JSON Message Field Conversion	The method for processing fields in a JSON message. This parameter is displayed if you set Message Key Format or Message Value Format to JSON. Valid values: Write All as String: Converts all fields to the String type in Tablestore. Automatically Identify Field Types: Converts String and Boolean fields in the JSON message body to the corresponding String and Boolean types in Tablestore. Integer and Float data types in the JSON message body are converted to the Double type in Tablestore.	Write All as String
      Primary Key Mode	Specifies the primary key mode. You can extract table primary keys from different parts of ApsaraMQ for Kafka message records, including the coordinates (topic, partition, and offset), key, and value. This parameter is displayed only when Tablestore is set to Wide Column Model. The default value is kafka. kafka: Uses <connect_topic>_<connect_partition> and <connect_offset> as the primary keys of the data table. record_key: Uses fields in the record key as the primary keys of the data table. record_value: Uses fields in the record value as the primary keys of the data table.	kafka
      Primary Key Column Names	The primary key column names of the data table and their corresponding data types. String and Integer data types are supported. This means that fields in the record key or value that match the configured column names are used as the primary keys for the data table. This parameter is displayed if you set Message Key Format to JSON and Primary Key Mode to record_key, or if you set Message Value Format to JSON and Primary Key Mode to record_value. Click Create to add a column name. You can configure a maximum of four column names.	None
      Write Mode	Specifies the write mode. Valid values are put and update. The default value is put. This parameter is displayed only when Tablestore is set to Wide Column Model. put: Overwrites existing data. update: Updates existing data.	put
      Delete Mode	If an ApsaraMQ for Kafka message record contains null values, you can choose whether to delete rows or attribute columns. This parameter is displayed when Primary Key Mode is set to record_key. Valid values: none: The default value. No deletion is allowed. row: Allows row deletion. column: Allows attribute column deletion. row_and_column: Allows row and attribute column deletion. The delete behavior depends on the write mode: If Write Mode is put, any delete mode results in overwrite writes to the Tablestore data table, even when the value contains null fields. If Write Mode is update and the delete mode is none or row, records in which all value fields are null are treated as dirty data. If only some value fields are null, the connector ignores the null fields and writes the non-null fields to the Tablestore data table. If the delete mode is column or row_and_column, the connector deletes rows and attribute columns for null fields and then writes the data to the Tablestore data table.	None
      Metric Name Field	This field is mapped to the measurement name field (_m_name) in the Tablestore TimeSeries Model. The measurement name specifies the name of the physical quantity or monitoring metric for the time-series data, such as temperature or speed. This field cannot be empty. This parameter is displayed when you select the TimeSeries Model for Tablestore.	measurement
      Data Source Field	Map this field to the data source field (_data_source) in the Tablestore TimeSeries Model. This field acts as an identifier for the source of time-series data, such as a machine name or device ID, and can be empty. This parameter is displayed when the TimeSeries model is selected for Tablestore.	source
      Tag Field	Use one or more fields as the tag fields (_tags) for a Tablestore TimeSeries Model. Each tag is a key-value pair of strings. The key is the name of the configured field, and the value is the content of the field. Tags are part of the time series metadata. A time series is uniquely identified by the combination of its metric name, data source, and tags. Tags can be empty. This parameter is displayed when Tablestore is selected as the time series model.	tag1, tag2
      Timestamp Field	Maps this field to the timestamp field (_time) in the Tablestore TimeSeries Model. It represents the point in time for this row of time-series data, such as the time when a physical quantity is generated. When data is written to Tablestore, the timestamp field is converted to microseconds for writing and storage. This parameter is displayed when you select the TimeSeries Model for Tablestore.	time
      Timestamp Unit	Configure this parameter based on the actual timestamp field. This parameter is displayed only when Tablestore is set to TimeSeries Model. Valid values: SECONDS MILLISECONDS MICROSECONDS NANOSECONDS	MILLISECONDS
      Whether to Map All Non-primary Key Fields	Specifies whether to map all non-primary key fields as data fields. Non-primary key fields are fields that are not already mapped as metric name, data source, tag, or timestamp. This parameter is displayed only when Tablestore is set to TimeSeries Model. Valid values: Yes: Fields are automatically mapped and their data types are determined. Numeric types are all converted to the Double type. No: You must specify the fields and types to be mapped.	Yes
      Configure Mapping for All Non-primary Key Fields	The field types corresponding to the non-primary key field names of the time-series table. Double, Integer, String, Binary, and Boolean data types are supported. This parameter is displayed if you set Whether to Map All Non-primary Key Fields to No.	String

      After the connector is created, you can view it on the Connectors page.

6. After the connector is created, find it on the Connectors page and click Deploy in the Actions column.

7. Click OK.

Send a test message

After you deploy the Tablestore sink connector, you can send a message to the source topic in ApsaraMQ for Kafka to verify that data is synchronized to Tablestore.

1. On the Connectors page, find the connector that you want to manage and click Test in the Actions column.

2. In the Send Message panel, configure the parameters to send a message for testing.

   • If you set the Sending Method parameter to Console, perform the following steps:

     1. In the Message Key field, enter the message key. Example: demo.

     2. In the Message Content field, enter the message content. Example: {"key": "test"}.

     3. Configure the Send to Specified Partition parameter to specify whether to send the test message to a specific partition.

        • If you want to send the test message to a specific partition, click Yes and enter the partition ID in the Partition ID field. Example: 0. For information about how to query partition IDs, see View partition status.

        • If you do not want to send the test message to a specific partition, click No.

   • If you set the Sending Method parameter to Docker, run the Docker command in the Run the Docker container to produce a sample message section to send the test message.

   • If you set the Sending Method parameter to SDK, select an SDK for the required programming language or framework and an access method to send and subscribe to the test message.

View table data