Docs

Volcano Engine Function as a Service

Data processing

Synchronizing Kafka data to ES by using veFaaS

The following content introduces how to use the microservices application template that is built on Volcano Engine Function as a Service (veFaaS) to quickly launch a data processing service, which can read data from Message Queue for Kafka to Cloud Search (ES).

Scenario introduction

Data processing services are mainly used for log processing and analysis, helping enterprises quickly identify and resolve issues and improve operational efficiency. In large systems, to ensure proper collection and analysis of complex log data, the common approach is using Kafka streaming architecture for large-scale asynchronous data collection. In turn, the collected log data is read by consumer components from Kafka and written in Elasticsearch. veFaaS is an alternative to open-source LogStash. Self-built LogStash solutions face issues such as data loss, poor collection performance, high resource consumption, and steep learning curves. The veFaaS is a Serverless data processing solution that is simpler, lighter, and provides advantages such as auto-scale, low cost, high throughput, and low latency.

Prerequisites

For Virtual Private Cloud (VPC)

Permission for using Volcano Engine Virtual Private Cloud has been obtained.

A VPC, subnets, and security groups have been created in the region for function deployment. For details, see Creating Virtual Private Cloud, Creating subnet, and Creating security group.

tip

We recommend confining data transmission within the VPC, i.e., creating Kafka instances, ES instances, and veFaaS microservices applications within the same VPC.

Message Queue for Kafka

Permission for using Message Queue for Kafka on Volcano Engine has been obtained.

A Kafka instance and the corresponding SASL user, Topic, and Consumer Group have been created. For details, see Creating instance, Creating Topic, Creating Group, and Creating SASL user.

Cloud Search service (ES)

Permission for using Volcano Engine Cloud Search has been obtained.

An ES instance has been created. Note that there are requirements for selecting a specific ES version and setting a login password. For details, see Creating instances.

An ES index has been created. For details, see Creating ES index.

Procedure

Step 1: Create a microservices application

Select the vefaas-native-microservice-kafka-to-es-exporter function template to create the application instantly.

In the top navigation bar, select the target region.

On the Function list page, click Create function, and then select Create [Microservices applications].

Configure the options on the page as required. The table below contains descriptions of the key options on the page. The rest can keep the default settings.

Parameter	Description
Deployment method	Select function template: vefaas-native-microservice-kafka-to-es-exporter.
Network	Select the Virtual Private Cloud, subnet, and Security groups you created as required in the prerequisites above.

Click Create to create the function.

Step 2: Update the configuration file

Update the config.json file with your Kafka and ES configurations.

On the Code tab of the function, click the download icon in the upper right to download the code package.

Extract the code package.

Update the config.json file.

Copy the config_template.json file in the code package to a new file named config.json.

Modify the config.json file as needed.

{
// Kafka related configurations
"kafka": {
// Kafka access endpoint information
"bootstrap.servers": "127.0.0.1:8092",
// Security protocol
"security.protocol": "SASL_PLAINTEXT",
// Whether to enable debug mode
"debug": false,
// Name of the topic whose messages are to be synchronized
"topics": "my-topic",
// Consumer group ID
"consumer.group.id": "test",
// SASL related configurations
"sasl": {
// Whether to enable SASL
"enabled": true,
// Authentication algorithm
"mechanism": "PLAIN",
// Username
"username": "test-user",
// Password
"password": "test-password"
}
},
// Elasticsearch related configurations
"elasticsearch": {
// ES access address
"host": "127.0.0.1:9200",
// Username
"username": "test-user",
// Password
"password": "test-password",
// Name of the index to write data into
"index": "test-index"
}
}

Field	Description
Kafka-related configuration
security.protocol	Security protocol. Available values: PLAINTEXT SASL_PLAINTEXT (default) SASL_SSL
bootstrap.servers	Kafka endpoint information. Use the default endpoint if security.protocol is PLAINTEXT. If security.protocol is set to SASL_PLAINTEXT, enter the endpoint for SASL_PLAINTEXT. Use the SASL_SSL endpoint if security.protocol is SASL_SSL. tip You can view endpoint address in Instance Details > Service access on the Message Queue for Kafka console. Setting the address to a private IP is recommended.
debug	Whether to enable debug mode. true: enable. false: disable
topics	A list of topics from which messages will be read. Separate multiple topics with \",\" e.g., \"topic1,topic2\".
consumer.group.id	Corresponding consumer group ID.
sasl	SASL-related configuration. enabled: whether SASL is enabled. When security.protocol is SASL_PLAINTEXT or SASL_SSL, SASL-related configurations are required and the value of the field enabled needs to be true. mechanism: authentication algorithm. Currently, only PLAIN or SCRAM-SHA-256 are supported. Select one based on password type. username & password: username and password. You can view them in Instance Details > User Management on the Message Queue for Kafka console.
ES-related configuration
host	ES address. tip You can view the address on the Basic information page of the instance on the Cloud Search console. Setting the address to a private IP with HTTPS as the protocol is recommended.
username	User name. You can view the user name on the Basic information page of the instance on the Cloud Search console. The default user name is admin.
password	Password. The login password that is required when creating the instance. You can reset the password on the Basic information page of the instance on the Cloud Search console.
index	The index name under which data will be written needs to be created in advance in ES.

Run sh build_and_zip.sh to start code compilation and packaging.

warning

It must be executed on a Linux platform. Cross-compilation is not supported.

Upload code.

On the Code tab of the function, click Upload > Upload local file to upload the code package generated in step 4.

Step 3: Deploy the application

Deploy the current code and configuration of the function online.

On the function details page, click Release in the upper right corner.

Select Latest for the function version. Keep other parameters as default.

Click OK. Wait until the function release is finished.

Validating results

Send a message using Kafka Topic.

On the Message Queue for Kafka console, click the instance to go to the Instance details > Topic management tab, then click Send message in the Action column of the corresponding topic. Send the word hello as a message.

Use ES API to check if the document was created successfully.

# For HTTPS mode, you can specify --insecure for quick verification
# test-index is the index name used in this example
curl https://elasticsearch-o-xxxxx.escloud.volces.com:9200/test-index/_search --insecure -u admin:xxxx

The expected ES response is shown below:

{
"took": 3,
"timed_out": false,
"_shards": {
"total": 1,
"successful": 1,
"skipped": 0,
"failed": 0
},
// The created document includes information such as the topic, value, partition, offset and timestamp of Kafka messages. The "value": "hello" field indicates that message synchronization completed successfully.
"hits": {
"total": {
"value": 1,
"relation": "eq"
},
"max_score": 1.0,
"hits": [
{
"_index": "test-index",
"_type": "_doc",
"_id": "0xKjb5IBOCLUIact9aQp",
"_score": 1.0,
"_source": {
"header": {},
"key": "",
"offset": 0,
"partition": 7,
"timestamp": "2024-10-09 12:57:48.631 +0800 CST",
"topic": "my-topic",
"value": "hello"
}
}
]
}
}

Relevant operations

Once deployed, veFaaS automatically scales in or out based on workload. No manual adjustment to the instance count required. You can also set scaling policies as needed. For details, refer to Elastic scaling.

Last updated: 2026.08.19 14:06:07