本文介绍如何通过 Flink 将日志服务中的日志实时同步到 AWS 的 Kinesis 中。

应用场景

Amazon Kinesis 可实现轻松收集、处理和分析实时流数据。借助 Amazon Kinesis，可以获取视频、音频、应用程序日志和网站点击流等实时数据，也可以获取用于机器学习、分析和其他应用程序的 IoT 遥测数据。支持即可对收到的数据进行处理和分析并做出响应，无需等到收集完全部数据后才开始进行处理。

日志服务提供 Kafka 协议消费功能，使用 Flink 的 flink-connector-kafka 插件对接日志服务，可以获取日志服务中的数据。使用 AWS 提供的 flink-connector-kinesis 插件，可以将 Flink 中的数据传输到 Kinesis 中。

前提条件

已创建日志项目和日志主题。详细操作步骤请参考创建资源。

已为指定日志主题开启 Kafka 协议消费功能，并获取 Kafka 协议消费主题 ID。详细说明请参考通过 Kafka 协议消费日志。

推荐使用 IAM 用户进行访问鉴权。使用 IAM 用户前，需确认火山引擎主账号已创建 IAM 用户，且已为其授予消费相关的权限。详细说明请参考可授予的权限。

已获取当前登录账号的密钥 Access Key。详细信息请参考创建密钥。

已获取 AWS 相关访问凭证。

环境准备：

Flink：建议 1.17.x 及以后版本。

Maven：建议 3.9.x 及以后版本。

JDK：建议 1.8.x 及以后版本。

使用 SDK 同步数据至 Kinesis

配置依赖

项目需要用到如下依赖包：

配置项目代码

*说明

