ApsaraMQ for Kafka:Create an SLS sink connector

Step 1: Create destination resources

• Create a project. For more information, see Create a project.

• Create a Logstore. For more information, see Create a Logstore.

This topic uses a project named guide-sls-sink-project and a Logstore named guide-sls-sink-logstore as an example.

Step 2: Create and start the SLS 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.

   • Create a 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 configuration wizard, set Service Type to Simple Log Service, configure the following parameters, and then click Save.

   Table 1. Sink (Simple Log Service)

   Parameter	Description	Example
   Project	The Simple Log Service project that you created.	guide-sls-sink-project
   Logstore	The Simple Log Service Logstore that you created.	guide-sls-sink-logstore
   Topic	The method used to generate the SLS Topic. Null Fixed Value Partial Data Template	Empty
   Content	Default Format (JSON) Custom Format (Key/Value)	Default Format (Json)
   Role Name	Grants EventBridge permission to write to SLS. If a role does not exist, follow the on-screen instructions to create one.	sls_eb

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: Test the SLS sink connector

1. On the Tasks page, find the SLS sink connector task and click the source Topic 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, and then click OK.

   Select the Console tab. Set Message Key to sls-sink-k1 and Message Content to sls-sink-v1. For Send to a specified partition, select No.

4. On the Tasks page, find the SLS sink connector task and click the destination project in the Event Target column.

5. On the Logstore page, view the log content.

   The content field of the log record is in JSON format and contains key:"sls-sink-k1" and value:"sls-sink-v1". This confirms that the SLS sink connector successfully wrote data to the destination Logstore.