本文属于机器翻译版本。若本译文内容与英语原文存在差异，则一律以英文原文为准。
# 维持 Managed Service for Apache Flink 应用程序的最佳实践
本节包含有关开发稳定的高性能 Managed Service for Apache Flink应用程序的信息和建议。
**Topics**
+ [尽可能减小 uber JAR 的大小](#minimize-uber-JAR)
+ [容错：检查点和保存点](#how-dev-bp-checkpoint)
+ [连接器版本不受支持](#how-dev-bp-connectors)
+ [性能和并行度](#how-dev-bp-performance)
+ [设置每个运算符的并行度](#how-dev-bp-parallelism)
+ [日志记录](#how-dev-bp-logging)
+ [编码](#how-dev-bp-code)
+ [管理凭证。](#how-dev-bp-managing-credentials)
+ [从分片/分区很少的源中读取](#troubleshooting-few-shards-partitions)
+ [Studio 笔记本刷新间隔](#notebook-refresh-rate)
+ [Studio 笔记本的最佳性能](#notebook-refresh-rate)
+ [水印策略和空闲分片如何影响时间窗口](#notebook-watermarking)
+ [为所有运算符设置 UUID](#best-practices-setting-operator-ids)
+ [添加 ServiceResourceTransformer 到 Maven 阴影插件](#best-practices-service-resource-transformer)
## 尽可能减小 uber JAR 的大小
Java/Scala application must be packaged in an uber (super/fat) JAR，并包括运行时尚未提供的所有其他必需依赖项。但是，uber JAR 的大小会影响应用程序的启动和重新启动时间，并可能导致 JAR 超过 512 MB 的限制。
为优化部署时间，uber JAR **不**应包含以下内容：
+ **运行时提供的任何依赖项**，如以下示例所示。它们应该在 POM 文件中具有 `compileOnly` 范围或在 Gradle 配置中具有 `provided` 范围。
+ **任何仅用于测试的依赖项**，例如 JUnit 或 Mockito。它们应该在 POM 文件中具有 `testImplementation` 范围或在 Gradle 配置中具有 `test` 范围。
+ **您的应用程序未实际使用的任何依赖项**。
+ **您的应用程序所需的任何静态数据或元数据。**静态数据应由应用程序在运行时加载，例如从数据存储或 Amazon S3 加载。
+ 有关上述配置设置的详细信息，请参阅此 [POM 示例文件](https://github.com/aws-samples/amazon-managed-service-for-apache-flink-examples/blob/main/java/GettingStarted/pom.xml)。
**提供的依赖项**
Managed Service for Apache Flink 运行时提供许多依赖项。这些依赖项不应包含在 fat JAR 中，并且必须在 POM 文件中具有 `provided` 范围，或者在 `maven-shade-plugin` 配置中明确排除。fat JAR 中包含的任何依赖项在运行时都会被忽略，但会增加 JAR 的大小，从而在部署期间带来额外开销。
运行时版本 1.18、1.19 和 1.20 中由运行时提供的依赖项：
+ `org.apache.flink:flink-core`
+ `org.apache.flink:flink-java`
+ `org.apache.flink:flink-streaming-java`
+ `org.apache.flink:flink-scala_2.12`
+ `org.apache.flink:flink-table-runtime`
+ `org.apache.flink:flink-table-planner-loader`
+ `org.apache.flink:flink-json`
+ `org.apache.flink:flink-connector-base`
+ `org.apache.flink:flink-connector-files`
+ `org.apache.flink:flink-clients`
+ `org.apache.flink:flink-runtime-web`
+ `org.apache.flink:flink-metrics-code`
+ `org.apache.flink:flink-table-api-java`
+ `org.apache.flink:flink-table-api-bridge-base`
+ `org.apache.flink:flink-table-api-java-bridge`
+ `org.apache.logging.log4j:log4j-slf4j-impl`
+ `org.apache.logging.log4j:log4j-api`
+ `org.apache.logging.log4j:log4j-core`
+ `org.apache.logging.log4j:log4j-1.2-api`
此外，运行时还提供用于在 Managed Service for Apache Flink 中获取应用程序运行时属性的库，`com.amazonaws:aws-kinesisanalytics-runtime:1.2.0`。
运行时提供的所有依赖项都必须使用以下建议，以**免**将其包含在 uber JAR 中：
+ 在 Maven（`pom.xml`）和 SBT（`build.sbt`）中，使用 `provided` 范围。
+ 在 Gradle（`build.gradle`）中，使用 `compileOnly` 配置。
由于 Apache Flink 的父类优先加载，任何意外包含在 uber JAR 中的依赖项都将在运行时被忽略。有关更多信息，请参阅 Apache Flink 文档[parent-first-patterns](https://nightlies.apache.org/flink/flink-docs-master/docs/deployment/config/#classloader-parent-first-patterns-default)中的。
**连接器**
运行时中未包含的大多数 FileSystem 连接器（连接器除外）都必须包含在默认作用域 (`compile`) 的 POM 文件中。
**其他建议**
通常，提供给 Managed Service for Apache Flink 的 Apache Flink uber JAR 应包含运行应用程序所需的最少量代码。包含源类、测试数据集或引导状态的依赖项不应包含在此 JAR 文件中。如果需要在运行时提取静态资源，请将此问题分成诸如 Amazon S3 之类的资源。此方面的示例包括状态引导或推理模型。
花点时间考虑您的深度依赖项树并移除非运行时依赖项。
尽管 Managed Service for Apache Flink 支持 512MB 的 jar 大小，但这应视为规则的例外情况。Apache Flink 目前通过其默认配置支持大约 104MB 的 jar 大小，这应该是所需 jar 的最大目标大小。
## 容错：检查点和保存点
使用检查点和保存点在 Managed Service for Apache Flink 应用程序中实施容错功能。在开发和维护应用程序时，请牢记以下几点：
+ 我们建议您为应用程序保持启用检查点。在计划维护期间以及由于服务问题、应用程序依赖项故障以及其他问题而导致意外故障，检查点可以为应用程序提供容错功能。有关定期维护的更多信息，请参阅[管理 Managed Service for Apache Flink 的维护任务](maintenance.md)。
+ `false`在应用程序开发或故障排除期间SnapshotsEnabled将[ApplicationSnapshotConfiguration::](https://docs.aws.amazon.com/managed-flink/latest/apiv2/API_ApplicationSnapshotConfiguration.html) 设置为。在每次应用程序停止期间，将会创建一个快照；如果应用程序处于不正常状态或性能不佳，则可能会出现问题。在应用程序处于生产状态并保持稳定后，将 `SnapshotsEnabled` 设置为 `true`。
**注意**
我们建议您设置应用程序以每天创建几次快照，以便使用正确的状态数据正确重启。正确的快照频率取决于应用程序的业务逻辑。频繁拍摄快照可让您恢复更新的数据，但会增加成本并需要更多的系统资源。
有关监控应用程序停机时间的信息，请参阅[Managed Service for Apache Flink 中的指标和维度](metrics-dimensions.md)。
有关实施容错功能的更多信息，请参阅[实施容错能力](how-fault.md)。
## 连接器版本不受支持
从 Apache Flink 1.15 或更高版本开始，如果应用程序使用捆绑到应用程序中的不受支持的 Kinesis 连接器版本，则适用于 Apache Flink 的托管服务会自动阻止应用程序启动或更新。 JARs升级到 Managed Service for Apache Flink 1.15 或更高版本时，请确保使用的是最新的 Kinesis 连接器。可以是 1.15.2 或更高版本的任何版本。Managed Service for Apache Flink 不支持所有其他版本，因为它们可能会导致一致性问题或故障，而**使用保存点停止**功能会防止清理停止/更新操作。要了解有关 Amazon Managed Service for Apache Flink 版本中连接器兼容性的更多信息，请参阅[Apache Flink 连接器](https://docs.aws.amazon.com/managed-flink/latest/java/how-flink-connectors.html)。
## 性能和并行度
应用程序可以调整应用程序并行度并避免性能陷阱，从而进行扩展以满足任何吞吐量级别要求。在开发和维护应用程序时，请牢记以下几点：
+ 验证是否充分预置了所有应用程序源和接收器，而不会受到限制。如果源和接收器是其他 AWS 服务，请使用监控这些服务[CloudWatch](https://docs.aws.amazon.com/cloudwatch/?id=docs_gateway)。
+ 对于并行度较高的应用程序，请检查是否将较高的并行度应用于应用程序中的所有操作符。默认情况下，Apache Flink 为应用程序图中的所有操作符应用相同的应用程序并行度。这可能会导致在源或接收器上出现预置问题，或者出现操作符数据处理瓶颈。您可以使用 [setParallelism](https://nightlies.apache.org/flink/flink-docs-release-1.15/dev/parallel.html) 更改代码中的每个操作符的并行度。
+ 了解应用程序中的操作符的并行度设置的含义。如果更改操作符的并行度，您可能无法从操作符并行度与当前设置不兼容时创建的快照中还原应用程序。有关设置运算符并行度的更多信息，请参阅[为运算符显式设置最大并行度](https://nightlies.apache.org/flink/flink-docs-release-1.15/ops/production_ready.html#set-maximum-parallelism-for-operators-explicitly)。
有关实施扩展的更多信息，请参阅[实施应用程序扩展](how-scaling.md)。
## 设置每个运算符的并行度
默认情况下，所有运算符均在应用程序级别设置并行度。您可以使用 DataStream API 覆盖单个运算符的并行度。`.setParallelism(x)`您可以将运算符并行度设置为等于或低于应用程序并行度的任一并行度。
如果可能，将运算符并行度定义为应用程序并行度的函数。这样，运算符的并行度就会随应用程序的并行度而变化。例如，如果您使用自动缩放，则所有运算符都将以相同的比例改变其并行度：
```
int appParallelism = env.getParallelism();
...
...ops.setParalleism(appParallelism/2);
```
在某些情况下，您可能希望将运算符并行度设置为常数。例如，将 Kinesis Stream 源的并行度设置为分片数。在这些情况下，可考虑将运算符并行度作为应用程序配置参数传递，以在不更改代码的情况下对其进行更改，例如对源流进行重新分片。
## 日志记录
您可以使用 CloudWatch 日志监控应用程序的性能和错误情况。为应用程序配置日志记录时，请牢记以下几点：
+ 启用应用程序的 CloudWatch 日志记录，以便可以调试任何运行时问题。
+ 不要为应用程序中处理的每条记录创建一个日志条目。这会在处理期间出现严重瓶颈，并且可能会导致数据处理反向压力。
+ 创建 CloudWatch 警报，以便在应用程序无法正常运行时通知您。有关更多信息，请参阅 [在适用于 Apache Flink 的亚马逊托管服务中使用 CloudWatch 警报](monitoring-metrics-alarms.md)。
有关实施日志记录的更多信息，请参阅[](monitoring-overview.md)。
## 编码
您可以使用建议的编程做法以提高应用程序性能和稳定性。在编写应用程序代码时，请牢记以下几点：
+ 不要在应用程序代码（应用程序的 `main` 方法或用户定义的函数）中使用 `system.exit()`。如果要从代码中关闭应用程序，请引发一个从 `Exception` 或 `RuntimeException` 派生的异常，其中包含有关应用程序出现的错误的消息。
请注意下面有关该服务如何处理此类异常的信息：
+ 如果异常是从应用程序的 `main` 方法中引发的，在应用程序转变为 `RUNNING` 状态时，该服务将其封装在 `ProgramInvocationException` 中，并且任务管理器无法提交任务。
+ 如果异常是从用户定义的函数中引发的，任务管理器使任务失败并重新启动，并将异常详细信息写入到异常日志中。
+ 请考虑为应用程序 JAR 文件及其包含的依赖项填充阴影。如果应用程序和 Apache Flink 运行时系统的程序包名称存在潜在的冲突，则建议填充阴影。如果发生冲突，则应用程序日志可能包含 `java.util.concurrent.ExecutionException` 类型的异常。有关为应用程序 JAR 文件填充阴影的更多信息，请参阅 [Apache Maven Shade 插件](https://maven.apache.org/plugins/maven-shade-plugin/)。
## 管理凭证。
您不应在生产（或任何其他）应用程序中加入任何长期凭证。长期凭证很可能会被签入版本控制系统，很容易丢失。相反，您可以将角色与 Managed Service for Apache Flink 应用程序关联并为该角色授予权限。然后，正在运行的 Flink 应用程序可以从环境中选择具有相应权限的临时证书。如果未与 IAM 原生集成的服务（例如，需要用户名和密码进行身份验证的数据库）需要身份验证，则应考虑将密钥存储在 [AWS Secrets Manager](https://aws.amazon.com/secrets-manager/) 中。
许多 AWS 本机服务都支持身份验证：
+ [Kinesis Data Stream ProcessTaxiStream s — .java](hhttps://github.com/aws-samples/amazon-kinesis-data-analytics-taxi-consumer/blob/master/src/main/java/com/amazonaws/samples/kaja/taxi/consumer/ProcessTaxiStream.java#L90)
+ 亚马逊 MSK — [https://github.com/aws/aws-msk-iam-auth/using-the-amazon-msk](https://github.com/aws/aws-msk-iam-auth/#using-the-amazon-msk-library-for-iam-authentication)\$1-library-for-iam-authentication
+ [亚马逊 Elasticsearch Service — .java AmazonElasticsearchSink](https://github.com/aws-samples/amazon-kinesis-data-analytics-taxi-consumer/blob/master/src/main/java/com/amazonaws/samples/kaja/taxi/consumer/operators/AmazonElasticsearchSink.java)
+ Amazon S3 — 在 Managed Service for Apache Flink 上开箱即用
## 从分片/分区很少的源中读取
从 Apache Kafka 或 Kinesis 数据流读取数据时，流的并行度（Kafka 的分区数和 Kinesis 的分片数）与应用程序的并行度之间可能存在不匹配。在简单的设计中，应用程序的并行度不能超出流的并行度：源运算符的每个子任务只能从 1 个或多个分片/分区读取。这意味着，对于一个只有 2 个分片的流和一个并行度为 8 的应用程序，实际上只有两个子任务从流中消耗，6 个子任务处于空闲状态。这会大大限制应用程序的吞吐量，尤其是在反序列化成本高昂且由源端执行的情况下（这是默认设置）。
为了减轻这种影响，可以扩展流。但这不一定可取或可行。或者，您可以重构源，使其不进行任何序列化，而只是在 `byte[]` 上传递。然后，您可以[重新平衡](https://nightlies.apache.org/flink/flink-docs-stable/docs/dev/datastream/overview/)数据，使其在所有任务中均匀分布，然后在那里反序列化数据。通过这种方式，您可以利用所有子任务进行反序列化，并且这种可能昂贵的操作不再受流数量 shards/partitions 的约束。
## Studio 笔记本刷新间隔
如果更改段落结果刷新间隔，请将其值设置为不低于 1000 毫秒。
## Studio 笔记本的最佳性能
我们使用以下语句进行了测试，`events-per-second` 乘以 `number-of-keys` 低于 25,000,000 时获得了最佳性能。`events-per-second` 低于 150,000。
```
SELECT key, sum(value) FROM key-values GROUP BY key
```
## 水印策略和空闲分片如何影响时间窗口
从 Apache Kafka 和 Kinesis 数据流读取事件时，源可以根据流的属性设置事件时间。对于 Kinesis，事件时间等于事件的大致到达时间。但是，在源为事件设置事件时间不足以让 Flink 应用程序使用事件时间。源还必须生成水印，将有关事件时间的信息从源传播到所有其他运算符。[Flink 文档](https://nightlies.apache.org/flink/flink-docs-release-1.14/docs/concepts/time/)很好地概述了该过程的工作原理。
默认情况下，从 Kinesis 读取的事件的时间戳设置为 Kinesis 确定的近似到达时间。要使事件时间在应用程序中发挥作用，另一个先决条件是水印策略。
```
WatermarkStrategy s = WatermarkStrategy
.forMonotonousTimestamps()
.withIdleness(Duration.ofSeconds(...));
```
然后，使用`assignTimestampsAndWatermarks`方法将水印策略应用于`DataStream`。有一些有用的内置策略：
+ `forMonotonousTimestamps()`只会使用事件时间（大概到达时间），并定期将最大值作为水印发出（针对每个特定的子任务）
+ `forBoundedOutOfOrderness(Duration.ofSeconds(...))`与之前的策略类似，但会使用事件时间 — 生成水印的持续时间。
来自 [Flink 文档](https://nightlies.apache.org/flink/flink-docs-release-1.14/docs/concepts/time/)：
*源函数的每个并行子任务通常会独立生成其水印。这些水印定义了该特定并行源上的事件时间。*
*当水印流经流式处理程序时，它们会推进到达运算符的事件时间。每当运算符推进其事件时间时，它都会为其后继运算符在下游生成一个新的水印。*
*有些运算符消耗多个输入流；例如，一个并集或 keyBy(…) 或分区(…) 函数之后的运算符。此类运算符的当前事件时间是其输入流事件时间的最小值。当其输入流更新其事件时间时，运算符也会更新。*
这意味着，如果源子任务从空闲分片中消耗，则下游运算符不会从该子任务中收到新的水印，因此所有使用时间窗口的下游运算符的处理都会停止。为避免这种情况，客户可以在水印策略中添加`withIdleness`选项。使用该选项，在计算运算符的事件时间时，运算符会将水印从空闲的上游子任务中排除。因此，空闲子任务不再阻碍下游运算符的事件时间的推进。
根据您使用的分片分配器，可能不会为某些工作线程分配任何 Kinesis 分片。在这种情况下，即使所有 Kinesis 分片都持续提供事件数据，这些工作线程也会表现出空闲源行为。您可以通过将 `uniformShardAssigner` 与源运算符结合使用来降低这种风险。这样可以确保所有源子任务都有要处理的分片，前提为工作线程的数量小于或等于活动分片的数量。
但是，如果没有子任务在读取任何事件（即流中没有事件），则带有内置水印策略的空闲选项不会推进事件时间。对于从流中读取有限事件集的测试用例来说，这一点尤其明显。由于读取最后一个事件后，事件时间不会推进，因此最后一个窗口（包含最后一个事件）将不会关闭。
### Summary
+ 如果分片处于空闲状态，则 `withIdleness` 设置将不会生成新的水印。它会将空闲子任务发送的最后一个水印排除在下游运算符的最小水印计算之外。
+ 使用内置水印策略，最后一个打开的窗口将不会关闭（除非将发送推进水印的新事件，但这会创建一个随后保持打开状态的新窗口）。
+ 即使时间由 Kinesis 流设定，但如果一个分片的消耗速度比其他分片快（例如，在应用程序初始化期间，或者在并行消耗所有现有分片时忽略它们的关系），仍可能发生延迟到来的事件。`TRIM_HORIZON` parent/child
+ 水印策略的 `withIdleness` 设置似乎中断空闲分片 `(ConsumerConfigConstants.SHARD_IDLE_INTERVAL_MILLIS` 的 Kinesis 源特定设置。
### 示例
以下应用程序正在从流中读取数据，并根据事件时间创建会话窗口。
```
Properties consumerConfig = new Properties();
consumerConfig.put(AWSConfigConstants.AWS_REGION, "eu-west-1");
consumerConfig.put(ConsumerConfigConstants.STREAM_INITIAL_POSITION, "TRIM_HORIZON");
FlinkKinesisConsumer consumer = new FlinkKinesisConsumer<>("...", new SimpleStringSchema(), consumerConfig);
WatermarkStrategy s = WatermarkStrategy
.forMonotonousTimestamps()
.withIdleness(Duration.ofSeconds(15));
env.addSource(consumer)
.assignTimestampsAndWatermarks(s)
.map(new MapFunction() {
@Override
public Long map(String s) throws Exception {
return Long.parseLong(s);
}
})
.keyBy(l -> 0l)
.window(EventTimeSessionWindows.withGap(Time.seconds(10)))
.process(new ProcessWindowFunction() {
@Override
public void process(Long aLong, ProcessWindowFunction.Context context, Iterableiterable, Collector