import com.google.common.util.concurrent.FutureCallback;​
import com.google.common.util.concurrent.Futures;​
import com.google.common.util.concurrent.ListenableFuture;​
import com.volcengine.model.tls.consumer.ConsumerConfig;​
import com.volcengine.model.tls.pb.PutLogRequest;​
import com.volcengine.service.tls.consumer.Consumer;​
import com.volcengine.service.tls.consumer.ConsumerImpl;​
import com.volcengine.service.tls.consumer.LogProcessor;​
import org.jetbrains.annotations.NotNull;​
import software.amazon.awssdk.auth.credentials.AwsCredentials;​
import software.amazon.awssdk.auth.credentials.StaticCredentialsProvider;​
import software.amazon.kinesis.producer.*;​
​
import java.nio.ByteBuffer;​
import java.nio.charset.StandardCharsets;​
import java.text.SimpleDateFormat;​
import java.util.ArrayList;​
import java.util.Date;​
import java.util.List;​
import java.util.concurrent.ExecutorService;​
import java.util.concurrent.Executors;​
​
public class Tls2Kinesis implements LogProcessor, AwsCredentials, FutureCallback<UserRecordResult> {​
​
    private final SimpleDateFormat dateFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss.SSS");​
​
    private KinesisProducer kinesisProducer;​
    private Consumer tlsConsumer;​
    private ExecutorService executor;​
​
    private String awsAK;​
    private String awsSK;​
    private String streamArn;​
​
​
    public static void main(String[] args) throws Exception {​
        // 初始化客户端​
        Tls2Kinesis tls2Kinesis = new Tls2Kinesis();​
        tls2Kinesis.initKinesisProducer();​
        tls2Kinesis.initTlsConsumer();​
        ​
        // 运行10分钟​
        Thread.sleep(600000);​
        ​
        // 停止运行​
        tls2Kinesis.stop();​
    }​
​
    /**​
     * 火山 TLS 数据处理方法​
     *​
     * @param topicID 数据来源 TopicID​
     * @param shardID 数据来源 ShardID​
     * @param logGroupList 数据列表​
     */​
    @Override​
    public void process(String topicID, int shardID, PutLogRequest.LogGroupList logGroupList) {​
        // 创建数据转换 buffer​
        ByteBuffer data = ByteBuffer.allocate(1024);​
        List<PutLogRequest.LogGroup> logGroups = logGroupList.getLogGroupsList();​
        for (PutLogRequest.LogGroup logGroup: logGroups) {​
            List<PutLogRequest.Log> logs = logGroup.getLogsList();​
            for (PutLogRequest.Log log: logs) {​
                // 填充 时间、topicID、shardID、Source、Path​
                data.putChar('{');​
                this.put(data, "__time__", this.dateFormat.format(new Date(log.getTime())));​
                data.putChar(',');​
                this.put(data, "__topic_id__", topicID);​
                data.putChar(',');​
                this.put(data, "__shard_id__", String.valueOf(shardID));​
                data.putChar(',');​
                this.put(data, "__source__", logGroup.getSource());​
                data.putChar(',');​
                this.put(data, "__path__", logGroup.getFileName());​
                ​
                // 填充tag内容​
                for (PutLogRequest.LogTag tag : logGroup.getLogTagsList()) {​
                    data.putChar(',');​
                    this.put(data, tag.getKey(), tag.getValue());​
                }​
​
                // 填充数据内容​
                List<PutLogRequest.LogContent> logContents = log.getContentsList();​
                for (PutLogRequest.LogContent logContent: logContents) {​
                    data.putChar(',');​
                    this.put(data, logContent.getKey(), logContent.getValue());​
                }​
​
                data.putChar('}');​
                data.flip();​
​
                // 将数据写入到 aws kinesis stream 中​
                ListenableFuture<UserRecordResult> f = this.kinesisProducer.addUserRecord(this.streamArn, String.valueOf(shardID), data);​
                Futures.addCallback(f, this, this.executor);​
                data.clear();​
            }​
        }​
    }​
​
    /**​
     * 写入 kinesis 成功​
     *​
     * @param result 写入结果​
     */​
    @Override​
    public void onSuccess(UserRecordResult result) {​
        System.out.printf("Successfully added user record, shard: %s, number: %s%n", result.getShardId(), result.getSequenceNumber());​
    }​
​
    /**​
     * 写入 kinesis 失败​
     *​
     * @param t 失败异常​
     */​
    @Override​
    public void onFailure(@NotNull Throwable t) {​
        // 此处仅简单的执行了错误打印，消息发送失败后不会再次执行发送，生产环境需要自行增加重试等逻辑。​
        if (t instanceof KinesisProducerException) {​
            UserRecord userRecord = ((KinesisProducerException) t).getUserRecord();​
            if (t instanceof UserRecordFailedException) {​
                int attempts = ((UserRecordFailedException) t).getResult().getAttempts().size()-1;​
                Attempt last = ((UserRecordFailedException) t).getResult().getAttempts().get(attempts);​
                if(attempts > 1) {​
                    Attempt previous = ((UserRecordFailedException) t).getResult().getAttempts().get(attempts - 1);​
                    System.out.printf("UserRecord %s failed to put - %s : %s. Previous failure - %s : %s%n",​
                            userRecord, last.getErrorCode(), last.getErrorMessage(), previous.getErrorCode(),​
                            previous.getErrorMessage());​
                } else {​
                    System.out.printf("UserRecord %s failed to put - %s : %s.%n",​
                            userRecord, last.getErrorCode(), last.getErrorMessage());​
                }​
            } else {​
                System.out.printf("UserRecord %s failed to put with unexpected error: %s%n",​
                        userRecord, t);​
            }​
        }​
​
        System.out.println("Failed to add user record: " + t);​
    }​
​
    /**​
     * 获取 aws 访问凭证 AccessKeyID​
     *​
     * @return AccessKeyID​
     */​
    @Override​
    public String accessKeyId() {​
        return this.awsAK;​
    }​
​
    /**​
     * 获取 aws 访问凭证 SecretAccessKey​
     *​
     * @return SecretAccessKey​
     */​
    @Override​
    public String secretAccessKey() {​
        return this.awsSK;​
    }​
​
    /**​
     * put 将 key 和 value 按照 Json 格式写入 data​
     */​
    private void put(ByteBuffer data, String key, String value) {​
        data.putChar('"').put(key.getBytes(StandardCharsets.UTF_8)).putChar('"')​
                .putChar(':')​
                .putChar('"').put(value.getBytes(StandardCharsets.UTF_8)).putChar('"');​
    }​
​
    /**​
     * 初始化 Kinesis Stream 写入客户端​
     */​
    private void initKinesisProducer() {​
        // 初始化客户端，推荐通过环境变量动态获取火山引擎密钥等身份认证信息，以免 AccessKey 硬编码引发数据安全风险。​
        String awsRegion = "xxx"; // region​
        this.awsAK = "xxx"; // accessKeyID​
        this.awsSK = "xxx"; // SecretAccessKey​
        this.streamArn = "xxx"; // 待写入的 stream，此处仅需填写名称即可​
​
        KinesisProducerConfiguration config = new KinesisProducerConfiguration()​
                .setRecordMaxBufferedTime(300)​
                .setMaxConnections(1)​
                .setRequestTimeout(60000)​
                .setRegion(awsRegion)​
                .setCredentialsProvider(StaticCredentialsProvider.create(this));​
​
​
        this.kinesisProducer = new KinesisProducer(config);​
​
        // 用于处理 aws 写入结果的回调线程池​
        this.executor = Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors());​
    }​
