使用ApacheFlink创建运营洞察

### 使用 Apache Flink 创建运营洞察 Amazon Kinesis Data Analytics for Apache Flink 使我们能够超越 SQL，使用 Java 或 Scala 作为编程语言，并利用数据流 API 构建分析应用程序。接下来我们将重点介绍 KDA for Flink。 如果您不熟悉 Apache Flink，建议您先阅读 Flink 概述：[https://ci.apache.org/projects/flink/flink-docs-release-1.11/learn-flink/](https://ci.apache.org/projects/flink/flink-docs-release-1.11/learn-flink/)。 #### KDA SQL 与 KDA Flink 的比较 创建 KDA for Flink 应用程序时，与 KDA SQL 有相同之处，也有一些不同，如下表所示： | 对比项 | KDA SQL | KDA Flink | | --- | --- | --- | | 编程语言 | SQL | Java 或 Scala | | 跨账户功能 | 不支持 | 支持多种方式实现 | | 数据源 | - | 可处理非 AWS 数据源，如 Kafka | | 语义支持 | - | 支持 Exactly-once 语义 | | 灵活性 | 相对较低 | 更高，有更多选项 | KDA Flink 应用程序具有更多选项和灵活性，这在选择使用哪个引擎时可能是一个决定性因素。例如，公司通常有多个 AWS 账户，数据流会跨越这些账户。在 AWS 跨账户共享和数据流消费方面，基于 SQL 的 Kinesis Data Analytics 应用程序目前不支持该功能。假设我们有两个 AWS 账户 Account01 和 Account02，实现跨账户功能有以下三种选择： 1. **Java 应用程序**：设置一个 Java 应用程序，从 Account01 的 Kinesis 数据流中读取数据，然后写入 Account02 的 Kinesis 数据流。SQL KDA 应用程序将在 Account02 中运行，并处理该账户中 Kinesis 数据流的数据。示例可参考：[https://docs.aws.amazon.com/kinesisanalytics/latest/java/get-started-exercise.html#get-started-exercise-5](https://docs.aws.amazon.com/kinesisanalytics/latest/java/get-started-exercise.html#get-started-exercise-5)。 2. **AWS Lambda**：使用 AWS Lambda 将数据从 Account01 数据流移动到 Account02 数据流。该解决方案与第一个类似，关键区别在于使用 Lambda 而不是编写 Java KDA 应用程序。解决方案可参考：[https://github.com/awslabs/kinesis-aggregation/blob/master/java/KinesisLambdaForwarder/README.md](https://github.com/awslabs/kinesis-aggregation/blob/master/java/KinesisLambdaForwarder/README.md)。 3. **Flink 应用程序**：不使用 SQL KDA，而是切换到基于 Flink 的应用程序：[https://docs.aws.amazon.com/kinesisanalytics/latest/java/examples-cross.html](https://docs.aws.amazon.com/kinesisanalytics/latest/java/examples-cross.html)。 #### KDA Flink 处理非 AWS 数据源 KDA for Flink 应用程序可以调用 Kafka 或其他 Flink 连接器支持的非 AWS 部署的数据源，这些数据源可以是本地的，也可以来自不同的云提供商。这些资源必须对 KDA 可访问，我们需要正确配置防火墙和安全性。 Kinesis 数据流本身不提供 Exactly-once 语义。有一些解决方法，例如在消费者应用程序中使用 DynamoDB、S3 或 ElasticCache 来检查和避免重复数据。我们可以使用 KDA for Flink 实现 Exactly-once 语义，因为 Flink 本身支持该功能。KDA 会开启 Flink 的 Exactly-once 检查点配置，确保检查点持久化到 S3 等持久存储中。Exactly-once 处理会带来一定开销，无论在 EKS 还是 KDA 上运行 Flink 都是如此。在使用 Exactly-once 交付功能之前，请确保了解其对应用程序的影响。有关 Flink 的 Exactly-once 语义的详细信息，请参考：[https://flink.apache.org/features/2018/03/01/end-to-end-exactly-once-apache-flink.html](https://flink.apache.org/features/2018/03/01/end-to-end-exactly-once-apache-flink.html)。 #### 在 AWS 云中运行 Flink 应用程序的选项 除了在 KDA 上运行 Flink 应用程序外，在 AWS 云中还有其他一些选项，这取决于我们想要的灵活程度。以下是不同方法运行 Flink 应用程序的责任划分： | 运行方式 | 灵活性 | 自动缩放 | 管理难度 | | --- | --- | --- | --- | | 自建 EC2 实例 | 最高 | 需自行实现 | 高 | | Amazon EKS | 较高 | 内置，可通过 CLI 或控制台设置阈值 | 中 | | Amazon EMR | 较高 | 内置，可通过 CLI 或控制台设置阈值 | 中 | | KDA for Flink | 低 | 可配置，自动缩放会覆盖部分并行度设置 | 低 | 运行自己的 EC2 实例并为 Flink 应用程序提供自己的运行时，提供了最大程度的灵活性，但也需要我们付出最多的努力。对于 EC2 实例，自动缩放需要我们自己实现，即输出 Flink 指标，创建阈值和缩放策略。而 Amazon EKS 和 Amazon EMR 为底层资源内置了自动缩放功能，我们可以通过 CLI 或控制台设置阈值。 EMR 最近宣布能够在 EKS 上运行大数据作业，现在我们可以使用 EMR 而无需担心管理底层 Kubernetes 集群。 KDA for Flink 则由 AWS 完全管理，从我们的角度来看是完全无服务器的（无需担心服务器或存储）。由于 AWS 负责管理 Flink 集群，我们无法直接使用 Flink 的 REST API 管理作业。 #### KDA 上的 Flink 应用程序 KDA Flink 引擎架构与 KDA SQL 不同，KDA Flink 实际上作为 Kubernetes 集群运行。下面我们来逐步了解数据流和核心组件： ```mermaid graph LR classDef process fill:#E5F6FF,stroke:#73A6FF,stroke-width:2px; A(API 层):::process --> B(生命周期管理):::process B --> C(创建 IAM 角色):::process B --> D(创建 EKS 集群和工作节点):::process D --> E(加载 Flink 应用程序):::process E --> F(监控和缩放组件):::process ``` 1. **API 层交互**：我们通过 API 层与 KDA for Flink 进行交互。AWS 控制台和 AWS CLI 使用相同的 API，可在 AWS CloudTrail 中查看。API 层接收我们的请求（如 CreateApplication）及参数，并将其作为元数据存储在 DynamoDB 表中。Flink 和 SQL 两个 KDA 引擎的控制平面架构是共享的。 2. **生命周期管理**：接管并协调后续步骤。KDA 获取元数据后，会创建 KDA 应用程序可以承担的 IAM 角色，我们也可以预先创建该角色并将其作为元数据的一部分提供。 3. **启动应用程序**：调用 StartApplication API 时，生命周期管理器将通过创建 EKS 集群和工作节点来实例化 Flink 应用程序的基础设施。Flink 应用程序有自己的 EKS 集群，此配置在 Kubernetes 网络级别提供了 KDA 应用程序的网络隔离。我们可以在开发和生产两种模板中选择，模板定义了诸如是否自动启用快照、每个 KPU 的并行度和日志级别等设置。KDA 将从 S3 加载我们的 Flink 应用程序，应用程序 JAR 文件的最大大小为 512 MB。 在编写本文时，KDA 支持 Flink 1.11.1，要部署 Flink JAR，必须使用 Maven 3.1 进行构建。 KDA 根据模板设置为 Flink 应用程序调整 EKS 集群大小。KDA 使用 Kinesis 处理单元（KPU），每个 KPU 包含 1 个 vCPU、4 GB 内存和 50 GB 存储，基本应用程序从 1 个 KPU 开始。KDA 会为运行应用程序所需的其他组件使用单独的核心，并为我们加载所有相关的 Flink 库。每个 KDA KPU 获得 3 GiB 的 JVM 堆，其余内存分配给 KDA 管理组件。KDA 使用 KMS 对静态数据和传输中的数据进行加密，每个应用程序都会生成 KMS 密钥。 KDA 代表我们管理 Flink 应用程序的整个生命周期，在更新 Flink 应用程序时会自动进行保存点（KDA 术语为快照）。KDA 最多维护 1000 个保存点，以便在需要时恢复。KDA 应用程序通常需要跟踪状态，例如使用窗口时。KDA 会在某处聚合数据并保留窗口状态。激活检查点时，KDA 会跟踪状态。Flink 本身有三种状态存储：MemoryStateBackend（在 Java 堆中维护状态）、FsStateBackend（在文件系统中维护状态）和 RocksDBStateBackend（在 RocksDB 数据库中存储状态）。KDA 的状态后端是 RocksDB 和 S3，由 KDA 完全管理。有关 RocksDBStateBackend 的更多信息，请参考：[https://ci.apache.org/projects/flink/flink-docs-stable/ops/state/state_backends.html#the-rocksdbstatebackend](https://ci.apache.org/projects/flink/flink-docs-stable/ops/state/state_backends.html#the-rocksdbstatebackend