ApsaraMQ for Kafka:Create an Elasticsearch sink connector

Step 1: Create target service resources

• Create an instance and an index in the Elasticsearch console. For more information, see Getting started.

• Add the CIDR block of the Function Compute service to the whitelist of your Elasticsearch instance. This step is required only when connecting via a Virtual Private Cloud (VPC) and is not needed for internet connections. For more information, see Configure a public or private IP address whitelist for an instance.

Step 2: Create an Elasticsearch 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.

4. In the Create Task page, configure the Task Name and Description parameters. Then, follow the on-screen instructions to configure other parameters.

   • Configure the task

     1. In the Source step, set the Data Provider parameter to ApsaraMQ for Kafka and follow the on-screen instructions to configure other parameters. Then, click Next Step. The following table describes the parameters.

        Parameter	Description	Example
        Region	The region where the ApsaraMQ for Kafka instance resides.	China (Hangzhou)
        ApsaraMQ for Kafka Instance	The ID of the ApsaraMQ for Kafka instance in which the data that you want to route are produced.	alikafka_post-cn-9hdsbdhd****
        Topic	The topic on the ApsaraMQ for Kafka instance in which the data that you want to route are produced.	guide-sink-topic
        Group ID	The ID of the group on the ApsaraMQ for Kafka instance in which the data that you want to route are produced. Quickly Create: The system automatically creates a group whose ID is in the GID_EVENTBRIDGE_xxx format. Use Existing Group: Select the ID of an existing group that is not in use. If you select an existing group that is in use, the publishing and subscription of existing messages are affected.	Use Existing Group
        Consumer Offset	Latest Offset: Messages are consumed from the latest offset. Earliest Offset: Messages are consumed from the earliest offset.	Latest Offset
        Network Configuration	If cross-border data transmission is required, select Self-managed Internet. In other cases, select Basic Network.	Basic Network
        Data Format	The data format feature is used to encode binary data delivered from the source into a specific data format. Multiple data formats are supported. If you do not have special requirements on encoding, specify Json as the value. Json: Binary data is encoded into JSON-formatted data based on UTF-8 encoding and then put into the payload. This is the default value. Text: Binary data is encoded into strings based on UTF-8 encoding and then put into the payload. Binary: Binary data is encoded into strings based on Base64 encoding and then put into the payload.	Json
        Messages	The maximum number of messages that can be sent in each function invocation. Requests are sent only when the number of messages in the backlog reaches the specified value. Valid values: 1 to 10000.	2000
        Interval (Unit: Seconds)	The time interval at which the function is invoked. The system sends the aggregated messages to Function Compute at the specified time interval. Valid values: 0 to 15. Unit: seconds. The value 0 indicates that messages are sent immediately after aggregation.	3

     2. In the Filtering step, define an event pattern to filter requests. For more information, see Event patterns.

     3. In the Transformation step, configure data cleansing for complex 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 Alibaba Cloud Elasticsearch acs.elasticSearch for Service Type and configure the following parameters.

        Parameter	Description	Example
        Elasticsearch Cluster	The Elasticsearch instance that you created.	es-cn-pe336j0gj001e****
        Cluster Logon Name	The logon name for the instance, which is elastic by default.	elastic
        Instance logon password	The password that you configured when you created the instance.	******
        Index Name	The name of the index that you created. For more information about how to create an index, see Getting started. The index name can be a string constant or a JSONPath variable, such as product_info or $.data.key.	product_info
        Document Type	The type of the data document. This can be a string constant or a variable extracted by a JSONPath expression. Examples: _doc or $.data.key. Note This parameter can be configured only for Elasticsearch instance versions earlier than 7.0. The default value is the constant _doc.	_doc
        Document	Select whether to deliver the complete event or a partial event to Elasticsearch. For partial events, you must configure a JSONPath extraction rule.	Complete Event
        Network Configuration	VPC: Deliver Kafka messages to Elasticsearch through a VPC. Public Network: Deliver Kafka messages to Elasticsearch over the internet.	Public Network
        VPC	The VPC that contains the Elasticsearch instance. This parameter is required only when Network Configuration is set to VPC.	vpc-bp17fapfdj0dwzjkd****
        vSwitch	The vSwitch to which the Elasticsearch instance belongs. This parameter is required only when Network Configuration is set to VPC.	vsw-bp1gbjhj53hdjdkg****
        Security Group	The 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 error handling method. For more information, see Retries and dead-letter queues.

5. After you complete the configuration, click Save. On the Tasks page, find the Elasticsearch sink connector task that you created. The Status column shows Starting. When the status changes to Running, the connector is created and ready.

Step 3: Test the Elasticsearch sink connector

1. On the Tasks page, click the source topic in the Event Source column of the Elasticsearch sink connector task.

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

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

   Set the sending method to Console. Set Message Key to es-sink-k1 and Message Content to {"esk1":1,"esk2":"v2"}. For Send to Specified Partition, select No.

4. Log on to the Elasticsearch console and access the instance via Kibana. For more information, see Getting started.

5. On the Kibana console, run the following command to view the data insertion result.

   GET /your-index-name/_search

   Data insertion result: The query returns a 200 OK status and includes one document, where _index is product_info and _id is 1717558528. The _source field contains the topic, partition, offset, timestamp, headers, key, and value fields. The key is es-sink-k1 and the value is {"esk1": 1, "esk2": "v2"}. This confirms that the data was successfully written to Elasticsearch.