​
    /**​
     * 初始化 TLS 消费客户端​
     *​
     * @throws Exception 初始化过程中发生的异常​
     */​
    private void initTlsConsumer() throws Exception {​
        // 初始化客户端，推荐通过环境变量动态获取火山引擎密钥等身份认证信息，以免 AccessKey 硬编码引发数据安全风险。​
        String tlsEndpoint = "xxx"; // TLS 接入点​
        String tlsRegion = "xxx"; // 待访问的 Region​
        String tlsAccessKeyId = "xxx"; // 火山 AccessKey​
        String tlsAccessKeySecret = "xxx"; // 火山SecretKey​
        String projectID = "xxx"; // 请配置您的日志项目ID​
        String topicId = "xxx"; // 请配置您待消费的日志主题ID列表​
        String consumerGroupName = "xxx"; // 请配置您的消费组名称​
        String consumerName = "xxx"; // 请配置消费者名称​
​
        ConsumerConfig config = new ConsumerConfig(tlsEndpoint, tlsRegion, tlsAccessKeyId, tlsAccessKeySecret);​
        config.setProjectID(projectID);​
        config.setTopicIDList(new ArrayList<String>(){{add(topicId);}});​
        config.setConsumerGroupName(consumerGroupName);​
        config.setConsumerName(consumerName);​
​
        // 实例化ConsumerImpl，调用consumer.start()开始持续消费​
        this.tlsConsumer = new ConsumerImpl(config, this);​
        this.tlsConsumer.start();​
    }​
​
    /**​
     * 停止​
     *​
     * @throws InterruptedException 停止过程中的中断异常​
     */​
    private void stop() throws InterruptedException {​
        this.tlsConsumer.stop();​
        this.kinesisProducer.flush();​
        this.kinesisProducer.destroy();​
        this.executor.shutdown();​
    }​
}​

运行项目代码同步数据

运行项目代码。

查询 Kinesis 中的日志记录。

任务运行成功后，即可在 Kinesis 中查询到日志服务同步的日志记录。

基于 Flink 同步数据至 Kinesis

以 Flink 1.17.2、kinesis 4.2.0 版本为例，介绍如何将日志数据同步至 Kinesis。

创建 Flink Jar 包

初始化 Maven 项目。

使用编译工具或直接使用以下命令初始化 Maven 项目。

mvn archetype:generate -DgroupId=com.my.flink -DartifactId=tls2kinesis -DarchetypeArtifactId=maven-archetype-quickstart

配置项目依赖。

进入初始化 Maven 项目时生成的tls2kinesis目录，复制以下内容，替换pom.xml文件中的内容。

<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/maven-v4_0_0.xsd">
<modelVersion>4.0.0</modelVersion>
<groupId>com.my.flink</groupId>
<artifactId>tls2kinesis</artifactId>
<version>1.0-SNAPSHOT</version>
<name>Archetype - tls2kinesis</name>
<url>http://maven.apache.org</url>

<properties>

<flink.version>1.17.2</flink.version>

<kinesis.version>4.2.0-1.17</kinesis.version>
</properties>

<dependencies>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-connector-aws-kinesis-streams</artifactId>
<version>${kinesis.version}</version>
</dependency>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-connector-kafka</artifactId>
<version>${flink.version}</version>
</dependency>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-streaming-java</artifactId>
<version>${flink.version}</version>
<scope>provided</scope>
</dependency>
</dependencies>

