All Products
Search
Document Center

Simple Log Service:Use Kafka to consume data based on SPL statements

Last Updated:Oct 23, 2023

This topic describes how to use Kafka to consume data based on Simple Log Service Processing Language (SPL) statements.

Supported regions

The SPL-based consumption feature is available only in the China (Guangzhou) region.

How it works

If you add an SPL statement to a virtual topic, the consumption-compatible Kafka interface processes data based on the SPL statement after if receives the virtual topic. Then, Kafka consumers obtain the data that is processed based on the SPL statement.

kafkaConsumer -> [Virtual Topic] -> Actual Logstore
Important
  • You can use only Kafka official SDK libraries (Kafka SDK for Java and librdkafka) to perform SPL-based consumption.

  • Issues exist if you use the third-party library python-kafka to perform SPL-based consumption. If you want to consume data in Python, use the official Python" commentTypeId="40" timestamp="2023-08-31T15:46:17+08:00" authorId="8986" uuid="b357b32d-3f35-4162-b32c-41e368c93195">Python library confluent_kafka.

Create a virtual topic

  1. Modify the config.json file.

    {
        "Project" : "Project",
        "Endpoint" : "Endpoint",
        "Port" : 10012,
        "AccessKeyId" : "",
        "AccessKeySecret" : "",
        "Topic" : "logstore.0",
        "Query" : "SPL statement"
    }

    Parameter

    Description

    Project

    The project from which data is consumed.

    Endpoint

    The Simple Log Service endpoint. For more information, see Endpoints.

    Port

    The port number. Valid values:

    • Alibaba Cloud internal network: 10011

    • Internet: 10012

    AccessKeyId

    The AccessKey ID of your Alibaba Cloud account.

    We recommend that you use the AccessKey pair of a Resource Access Management (RAM) user that is granted the write permissions on the Simple Log Service project. For information about how to grant a RAM user the write permissions on a project, see Use custom policies to grant permissions to a RAM user. For information about how to obtain an AccessKey pair, see AccessKey pair.

    AccessKeySecret

    The AccessKey secret of your Alibaba Cloud account.

    We recommend that you use the AccessKey pair of a RAM user that is granted the write permissions on the Simple Log Service project. For information about how to grant a RAM user the write permissions on a project, see Use custom policies to grant permissions to a RAM user. For information about how to obtain an AccessKey pair, see AccessKey pair.

    Topic

    The topic. Set the value to ${logstore}.<Identifier string>. The identifier string must be in 0 to 31 characters in length.

    Query

    The SPL statement.

  2. Download the kafka_admin_tool.

    Important

    You can perform this operation only on a Linux server.

    wget https://sls-resource.oss-cn-shanghai.aliyuncs.com/tools/kafka_admin_tool
  3. Create a virtual topic.

    ./kafka_admin_tool create_virtual_topic -f config.json

View the virtual topic

You can run the following command to view the virtual topic:

./kafka_admin_tool info_virtual_topic -f config.json

FAQ

  • How do I view consumption errors?

    If consumption errors occur, a KafkaConsumer cannot read the error messages during SPL-consumption. You must enable the collection of service logs in Simple Log Service and view the error messages in detailed logs. For more information, see Enable the service log feature and Detailed logs.

    You can execute the following SQL statement in the detailed logs to find the error information.

    Logstore: <Logstore name> and Method: PullData and UserAgent: kafka-forwarderrdkafka 
  • How do I update the SPL statement of a virtual topic?

    You must delete the virtual topic and then create a topic to update the SPL statement. Procedure:

    1. Stop the consumers of the virtual topic.

    2. Deletes the virtual topic.

      ./kafka_admin_tool delete_virtual_topic -f config.json
    3. Update the SPL statement in the config.json file.

    4. Create a virtual topic.

      Note

      After you recreate a virtual topic, you must wait 10 minutes before the SPL statement takes effect.

      ./kafka_admin_tool create_virtual_topic -f config.json