Kafka data ingestion

Usage notes

• You can only export data from the source Topic of a Message Queue for Apache Kafka instance to Tablestore within the same region. For information about the limitations of the connector, see Limitations.

• When you create a connector, Message Queue for Apache Kafka uses a service-linked role to access other cloud services.

  • If a service-linked role has not been created, Message Queue for Apache Kafka automatically creates a service-linked role for you so that you can use Message Queue for Apache Kafka to export data to Tablestore.

  • If a service-linked role already exists, it is reused.

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

• You can currently export data only to tables that use the wide column model. To request support for the time series model, please submit a ticket.

Step 1: Create destination resources

1. Activate Tablestore and create an instance. For more information, see Activate Tablestore and create an instance.

2. Create a data table. For more information, see Create a data table.

Step 2: Create and start the Tablestore connector

1. Log on to the Message Queue for Apache Kafka console, and in the Resource Distribution section of the Overview page, select a region.

2. In the left navigation bar, select Connector Ecosystem Integration > Tasks.

3. On the Tasks page, click Create Task.

4. On the Create Task page, specify the Task Name and Description, and then configure the following parameters.

   • Create Task

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

        Parameter	Description	Example
        Region	Select the region where the source Message Queue for Apache Kafka instance is located.	China (Beijing)
        Kafka instance	Select the source Message Queue for Apache Kafka instance that produces messages.	MQ_INST_115964845466****_ByBeUp3p
        Topic	Select the source topic to which messages are produced in the Message Queue for Apache Kafka instance.	topic
        Consumer group ID	The ID of the consumer group that consumes messages from the source topic. Quick Create Recommended. Automatically creates a consumer group ID in the GID_EVENTBRIDGE_xxx format. Use Existing Select a dedicated consumer group ID. Do not share a consumer group ID with other business services to avoid affecting existing message consumption.	Quick Create
        Concurrency quota (number of consumers)	The number of consumers for the source instance.	1
        Consumer offset	The starting position (offset) in the topic from which to consume messages.	Latest offset
        Network type	The network type used to route messages.	classic network
        Data format	The encoding format of the source data. Json: Data is encoded in JSON format. Text: Data is encoded in plain text format. Binary: Data is encoded in binary format. Note Multiple data formats are supported. If you do not have specific encoding requirements, set the format to Json.	Json
        Virtual private cloud (VPC)	The ID of the VPC. This parameter is required only when Network type is set to VPC.	vpc-bp17fapfdj0dwzjkd****
        vSwitch	The ID of the vSwitch. This parameter is required only when Network type is set to VPC.	vsw-bp1gbjhj53hdjdkg****
        Security group	The security group. This parameter is required only when Network type is set to VPC.	alikafka_pre-cn-7mz2****
        Advanced configurations	Batch push allows you to aggregate multiple events. A batch is pushed when either the Messages or the Interval (Unit: Seconds) threshold is met. For example, if you set the number of messages to 100 and the interval to 15s, a batch is pushed if the message count reaches 100 in 10s, without waiting for the full 15s.	Enabled
        Messages	The maximum number of messages to include in a single batch. A delivery request is triggered when the number of accumulated messages reaches this value. Valid values: 1 to 10,000.	100
        Interval (Unit: Seconds)	The interval, in seconds, at which the system aggregates and delivers messages to Function Compute. Valid values: 0 to 15. A value of 0 specifies that messages are delivered immediately without a waiting period.	3

     2. In the Filtering step, configure the Pattern Content to filter requests based on their data. For more information, see Event patterns.

     3. In the Transformation step, configure data cleansing to perform complex data processing such as splitting, mapping, enriching, and dynamic routing. For more information, see Use Function Compute to cleanse message data.

     4. In the Sink step, select Tablestore for Service Type 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 data table in Tablestore.	ots_sink_table
        Tablestore model	Currently, only the wide column model is supported. To request support for other models, please submit a ticket.	Wide column model
        Primary key value	Specifies how to generate values for the Tablestore primary keys. You must define an extraction rule for each primary key by using JSONPath syntax.	For example, if the primary key is named pk, you can set the value extraction rule to $.data.value.pk.
        Attribute column name	The name of the attribute column in the Tablestore data table.	N/A
        Attribute type	The type of the attribute column in the Tablestore data table.	N/A
        Attribute value	Specifies how to generate values for attribute columns. You must define an extraction rule for each attribute column by using JSONPath syntax.	For example, if the attribute column is named attr, you can set the value extraction rule to $.data.value.attr.
        Storage format	The storage format for attribute column values. Default: The content is converted to a string and then stored. json: Each field in the JSON object is stored in a separate column. The column name is in the key_{fieldName} format.	Default
        Write mode	Specifies how to write data to Tablestore. put: Overwrites an entire row. update: Updates the specified columns of a row.	put
        Network type	VPC: Delivers Kafka messages to Tablestore through a VPC. Public Network: Delivers Kafka messages to Tablestore over the public network.	Public Network
        VPC	The ID of the VPC. This parameter is required only when Network type is set to VPC.	vpc-bp17fapfdj0dwzjkd****
        vSwitch	The ID of the vSwitch. This parameter is required only when Network type is set to VPC.	vsw-bp1gbjhj53hdjdkg****
        Security group	The security group. This parameter is required only when Network type is set to VPC.	test_group
        Batch push	Batch push allows you to aggregate multiple events. A batch is pushed when either the Number of batch messages or the Batch push interval threshold is met. For example, if you set the number of messages to 100 and the interval to 15s, a batch is pushed if the message count reaches 100 in 10s, without waiting for the full 15s. Enable: Enables the batch push feature. Disable: Disables the batch push feature. By default, only one message is delivered to Function Compute per invocation.	Disable
        Number of batch messages	The maximum number of messages to send in a single function call. A request is sent when the number of accumulated messages reaches this value. Valid values: 1 to 10,000. This parameter is required only when Batch push is set to Enable.	100
        Batch push interval	The time interval for invoking the function. The system aggregates messages and sends them to Function Compute at each interval. Valid values: 0 to 15. The unit is seconds. A value of 0 specifies that messages are delivered immediately without a waiting period. This parameter is required only when Batch push is set to Enable.	10

   • Task Properties

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

5. Click Save. The system starts the task and redirects you to the Tasks page.

   Note

   The task starts after a 30- to 60-second delay. You can view the startup progress in the Status column on the Tasks page.

Step 3: Test the Tablestore connector

1. On the Task List page, in the Event Source column of the task, click the source Topic.

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

3. In the Quick Experience for Message Sending and Receiving panel, configure the following message content, and then click OK.

   Set Send Method to Console. Set Message Key to demo. For Message Content, enter a JSON object with the following key-value pairs: pk1 as guide-tablestore-sink, pk2 as 0, attr1 as ots-sink-k1, and attr2 as ots-sink-v1. Set Send to Specified Partition to No.

4. On the Task List page, click the target table name in the Event Target column of the Tablestore Sink Connector task.

5. On the Table Management page, click the Data Management tab, then click Query Data, set the Query Range, and click OK.

   The query result shows that the data was successfully written to Tablestore. The data table now contains one row with the following values: primary key pk1 is guide-tablestore-sink, primary key pk2 is 0, attr1 is ots-sink-k1, and attr2 is ots-sink-v1.