Torch Log Service
Torch Log Service
Torch Log Service
Document
Product News
Product Overview
Product Billing
User Guide
TLS Agent
AgentLoop
Resource management
Host group
Data collection
LogCollector collection
Introduction to LogCollector
Installation and upgrade
Collect host logs
Collect container logs
LogCollector plugins
Manage LogCollector
general reference
Data import
Cloud product log collection
TOS log collection
DBW log collection
VPN Log collection
VKE log collection
CLB log collection
CloudTrail log collection
CFW log collection
NTA log collection
ALB log collection
APMPlus log collection
veFaaS log collection
VPC log collection
veImageX log collection
TrafficRoute DNS log collection
DDoS log collection
WAF log collection
DCDN log collection
Multi-cloud security platform log collection
APIG log collection
Machine learning task log collection
CR log collection
Volcano Ark application lab log collection
FileNAS log collection
LLM Application Firewall Service log collection
vePFS log collection
Cloud security center log collection
EFS log collection
Data storage
Query and analysis
Configure log indexes
Query log
Analyze logs
Query and analyze logs using third-party tools
Associate external data source for log query
SQL function
SQL syntax
Visualization
Query and analysis charts
Summaries of query and analysis charts
Create dashboard
Use dashboard
Add dashboard filters and variables
Use other systems to analyze logs
Log application
Log applications for product observability
Log applications for large models observability
Coding Agent log app
ModelProvider log app
No-coding Agent log app
Log applications for basic service monitoring
Data processing
Data transformation
Syntax and functions
Global operation function
expression function
Data processing best practices
general reference
Data write processor
Data consumer processor
Data consumption and shipping
Ship logs to TOS
Ship logs to Kafka
Ship logs to Splunk
Alarm
Alarm policy
Alarm notification content
Alarm notification content template
alarm notification channel
Tracing service
Monitor Log Service
API reference
Project management API
Topic management API
Shard management API
Index management API
Log management API
Host group management API
Collection configuration management API
Data import API
Data processor API
Data transformation API
Data backflow API
SQL analysis API
Alarm management API
Consumer group management API
Kafka protocol consumption API
Tag management API
Trace storage API
Account management API
SDK reference
Go SDK
Java SDK
Python SDK
iOS SDK
C++ SDK
CLI reference
Best practices
Data collection
Query and analysis
permission management
Consumption and delivery
Visualization
Alarm
FAQs
Billing and cost FAQ
Data collection FAQ
Data storage FAQ
Index configuration FAQ
Query and analysis FAQ
Scheduled SQL tasks FAQ
Alarm FAQ
- Documentation
Torch Log ServiceBest practicesConsumption and deliveryConsume logs using Spark Streaming
Copy page
Download PDFConsumption and delivery
Consume logs using Spark Streaming
Copy pageDownload PDFConsume logs using Spark Streaming
TLS provides Kafka protocol consumption functionality. You can use the spark-streaming-kafka component to connect to TLS and consume data from TLS to downstream big data components or data warehouses.
Scenario overview
Spark Streaming is a real-time computing framework built on Spark, providing scalable, high-throughput, and fault-tolerant stream processing capabilities based on Spark. Spark Streaming can integrate with multiple data sources. For example, it can integrate with Kafka using the spark-streaming-kafka component to enable consumption of Kafka messages. TLS supports enabling Kafka protocol consumption functionality for specified log topics. Once enabled, Spark Streaming can consume log topics as Kafka Topics, such as consuming to downstream big data components or data warehouses. This is suitable for stream processing or big data storage scenarios.
Prerequisites
- A log project and log topic are created. For detailed instructions, see .
- The Kafka protocol consumption functionality has been enabled for the specified log topic, and you have obtained the Kafka protocol consumption topic ID. For more details, see .
- We recommend using IAM users for access authorization. Before using an IAM user, ensure that the Volcanoengine primary account has created an IAM user and granted consumption-related permissions. For more details, see .
- Access Key for the current login account has been obtained. For more information, see Create Access Key.
Configuration steps
1. Add Maven dependencies
When consuming log data from TLS using Spark Streaming, you need to use spark-streaming-kafka-0-10 provided by Spark Streaming.
Add the relevant spark-streaming-kafka Maven dependencies to your project. Dependency details are as follows:
Tip
The TLS Kafka consumption functionality only supports Kafka protocol version 0.11.0 and above. You need to manually specify the Kafka client version for Spark as 0.11.0 or above.
<dependency>
<groupId>org.apache.spark</groupId>
<artifactId>spark-streaming-kafka-0-10_2.11</artifactId>
<version>${spark.version}</version>
<exclusions>
<exclusion>
<groupId>org.apache.kafka</groupId>
<artifactId>kafka-clients</artifactId>
</exclusion>
</exclusions>
</dependency>
<dependency>
<groupId>org.apache.kafka</groupId>
<artifactId>kafka-clients</artifactId>
<version>1.0.0</version>
</dependency>
2. Configure Spark input stream
Refer to the following sample code to complete the configuration of the Spark input stream. For detailed configuration instructions, see Spark official documentation. For parameter descriptions, see the table below.
The following example demonstrates how to build a Spark input stream to consume log data from the log topic 0fdaa6b6-3c9f-424c-8664-fc0d222c****.
// Build Spark Streaming context
SparkConf conf = new SparkConf().setAppName("TlsDemo").setMaster("local").set("spark.serializer", "org.apache.spark.serializer.KryoSerializer");
conf.registerKryoClasses((Class<?>[]) Arrays.asList(ConsumerRecord.class).toArray());
// Every 5 seconds, the SparkStreaming job collects data received from the data source within the past 5 seconds
JavaStreamingContext jssc = new JavaStreamingContext(conf, Durations.seconds(5));
// Build the Kafka parameter map
// The main requirement is to specify the address of the Kafka cluster to connect to (the broker cluster address list)
Map<String, Object> kafkaParams = new HashMap<>();
// Initial cluster connection address. The format is service address:port number, for example tls-cn-beijing.ivolces.com:9093, where:
// The service address is the TLS service address for the current region. Select the correct service endpoint based on region and network type. For details, see Service address.
// The port number is fixed at 9093.
kafkaParams.put("bootstrap.servers", tlsEndConsumePoint);
// Specify the data type and encoding format for the Kafka output key (default is string type with UTF-8 encoding)
kafkaParams.put("key.deserializer", StringDeserializer.class);
// Specify the data type and encoding format for the Kafka output value (default is string type with UTF-8 encoding)
kafkaParams.put("value.deserializer", StringDeserializer.class);
// Consumer group ID, can be specified arbitrarily
kafkaParams.put("group.id", consumeGroupID);
kafkaParams.put("auto.offset.reset", "latest");
kafkaParams.put("enable.auto.commit", false);
// Security mode, TLS only supports SASL_SSL
kafkaParams.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, "SASL_SSL");
// Authentication mechanism, TLS only supports PLAIN
kafkaParams.put(SaslConfigs.SASL_MECHANISM, "PLAIN");
// Account and password settings:
// Username: projectId: the ProjectID of the Topic's project. net: Enter public for public network access, private for private network access.
String userName = "${projectId}#${net}";
// Password: should be configured as the Volcano Engine account key
// Format is ${access-key-id}#${access-key-secret}, where:
// ${access-key-id} should be replaced with your AccessKey ID.
// ${access-key-secret} should be replaced with your AccessKey Secret.
String passWord = "${access-key-id}#${access-key-secret}";
// Account and password settings
kafkaParams.put(SaslConfigs.SASL_JAAS_CONFIG,"org.apache.kafka.common.security.plain.PlainLoginModule " +"required username=\"" + userName + "\" password=\"" + passWord + "\";");
// Build the topic set
Collection<String> topics = new HashSet<>();
// The topic to consume, format is out-xxxxxxxxx
topics.add("out-0fdaa6b6-3c9f-424c-8664-fc0d222c****");
try {
// Retrieve data from Kafka
final JavaInputDStream<ConsumerRecord<String, String>> stream = KafkaUtils.createDirectStream(jssc, LocationStrategies.PreferConsistent(),
ConsumerStrategies.Subscribe(topics, kafkaParams));
// todo: Consume to downstream big data components
stream.print();
jssc.start();
jssc.awaitTermination();
jssc.close();
} catch (Exception e) {
e.printStackTrace();
}
The required settings are as follows:
Configuration item | Description |
Configuration item | Description |
bootstrap.servers | Initial cluster connection address. The format is service address:port number, for example tls-cn-beijing.ivolces.com:9093.
|
consumeTopic | The Kafka Topic to subscribe to. This should be configured as the TLS Kafka protocol consumption topic ID, in the format out-topic ID, for example out-0fdaa6b6-3c9f-424c-8664-fc0d222c****. You can view and copy the Kafka protocol consumption topic ID on the Topic details page in the TLS console. |
SECURITY_PROTOCOL_CONFIG | Security mode, should be specified as SASL_SSL. |
SASL_MECHANISM | Authentication mechanism, specify as PLAIN. |
SASL_JAAS_CONFIG | Authentication information for the current logged-in user. Note:
|
Last updated: 2026.10.29 16:57:56
All rights reserved © Beijing Volcano Engine Technology Co., Ltd2024
