日志服务提供对接 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）。详情请参考[日志项目](https://www.volcengine.com/docs/6470/72005)。 * 已获取当前登录账号的密钥 Access Key。详细信息请参考[创建密钥](https://www.volcengine.com/docs/6291/65568)。 * 使用 IAM 子用户时，需确保已完成相关授权： * 已授权 IAM 用户访问日志服务资源，详情请参考[可授权的操作](https://www.volcengine.com/docs/6470/103038)。 * 已通过 IAM 策略对依赖的 API 进行授权，详细请参考[配置身份认证信息](https://www.volcengine.com/docs/6470/1166455)。 |分类 |依赖的 API |说明 | |---|---|---| |Source |[DescribeCursor](https://www.volcengine.com/docs/6470/112193) |通过 DescribeCursor 接口获取日志下载游标。 | ||[ConsumeLogs](https://www.volcengine.com/docs/6470/112194) |通过 ConsumeLogs 接口获取数据消费日志。 | ||[DescribeShards](https://www.volcengine.com/docs/6470/112197) |调用 DescribeShards 接口获取日志主题的分区列表。 | ||[CreateConsumerGroup](https://www.volcengine.com/docs/6470/1153137) |调用 CreateConsumerGroup 接口创建日志服务消费组（设置消费组时使用）。 | ||[ModifyCheckpoint](https://www.volcengine.com/docs/6470/1153143) |调用 ModifyCheckpoint 接口为指定消费组重置指定分区的消费位点，即更新消费组 Checkpoint。 | ||[DescribeCheckpoint](https://www.volcengine.com/docs/6470/1153142) |调用 DescribeCheckpoint 接口查看指定消费组在指定日志分区上的当前消费位点，即获取消费组 Checkpoint。 | ||[DescribeConsumerGroups](https://www.volcengine.com/docs/6470/1153139) |调用 DescribeConsumerGroups 接口获取一个日志项目下的所有消费组信息。 | |Sink |[PutLogs](https://www.volcengine.com/docs/6470/112191) |调用 PutLogs 接口上传日志到指定的日志主题中。 | ## 添加 Maven 依赖通过 Flink 消费火山引擎日志服务的日志数据时，需要在项目中添加 ve\-tls\-log\-flink\-connector 相关的 Maven 依赖。依赖的详细信息如下： ```Shell com.volcengine ve-tls-log-flink-connector 1.0.0 ``` ## FlinkLogSource（消费者） FlinkLogSource 提供了订阅日志服务中某个日志项目下单个或多个日志主题的能力，支持 Exactly\-Once 语义。使用时无需关心日志主题中分区（Shard）数量的变化，FlinkLogSource 会自动感知。 Flink 中每个子任务负责消费部分 Shard，如果 Shard 发生分裂，子任务消费的 Shard 也会随之改变。Shard 与任务之间的默认分配关系为： ```Shell shardId % numberOfParallelSubtasks == indexOfThisSubtask ```

说明

支持通过设置自定义 [ShardAssigner] 实现个性化的分配方式。

Flink 子任务数量和日志服务日志主题中的分区（Shard）数量是独立的。

如果 Shard 数量多于 Flink 子任务数量，每个 Flink 子任务不重复地消费多个 Shard。

如果 Shard 数量少于 Flink 子任务数量，部分 Flink 子任务会空闲，直到产生新的 Shard。

### 1 配置启动参数以下是一个简单的消费示例，所有配置常量均可在 LogConstants 中查找。 ```Shell 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 deserializer = new LogGroupListDeserializer(); final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); DataStream logStream = env.addSource( new FlinkLogSource<>(deserializer, configProps)); ``` LogGroupListDeserializer 的参考实现如下： ```Shell public class LogGroupListDeserializer implements LogDeserializationFunction { @Override public List deserialize(PullLogsResult record) { List groups = new ArrayList<>(); for (PutLogRequest.LogGroup logGroup : record.getLogGroupList().getLogGroupsList()) { FlinkLogGroup flinkLogGroup = new FlinkLogGroup(); flinkLogGroup.setSource(logGroup.getSource()); int logCount = logGroup.getLogsCount(); List 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 getProducedType() { return GenericTypeInfo.of(FlinkLogGroup.class); } } ``` ### 2 设置消费起始位置通过设置 LogConstants.LOG_CONSUMER_BEGIN_POSITION 参数的值，可以指定消费日志主题中分区的起始位置，具体取值如下： |参数值 |说明及配置示例 | |---|---| |LogConstants.LOG_BEGIN_CURSOR |从 Shard 的头部开始消费，即从最旧的数据开始。

```Shell```
```configProps.put(LogConstants.LOG_CONSUMER_BEGIN_POSITION, LogConstants.LOG_BEGIN_CURSOR);```
| |LogConstants.LOG_END_CURSOR |从 Shard 的尾部开始消费，即从最新的数据开始。

```Shell```
```configProps.put(LogConstants.LOG_CONSUMER_BEGIN_POSITION, LogConstants.LOG_END_CURSOR);```
| |LogConstants.LOG_FROM_CHECKPOINT |以消费组的 Checkpoint 作为消费的起始位置。当从 Checkpoint 获取起始位置时，必须提供消费组名称。

```Shell```
```configProps.put(LogConstants.LOG_CONSUMER_BEGIN_POSITION, LogConstants.LOG_FROM_CHECKPOINT);```

* 支持设置默认的起始位置（默认值：LOG_BEGIN_CURSOR），在消费组不存在或无法获取 Checkpoint 时，将自动切换到默认的起始位置。

* 不支持设置默认位置为 LOG_FROM_CHECKPOINT。

```Shell```
```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 中该时间点开始写入的数据。

```Shell```
```configProps.put(LogConstants.LOG_CONSUMER_BEGIN_POSITION, "1512439000");```
| ### 3 消费进度监控（可选） FlinkLogSource 支持设置消费进度监控，消费进度即获取每个 Shard 实时的消费位置。设置消费组名称后，Connector 会自动创建消费组（如果不存在），并将 Checkpoint 同步到日志服务端，用户可以在日志服务控制台查看消费进度。 ```Shell configProps.put(LogConstants.LOG_CONSUMERGROUP, "your-consumer-group-name"); ``` ### 4 消费组 Checkpoint 提交模式通过配置 LogConstants.LOG_CHECKPOINT_MODE 可以指定消费组 Checkpoint 的提交模式，支持以下三种： * 配置示例： ```Shell 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 恢复。

```Shell```
```final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();```
```env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);```
```env.enableCheckpointing(5000);```
| |DISABLED |Checkpoint 不会被提交到日志服务端。 | |PERIODIC |Checkpoint 被定时提交到日志服务端，与 Flink Checkpointing 完全独立。仅与消费组提交 Checkpoint 到日志服务有关，不影响 Flink 的 State。

支持自定义提交间隔，默认为 10 秒提交一次。

```Shell```
```configProps.put(LogConstants.LOG_COMMIT_INTERVAL_MILLIS, "10000");```
| ### 5 消费多个 Topic FlinkLogSource 支持同时消费同一 Project 下的多个 Topic： ```Shell List topics = Arrays.asList("topic-id-1", "topic-id-2", "topic-id-3"); FlinkLogSource source = new FlinkLogSource<>( "your-project-id", topics, deserializer, configProps); DataStream logStream = env.addSource(source); ``` ### 6 自定义 Shard 分配策略默认的 Shard 分配策略为 shardId % numberOfParallelSubtasks，可以通过实现 ShardAssigner 接口自定义分配策略： ```Shell FlinkLogSource 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 等信息。 ```Shell 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 stream = env.addSource( new FlinkLogSource<>(deserializer, configProps)); DataStream res = stream.flatMap((FlatMapFunction) (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 { @Override public List deserialize(PullLogsResult record) { List groups = new ArrayList<>(); for (PutLogRequest.LogGroup logGroup : record.getLogGroupList().getLogGroupsList()) { FlinkLogGroup flinkLogGroup = new FlinkLogGroup(); flinkLogGroup.setSource(logGroup.getSource()); int logCount = logGroup.getLogsCount(); List 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 getProducedType() { return GenericTypeInfo.of(FlinkLogGroup.class); } } } ``` ## FlinkLogSink（生产者） FlinkLogSink 用于将数据写入火山引擎日志服务。 Sink 支持 Flink At\-Least\-Once 语义，这意味着在作业失败的情况下，写入日志服务中的数据可能会重复，但不会丢失。 ### 1 基本用法 ```Shell final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); env.setParallelism(3); env.enableCheckpointing(5000); DataStream stream = env.addSource(new RandomStringSource()); FlinkLogSink flinkLogSink = new FlinkLogSink.Builder( "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： ```Shell public class SimpleStringFunction implements LogSerializationFunction { 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（不阻塞） | 示例： ```Shell FlinkLogSink flinkLogSink = new FlinkLogSink.Builder( endpoint, region, accessKeyId, accessKeySecret) .setSerializationFunction(new SimpleStringFunction(topicId)) .setTotalSizeInBytes(104857600) .setMaxBatchCount(4096) .setLingerMs(2000) .setRetryCount(3) .build(); ``` ### 4 Sink 完整示例以下代码片段仅为示例，使用时需根据实际环境填入 ENDPOINT、REGION、AK、SK 等信息。 ```Shell 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 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 flinkLogSink = new FlinkLogSink.Builder( 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 { private volatile boolean running = true; @Override public void run(SourceContext 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 { 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; } } } ```