日志服务
日志服务
日志服务
文档指南
请输入
用户指南
TLS Agent
资源管理
数据采集
LogCollector 采集
LogCollector 简介
安装与升级
采集容器日志
LogCollector 插件
管理 LogCollector
云产品日志采集
TOS 日志采集
CDN 日志采集
DBW 日志采集
VPN 日志采集
VKE 日志采集
CLB 日志采集
CloudTrail 日志采集
CFW 日志采集
NTA 日志采集
ALB 日志采集
APMPlus 日志采集
veFaaS 日志采集
VPC 日志采集
veImageX 日志采集
TrafficRoute DNS 日志采集
DDos 日志采集
WAF 日志采集
DCDN 日志采集
多云安全平台日志采集
APIG 日志采集
机器学习任务日志采集
采集火山方舟应用实验室日志
CR 日志采集
FileNAS 日志采集
大模型应用防火墙日志采集
vePFS 日志采集
云安全中心日志采集
EFS 日志采集
检索分析
使用第三方工具检索分析日志
关联外部数据源检索分析日志
SQL 函数
可视化
日志应用
云产品观测日志应用
大模型观测日志应用
Coding Agent 日志应用
ModelProvider 日志应用
No-coding Agent 日志应用
Framework 日志应用
基础服务监控日志应用
数据处理
数据加工
消费与投递
告警
Trace 存储
API 参考
日志管理
机器组管理
采集配置管理
数据处理器
定时 SQL 分析
告警管理
消费组管理
Trace 存储
SDK 参考
Go SDK
Java SDK
Android SDK
Node.js SDK
CLI 参考
最佳实践
消费与投递
可视化
常见问题
数据采集常见问题
索引配置常见问题
- 文档首页
日志服务最佳实践消费与投递通过 Flink 处理日志
复制全文
下载 pdf通过 Flink 处理日志
文档反馈
问问助手日志服务提供对接 Apache Flink 插件，您可以使用插件将日志服务中的数据消费到下游的大数据组件或者数据仓库。
场景概述
Apache Flink 是一个在有界数据流和无界数据流上进行有状态计算的分布式处理引擎和框架。火山引擎日志服务 TLS 提供 Flink Connector 插件 ve-tls-log-flink-connector，用于对接 Apache Flink，包含如下部分：
- FlinkLogSink（生产者）：将数据写入日志服务，支持 At-Least-Once 语义。
- FlinkLogSource（消费者）：从日志服务中读取数据，支持 Exactly-Once 语义，支持 Shard 负载均衡和自动感知 Shard 变更。
前提条件
- 已创建日志项目（Project）和日志主题（Topic）。详情请参考日志项目。
- 已获取当前登录账号的密钥 Access Key。详细信息请参考创建密钥。
- 使用 IAM 子用户时，需确保已完成相关授权：
- 已授权 IAM 用户访问日志服务资源，详情请参考可授权的操作。
- 已通过 IAM 策略对依赖的 API 进行授权，详细请参考配置身份认证信息。
- 分类依赖的 API说明Source通过 DescribeCursor 接口获取日志下载游标。通过 ConsumeLogs 接口获取数据消费日志。调用 DescribeShards 接口获取日志主题的分区列表。调用 CreateConsumerGroup 接口创建日志服务消费组（设置消费组时使用）。调用 ModifyCheckpoint 接口为指定消费组重置指定分区的消费位点，即更新消费组 Checkpoint。调用 DescribeCheckpoint 接口查看指定消费组在指定日志分区上的当前消费位点，即获取消费组 Checkpoint。调用 DescribeConsumerGroups 接口获取一个日志项目下的所有消费组信息。Sink调用 PutLogs 接口上传日志到指定的日志主题中。
添加 Maven 依赖
通过 Flink 消费火山引擎日志服务的日志数据时，需要在项目中添加 ve-tls-log-flink-connector 相关的 Maven 依赖。
依赖的详细信息如下：
<dependency>
<groupId>com.volcengine</groupId>
<artifactId>ve-tls-log-flink-connector</artifactId>
<version>1.0.0</version>
</dependency>
FlinkLogSource（消费者）
FlinkLogSource 提供了订阅日志服务中某个日志项目下单个或多个日志主题的能力，支持 Exactly-Once 语义。使用时无需关心日志主题中分区（Shard）数量的变化，FlinkLogSource 会自动感知。
Flink 中每个子任务负责消费部分 Shard，如果 Shard 发生分裂，子任务消费的 Shard 也会随之改变。Shard 与任务之间的默认分配关系为：
shardId % numberOfParallelSubtasks == indexOfThisSubtask
说明
- 支持通过设置自定义 [ShardAssigner] 实现个性化的分配方式。
- Flink 子任务数量和日志服务日志主题中的分区（Shard）数量是独立的。
- 如果 Shard 数量多于 Flink 子任务数量，每个 Flink 子任务不重复地消费多个 Shard。
- 如果 Shard 数量少于 Flink 子任务数量，部分 Flink 子任务会空闲，直到产生新的 Shard。
1 配置启动参数
以下是一个简单的消费示例，所有配置常量均可在 LogConstants 中查找。
Properties configProps = new Properties();
// 设置访问日志服务的 Endpoint
configProps.put(LogConstants.LOG_ENDPOINT, "https://tls-cn-beijing.volces.com");
// 设置访问日志服务的 Region
configProps.put(LogConstants.LOG_REGION, "cn-beijing");
// 设置访问日志服务的 AK/SK
configProps.put(LogConstants.LOG_ACCESS_KEY, "");
configProps.put(LogConstants.LOG_ACCESS_SECRET, "");
// 设置日志服务的 Project
configProps.put(LogConstants.LOG_PROJECT, "your-project-id");
// 设置日志服务的 Topic
configProps.put(LogConstants.LOG_TOPIC, "your-topic-id");
// 设置消费起始位置
configProps.put(LogConstants.LOG_CONSUMER_BEGIN_POSITION, LogConstants.LOG_END_CURSOR);
// 设置每次读取的 LogGroup 个数，默认 100
configProps.put(LogConstants.LOG_MAX_NUMBER_PER_FETCH, "100");
// 设置日志数据的反序列化方法
LogDeserializationFunction<FlinkLogGroup> deserializer = new LogGroupListDeserializer();
final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
DataStream<FlinkLogGroup> logStream = env.addSource(
new FlinkLogSource<>(deserializer, configProps));
LogGroupListDeserializer 的参考实现如下：
public class LogGroupListDeserializer implements LogDeserializationFunction<FlinkLogGroup> {
@Override
public List<FlinkLogGroup> deserialize(PullLogsResult record) {
List<FlinkLogGroup> groups = new ArrayList<>();
for (PutLogRequest.LogGroup logGroup : record.getLogGroupList().getLogGroupsList()) {
FlinkLogGroup flinkLogGroup = new FlinkLogGroup();
flinkLogGroup.setSource(logGroup.getSource());
int logCount = logGroup.getLogsCount();
List<LogItem> items = new ArrayList<>(logCount);
for (int i = 0; i < logCount; i++) {
PutLogRequest.Log log = logGroup.getLogs(i);
LogItem item = new LogItem();
for (int j = 0; j < log.getContentsCount(); j++) {
item.addContent(log.getContents(j).getKey(), log.getContents(j).getValue());
}
items.add(item);
}
flinkLogGroup.setLogs(items);
groups.add(flinkLogGroup);
}
return groups;
}
@Override
public TypeInformation<FlinkLogGroup> getProducedType() {
return GenericTypeInfo.of(FlinkLogGroup.class);
}
}
2 设置消费起始位置
通过设置 LogConstants.LOG_CONSUMER_BEGIN_POSITION 参数的值，可以指定消费日志主题中分区的起始位置，具体取值如下：
参数值 | 说明及配置示例 |
LogConstants.LOG_BEGIN_CURSOR | 从 Shard 的头部开始消费，即从最旧的数据开始。 configProps.put(LogConstants.LOG_CONSUMER_BEGIN_POSITION, LogConstants.LOG_BEGIN_CURSOR); |
LogConstants.LOG_END_CURSOR | 从 Shard 的尾部开始消费，即从最新的数据开始。 configProps.put(LogConstants.LOG_CONSUMER_BEGIN_POSITION, LogConstants.LOG_END_CURSOR); |
LogConstants.LOG_FROM_CHECKPOINT | 以消费组的 Checkpoint 作为消费的起始位置。当从 Checkpoint 获取起始位置时，必须提供消费组名称。 configProps.put(LogConstants.LOG_CONSUMER_BEGIN_POSITION, LogConstants.LOG_FROM_CHECKPOINT);
configProps.put(LogConstants.LOG_CONSUMER_BEGIN_POSITION, LogConstants.LOG_FROM_CHECKPOINT); configProps.put(LogConstants.LOG_CONSUMER_DEFAULT_POSITION, LogConstants.LOG_END_CURSOR); |
1512439000 | Unix 时间戳，一个整型数值的字符串，表示从 1970-01-01 到现在的秒数，用于消费 Shard 中该时间点开始写入的数据。 configProps.put(LogConstants.LOG_CONSUMER_BEGIN_POSITION, "1512439000"); |
3 消费进度监控（可选）
FlinkLogSource 支持设置消费进度监控，消费进度即获取每个 Shard 实时的消费位置。设置消费组名称后，Connector 会自动创建消费组（如果不存在），并将 Checkpoint 同步到日志服务端，用户可以在日志服务控制台查看消费进度。
configProps.put(LogConstants.LOG_CONSUMERGROUP, "your-consumer-group-name");
4 消费组 Checkpoint 提交模式
通过配置 LogConstants.LOG_CHECKPOINT_MODE 可以指定消费组 Checkpoint 的提交模式，支持以下三种：
- 配置示例：
- configProps.put(LogConstants.LOG_CHECKPOINT_MODE, CheckpointMode.ON_CHECKPOINTS.name());configProps.put(LogConstants.LOG_CHECKPOINT_MODE, CheckpointMode.PERIODIC.name());configProps.put(LogConstants.LOG_CHECKPOINT_MODE, CheckpointMode.DISABLED.name());
- 配置说明：
- 模式说明ON_CHECKPOINTS（默认值）当开启 Flink 的 Checkpointing 功能，每个 Shard 的消费进度会保存在 Flink 的 State 中，同时会提交到日志服务端。当作业 Failover 时，会从 Flink 的 State 中恢复；如果不存在对应的 Checkpoint，会从服务端保存的最新 Checkpoint 恢复。final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);env.enableCheckpointing(5000);DISABLEDCheckpoint 不会被提交到日志服务端。PERIODICCheckpoint 被定时提交到日志服务端，与 Flink Checkpointing 完全独立。仅与消费组提交 Checkpoint 到日志服务有关，不影响 Flink 的 State。支持自定义提交间隔，默认为 10 秒提交一次。configProps.put(LogConstants.LOG_COMMIT_INTERVAL_MILLIS, "10000");
5 消费多个 Topic
FlinkLogSource 支持同时消费同一 Project 下的多个 Topic：
List<String> topics = Arrays.asList("topic-id-1", "topic-id-2", "topic-id-3");
FlinkLogSource<FlinkLogGroup> source = new FlinkLogSource<>(
"your-project-id", topics, deserializer, configProps);
DataStream<FlinkLogGroup> logStream = env.addSource(source);
6 自定义 Shard 分配策略
默认的 Shard 分配策略为 shardId % numberOfParallelSubtasks，可以通过实现 ShardAssigner 接口自定义分配策略：
FlinkLogSource<FlinkLogGroup> source = new FlinkLogSource<>(deserializer, configProps);
source.setShardAssigner(new ShardAssigner() {
@Override
public int assign(TopicShardMeta shard, int numParallelSubtasks) {
return shard.getShardId() % numParallelSubtasks;
}
});
7 其他配置参数
参数 | 说明 | 默认值 |
LOG_MAX_NUMBER_PER_FETCH | 每次读取的 LogGroup 个数。 | 100 |
LOG_FETCH_DATA_INTERVAL_MILLIS | 拉取数据的时间间隔（毫秒）。 | 100 |
LOG_SHARDS_DISCOVERY_INTERVAL_MILLIS | Shard 发现周期（毫秒）。 | 60000 |
SOURCE_QUEUE_SIZE | Source 内部队列大小。 | 1 |
SOURCE_IDLE_INTERVAL | Source 空闲等待间隔（毫秒）。 | 10 |
STOP_TIME | 停止消费的时间戳（Unix 秒），设置后消费到该时间点自动停止。 | / |
8 Source 完整示例
以下代码片段仅为示例，使用时需根据实际环境填入 ENDPOINT、REGION、AK、SK 等信息。
import com.alibaba.fastjson.JSONObject;
import com.volcengine.model.tls.LogContent;
import com.volcengine.model.tls.LogItem;
import com.volcengine.model.tls.pb.PutLogRequest;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.java.typeutils.GenericTypeInfo;
import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.configuration.CheckpointingOptions;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.runtime.state.hashmap.HashMapStateBackend;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import com.volcengine.service.tls.flink.config.LogConstants;
import com.volcengine.service.tls.flink.entry.CheckpointMode;
import com.volcengine.service.tls.flink.entry.FlinkLogGroup;
import com.volcengine.service.tls.flink.entry.LogDeserializationFunction;
import com.volcengine.service.tls.flink.entry.PullLogsResult;
import java.util.ArrayList;
import java.util.List;
import java.util.Properties;
public class FlinkLogSourceSample {
private static final String ENDPOINT = "http://tls-cn-beijing.volces.com";
private static final String ACCESS_KEY_ID = "";
private static final String ACCESS_KEY_SECRET = "";
private static final String PROJECT = "";
private static final String TOPIC = "";
public static void main(String[] args) throws Exception {
final ParameterTool params = ParameterTool.fromArgs(args);
// For local testing
Configuration conf = new Configuration();
conf.setString(CheckpointingOptions.CHECKPOINTS_DIRECTORY,
"file:///tmp/tls-log-flink-connector/flink");
final StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironment(2, conf);
env.getConfig().setGlobalJobParameters(params);
env.setParallelism(2);
env.enableCheckpointing(5000);
env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
env.getCheckpointConfig().enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
env.setStateBackend(new HashMapStateBackend());
env.getCheckpointConfig().setCheckpointStorage("file:///tmp/tls-log-flink-connector/backend");
Properties configProps = new Properties();
configProps.put(LogConstants.LOG_ENDPOINT, params.get(LogConstants.LOG_ENDPOINT, ENDPOINT));
configProps.put(LogConstants.LOG_REGION, "cn-beijing");
configProps.put(LogConstants.LOG_ACCESS_KEY, params.get(LogConstants.LOG_ACCESS_KEY, ACCESS_KEY_ID));
configProps.put(LogConstants.LOG_ACCESS_SECRET, params.get(LogConstants.LOG_ACCESS_SECRET, ACCESS_KEY_SECRET));
configProps.put(LogConstants.LOG_TOPIC, params.get(LogConstants.LOG_TOPIC, TOPIC));
configProps.put(LogConstants.LOG_PROJECT, params.get(LogConstants.LOG_PROJECT, PROJECT));
configProps.put(LogConstants.LOG_MAX_NUMBER_PER_FETCH, "10");
configProps.put(LogConstants.LOG_CONSUMER_BEGIN_POSITION, LogConstants.LOG_FROM_CHECKPOINT);
configProps.put(LogConstants.LOG_CONSUMERGROUP, "tls-flink-source-1");
configProps.put(LogConstants.LOG_CHECKPOINT_MODE, CheckpointMode.ON_CHECKPOINTS.name());
configProps.put(LogConstants.LOG_COMMIT_INTERVAL_MILLIS, "10000");
// configProps.put(LogConstants.STOP_TIME, "1627878020");
LogGroupListDeserializer deserializer = new LogGroupListDeserializer();
DataStream<FlinkLogGroup> stream = env.addSource(
new FlinkLogSource<>(deserializer, configProps));
DataStream<String> res = stream.flatMap((FlatMapFunction<FlinkLogGroup, String>) (value, out) -> {
for (LogItem log : value.getLogs()) {
JSONObject jsonObject = new JSONObject();
for (LogContent content : log.getContents()) {
jsonObject.put(content.getKey(), content.getValue());
}
jsonObject.put("__source__", value.getSource());
out.collect(jsonObject.toJSONString());
}
}).returns(String.class);
res.writeAsText("log-" + System.nanoTime());
env.execute("Flink consumer");
}
public static class LogGroupListDeserializer implements LogDeserializationFunction<FlinkLogGroup> {
@Override
public List<FlinkLogGroup> deserialize(PullLogsResult record) {
List<FlinkLogGroup> groups = new ArrayList<>();
for (PutLogRequest.LogGroup logGroup : record.getLogGroupList().getLogGroupsList()) {
FlinkLogGroup flinkLogGroup = new FlinkLogGroup();
flinkLogGroup.setSource(logGroup.getSource());
int logCount = logGroup.getLogsCount();
List<LogItem> items = new ArrayList<>(logCount);
for (int i = 0; i < logCount; i++) {
PutLogRequest.Log log = logGroup.getLogs(i);
LogItem item = new LogItem();
for (int j = 0; j < log.getContentsCount(); j++) {
item.addContent(log.getContents(j).getKey(), log.getContents(j).getValue());
}
// 用户可根据需要读取tag里的数据 logGroup.getLogTagsList()
items.add(item);
}
flinkLogGroup.setLogs(items);
groups.add(flinkLogGroup);
}
return groups;
}
@Override
public TypeInformation<FlinkLogGroup> getProducedType() {
return GenericTypeInfo.of(FlinkLogGroup.class);
}
}
}
FlinkLogSink（生产者）
FlinkLogSink 用于将数据写入火山引擎日志服务。
Sink 支持 Flink At-Least-Once 语义，这意味着在作业失败的情况下，写入日志服务中的数据可能会重复，但不会丢失。
1 基本用法
final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setParallelism(3);
env.enableCheckpointing(5000);
DataStream<String> stream = env.addSource(new RandomStringSource());
FlinkLogSink<String> flinkLogSink =
new FlinkLogSink.Builder<String>(
"https://tls-cn-beijing.volces.com", // endpoint
"cn-beijing", // region
"your-access-key-id", // accessKeyId
"your-access-key-secret") // accessKeySecret
.setSerializationFunction(new SimpleStringFunction("your-topic-id"))
.build();
stream.addSink(flinkLogSink);
env.execute("flink log producer");
2 自定义序列化方法
使用 FlinkLogSink 需要实现 LogSerializationFunction 接口，将数据序列化为 FlinkLogGroup：
public class SimpleStringFunction implements LogSerializationFunction<String> {
private final String topic;
public SimpleStringFunction(String topic) {
this.topic = topic;
}
@Override
public FlinkLogGroup serialize(String element) {
FlinkLogGroup group = new FlinkLogGroup();
group.setTopic(topic);
LogItem log = new LogItem();
log.setTime(System.currentTimeMillis());
log.addContent("content", element);
group.getLogs().add(log);
return group;
}
}
3 Producer 配置参数（可选）
FlinkLogSink 使用 Builder 模式构建，支持以下可选配置，不设置时使用默认值：
方法 | 说明 | 默认值 |
setTotalSizeInBytes(int) | Producer 可使用的内存总大小。 | 100MB |
setMaxThreadCount(int) | 发送数据的 I/O 线程数。 | max(CPU核数/2, 1) |
setMaxBatchSizeBytes(int) | 缓存发送的包的最大值。 | 5MB |
setMaxBatchCount(int) | 缓存发送的包中日志的数量。 | 4096 |
setLingerMs(int) | 日志数据被缓存发送的时间（毫秒）。 | 3000 |
setRetryCount(int) | 发送失败后的重试次数。 | 3 |
setMaxReservedAttempts(int) | 最大保留的重试次数。 | 3 |
setMaxBlockMs(long) | 发送阻塞最大等待时间（毫秒）。 | 0（不阻塞） |
示例：
FlinkLogSink<String> flinkLogSink =
new FlinkLogSink.Builder<String>(
endpoint, region, accessKeyId, accessKeySecret)
.setSerializationFunction(new SimpleStringFunction(topicId))
.setTotalSizeInBytes(104857600)
.setMaxBatchCount(4096)
.setLingerMs(2000)
.setRetryCount(3)
.build();
4 Sink 完整示例
以下代码片段仅为示例，使用时需根据实际环境填入 ENDPOINT、REGION、AK、SK 等信息。
import com.volcengine.model.tls.LogItem;
import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.ParallelSourceFunction;
import com.volcengine.service.tls.flink.config.LogConstants;
import com.volcengine.service.tls.flink.entry.FlinkLogGroup;
import com.volcengine.service.tls.flink.entry.LogSerializationFunction;
import java.util.Random;
/** TLS sink sample test. */
public class FlinkLogSinkTest {
private static final String DEFAULT_TLS_ENDPOINT = "http://tls-cn-beijing.volces.com";
private static final String DEFAULT_REGION = "cn-beijing";
private static final String ACCESS_KEY_ID = "";
private static final String ACCESS_KEY_SECRET = "";
public static void main(String[] args) throws Exception {
final ParameterTool params = ParameterTool.fromArgs(args);
final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.getConfig().setGlobalJobParameters(params);
env.setParallelism(3);
env.enableCheckpointing(5000);
env.setRestartStrategy(RestartStrategies.fixedDelayRestart(3, Time.seconds(10)));
DataStream<String> stream = env.addSource(new RandomStringSource());
// Initialize the client. It is recommended to dynamically obtain volcano engine credentials
// like AccessKey via environment variables to prevent risks associated with hardcoding
// AccessKeys and compromising data security. For detailed instructions, please refer to
// https://www.volcengine.com/docs/6470/1166455.
// When using STS (Security Token Service), both ak (Access Key ID) and sk (Secret Access
// Key) should be temporary keys, and the VOLCENGINE_TOKEN environment variable should be
// set; when not using STS, leave the VOLCENGINE_TOKEN empty.The FlinkLogSink initialization
// comes with default configurations, adjust according to actual needs.
FlinkLogSink<String> flinkLogSink =
new FlinkLogSink.Builder<String>(
params.get(LogConstants.LOG_ENDPOINT, DEFAULT_TLS_ENDPOINT),
params.get(LogConstants.LOG_REGION, DEFAULT_REGION),
params.get(LogConstants.LOG_ACCESS_KEY, ACCESS_KEY_ID),
params.get(LogConstants.LOG_ACCESS_SECRET, ACCESS_KEY_SECRET))
.setSerializationFunction(new SimpleStringFunction(params.get("TOPIC")))
.build();
stream.addSink(flinkLogSink);
env.execute("flink log producer");
}
/** Message produce. */
public static class RandomStringSource implements ParallelSourceFunction<String> {
private volatile boolean running = true;
@Override
public void run(SourceContext<String> ctx) throws Exception {
long seq = 0;
while (running) {
Thread.sleep(10);
ctx.collect((seq++) + "-" + getLongString(30));
}
}
@Override
public void cancel() {
running = false;
}
public static String getLongString(int length) {
String str = "abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789";
Random random = new Random();
StringBuffer sb = new StringBuffer();
for (int i = 0; i < length; i++) {
int number = random.nextInt(62);
sb.append(str.charAt(number));
}
return sb.toString();
}
}
/** SimpleStringFunction. */
public static class SimpleStringFunction implements LogSerializationFunction<String> {
private final String topic;
public SimpleStringFunction(String topic) {
this.topic = topic;
}
@Override
public FlinkLogGroup serialize(String element) {
FlinkLogGroup group = new FlinkLogGroup();
group.setTopic(topic);
LogItem log = new LogItem();
log.setTime(System.currentTimeMillis());
log.addContent("content", element);
group.getLogs().add(log);
return group;
}
}
}
最近更新时间：2026.06.09 11:48:34
这个页面对您有帮助吗？
有用
有用
无用
无用