<build>
<plugins>
<plugin>
<artifactId>maven-compiler-plugin</artifactId>
<version>2.3.2</version>
<configuration>
<source>1.8</source>
<target>1.8</target>
</configuration>
</plugin>
<plugin>
<artifactId>maven-assembly-plugin</artifactId>
<configuration>
<descriptorRefs>
<descriptorRef>jar-with-dependencies</descriptorRef>
</descriptorRefs>
<archive>
<manifest>
<mainClass>com.my.flink.App</mainClass>
</manifest>
</archive>
</configuration>
<executions>
<execution>
<id>make-assembly</id>
<phase>package</phase>
<goals>
<goal>single</goal>
</goals>
</execution>
</executions>
</plugin>
</plugins>
</build>
</project>

配置项目代码。

复制以下内容，替换src/main/java/com/my/flink/App.java文件中的内容。

package com.my.flink;

import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.connector.aws.config.AWSConfigConstants;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer;
import org.apache.flink.connector.kinesis.sink.KinesisStreamsSink;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import java.util.Properties;

public class App {
public static void main(String[] args) throws Exception {
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setParallelism(1);

DataStream<String> dataStream = env.fromSource(
getKafkaSource(),
WatermarkStrategy.noWatermarks(),
"Tls Kafka Source");

// 将数据输出到kafka
dataStream.sinkTo(getKinesisSink());

env.execute();
}

private static KafkaSource<String> getKafkaSource() {
// 火山 TLS Kafka 消费协议接入点
String tlsKafkaServer = "xxx";
// 火山 TLS Kafka 消费 Topic
String tlsKafkaTopic = "xxx";
// 火山 TLS Kafka 消费组
String tlsKafkaGroup = "xxx";
// 火山 TLS Project ID
String projectID = "xxx";
// 火山 AccessKey
String accessKey = "xxx";
// 火山 SecretKey
String secretKey = "xxx";
// 初始化 Offset
OffsetsInitializer tlsKafkaOffset = OffsetsInitializer.earliest();

return KafkaSource.<String>builder()
.setBootstrapServers(tlsKafkaServer)
.setTopics(tlsKafkaTopic)
.setGroupId(tlsKafkaGroup)
.setStartingOffsets(tlsKafkaOffset)
.setValueOnlyDeserializer(new SimpleStringSchema())
.setProperty("security.protocol", "SASL_PLAINTEXT")
.setProperty("sasl.mechanism", "PLAIN")
.setProperty("sasl.jaas.config",
"org.apache.kafka.common.security.plain.PlainLoginModule required username=\"" + projectID + "\" password=\"" + accessKey + "#" + secretKey + "\";")
.build();
}

public static KinesisStreamsSink<String> getKinesisSink() {
Properties sinkProperties = new Properties();
// aws region
sinkProperties.put(AWSConfigConstants.AWS_REGION, "xxx");

// aws kinesis 数据流访问地址
String streamArn = "xxx";

// aws 访问凭证
sinkProperties.put(AWSConfigConstants.AWS_ACCESS_KEY_ID, "xxx");
sinkProperties.put(AWSConfigConstants.AWS_SECRET_ACCESS_KEY, "xxx");

return KinesisStreamsSink.<String>builder()
.setKinesisClientProperties(sinkProperties)
.setSerializationSchema(new SimpleStringSchema())
.setPartitionKeyGenerator(element -> String.valueOf(element.hashCode()))
.setStreamArn(streamArn) // Required
.setFailOnError(false)
.setMaxBatchSize(500)
.setMaxInFlightRequests(50)
.setMaxBufferedRequests(10_000)
.setMaxBatchSizeInBytes(5 * 1024 * 1024)
.setMaxTimeInBufferMS(5000)
.setMaxRecordSizeInBytes(1024 * 1024)
.build();
}
}

编译打包，获取 Flink jar 包。

执行以下命令后，将生成 target/tls2kinesis-1.0-SNAPSHOT-jar-with-dependencies.jar包。

mvn clean package -Dmaven.test.skip=true

运行 Flink 任务同步数据

上传创建的 Flink jar 包到 Flink 中。

创建并运行 Flink job 任务。

查询 Kinesis 中的日志记录。

任务运行成功后，即可在 Kinesis 中查询到日志服务同步的日志记录。