本文属于机器翻译版本。若本译文内容与英语原文存在差异，则一律以英文原文为准。
# Managed Service for Apache Flink 的故障排除
以下主题可以帮助您解决可能遇到的 Amazon Managed Service for Apache Flink 问题。
选择合适的主题来查看解决方案。
**Topics**
+ [开发故障排除](troubleshooting-development.md)
+ [运行时问题排查](troubleshooting-runtime.md)
# 开发故障排除
本节包含有关诊断和修复 Managed Service for Apache Flink 应用程序的开发问题的信息。
**Topics**
+ [系统回滚最佳实践](troubleshooting-system-rollback.md)
+ [Hudi 配置最佳实践](troubleshooting-hudi.md)
+ [Apache Flink Flame 图表](troubleshooting-update-flamegraphs.md)
+ [EFO 连接器 1.15.2 的凭证提供程序问题](troubleshooting-credential-provider.md)
+ [带有不支持的 Kinesis 连接器的应用程序](troubleshooting-unsupported-kinesis-connectors.md)
+ [编译错误：“无法解析项目的依赖项”](troubleshooting-compile.md)
+ [无效的选项：“kinesisanalyticsv2”](troubleshooting-cli-update.md)
+ [UpdateApplication 操作不会重新加载应用程序代码](troubleshooting-update.md)
+ [S3 StreamingFileSink FileNotFoundExceptions](troubleshooting-s3sink.md)
+ [FlinkKafkaConsumer 使用 savepoint 停止时出现问题](troubleshooting-FlinkKafkaConsumer.md)
+ [Flink 1.15 异步接收器死锁](troubleshooting-async-deadlock.md)
+ [Amazon Kinesis Data Streams 源在重新分片期间处理失序](troubleshooting-kinesis-data-streams-processing-out-of-order.md)
+ [实时向量嵌入蓝图常见问题解答和故障排除](troubleshooting-blueprints.md)
# 系统回滚最佳实践
借助 Amazon Managed Service for Apache Flink 中的自动系统回滚和操作可见性功能，您可以识别和解决应用程序存在的问题。
## 系统回滚
如果您的应用程序更新或扩展操作因客户错误（例如代码错误或权限问题）而失败，则在您已选择启用此功能时，Amazon Managed Service for Apache Flink 将自动尝试回滚到以前的运行版本。有关更多信息，请参阅 [为 Managed Service for Apache Flink 应用程序启用系统回滚](how-system-rollbacks.md)。如果此自动回滚失败，或者您尚未选择加入或选择退出，则您的应用程序将进入 `READY` 状态。若要更新应用程序，请完成以下步骤： 查看 Amazon Managed Service for Apache Flink 控制台或使用 `DescribeApplicationOperation` API 查看错误描述，了解操作失败的原因。 要查看完整的错误堆栈，请使用 [Cloudwatch 日志](https://docs.aws.amazon.com/managed-flink/latest/java/logging.html)。 常见问题包括权限不足、代码更改不兼容或基础设施配置错误。解决潜在问题。 使用 `UpdateApplicaton` API 重新部署您的新应用程序版本。
## 手动回滚
如果应用程序长时间处于停滞状态，或者应用程序已成功过渡到 `Running` 状态，但您看到下游问题（例如在已成功更新的 Flink 应用程序中处理错误），则可以使用 `RollbackApplication` API 手动回滚应用程序。
1. 调用 `RollbackApplication` - 这将恢复到之前的运行版本并还原之前的状态。
1. 使用 `DescribeApplicationOperation` API 监控回滚操作。
1. 如果回滚失败，请使用之前的系统回滚步骤。
## 操作可见性
`ListApplicationOperations` API 显示您的应用程序上所有客户和系统操作的历史记录。
1. 从列表中获取失败操作的 *operationId*。
1. 调用 `DescribeApplicationOperation` 并查看状态和 *statusDescription*。
1. 如果操作失败，描述将指出需要调查的潜在错误。
**常见的错误代码错误：**使用回滚功能恢复到上一个工作版本。解决错误并重试更新。
**权限问题：**使用 `DescribeApplicationOperation` 查看所需的权限。更新应用程序权限并重试。
**适用于 Apache Flink 的亚马逊托管服务服务问题：**查看 AWS Health Dashboard 或提交支持案例。
# Hudi 配置最佳实践
要在 Managed Service for Apache Flink 上运行 Hudi 连接器，我们建议进行以下配置更改。
禁用 `hoodie.embed.timeline.server`
Flink 上的 Hudi 连接器在 Flink 作业管理器（JM）上设置了一个嵌入式时间轴（TM）服务器，用于缓存元数据，从而在作业并行度较高时提高性能。我们建议您在 Managed Service for Apache Flink 上禁用此嵌入式服务器，因为我们禁用 JM 和 TM 之间的非 Flink 通信。
如果启用此服务器，Hudi 写入操作将首先尝试连接到 JM 上的嵌入式服务器，然后回退到从 Amazon S3 读取元数据。这意味着 Hudi 会导致连接超时，从而延迟 Hudi 的写入并对 Managed Service for Apache Flink 造成性能影响。
# Apache Flink Flame 图表
默认情况下，支持火焰图的 Managed Service for Apache Flink版本中的应用程序处于启用状态。如Fl [ink文档中所述，如果您保持图形处于打开状态，Fl](https://nightlies.apache.org/flink/flink-docs-release-1.15//docs/ops/debugging/flame_graphs/) ame Graphs可能会影响应用程序的性能。
如果您想为应用程序禁用 Flame Graphs，请创建一个案例，请求为您的应用程序 ARN 禁用该图表。有关更多信息，请参阅 [AWS 支持中心](https://console.aws.amazon.com/support/home#/)。
# EFO 连接器 1.15.2 的凭证提供程序问题
1.15.2 之前的 Kinesis Data Streams EFO 连接器版本存在一个[已知问题](https://issues.apache.org/jira/browse/FLINK-29205)，其中`FlinkKinesisConsumer`不符合配置。`Credential Provider`由于该问题，有效的配置被忽略，这会导致使用`AUTO`凭据提供程序。这可能会导致使用 EFO 连接器跨账户访问 Kinesis 时出现问题。
要解决此错误，请使用 EFO 连接器版本 1.15.3 或更高版本。
# 带有不支持的 Kinesis 连接器的应用程序
如果应用程序使用绑定[到应用程序或存档 (ZIP) 的不支持的 Kinesis Connector 版本（1.15.2 之前的版本），则适用于 Apache Flink 1.15 或更高版本的 Apache Flink 托管服务将自动拒绝应用程序的启动或更新](https://docs.aws.amazon.com/managed-flink/latest/java/flink-1-15-2.html)。 JARs
## 拒绝错误
通过提交创建/更新应用程序调用时，您将看到以下错误：
```
An error occurred (InvalidArgumentException) when calling the CreateApplication operation: An unsupported Kinesis connector version has been detected in the application. Please update flink-connector-kinesis to any version equal to or newer than 1.15.2.
For more information refer to connector fix: https://issues.apache.org/jira/browse/FLINK-23528
```
## 补救步骤
+ 更新应用程序对的依赖关系`flink-connector-kinesis`。如果您使用 Maven 作为项目的构建工具，请按照以下步骤操作[更新 Maven 依赖关系](#troubleshooting-unsupported-kinesis-connectors-update-maven-dependency)。如果您使用的是 Gradle，请按照[更新 Gradle 依赖关系](#troubleshooting-unsupported-kinesis-connectors-update-gradle-dependency)。
+ 重新打包应用程序。
+ 上传到 Amazon S3 存储桶
+ 重新提交创建/更新申请请求，修改后的应用程序刚刚上传到 Amazon S3 存储桶。
+ 如果您继续看到相同的错误消息，请重新检查您的应用程序依赖关系。如果问题仍然存在，请创建支持请求。
### 更新 Maven 依赖关系
1. 打开项目的`pom.xml`。
1. 查找项目的依赖关系。它们看起来像：
```
...
...
org.apache.flink
flink-connector-kinesis
...
...
```
1. 更新`flink-connector-kinesis`到等于或高于 1.15.2 的版本。例如：
```
...
...
org.apache.flink
flink-connector-kinesis
1.15.2
...
...
```
### 更新 Gradle 依赖关系
1. 打开项目`build.gradle`（或`build.gradle.kts`针对 Kotlin 应用程序）。
1. 查找项目的依赖关系。它们看起来像：
```
...
dependencies {
...
implementation("org.apache.flink:flink-connector-kinesis")
...
}
...
```
1. 更新`flink-connector-kinesis`到等于或高于 1.15.2 的版本。例如：
```
...
dependencies {
...
implementation("org.apache.flink:flink-connector-kinesis:1.15.2")
...
}
...
```
# 编译错误：“无法解析项目的依赖项”
要编译 Managed Service for Apache Flink示例应用程序，您必须先下载并编译 Apache Flink Kinesis 连接器，然后将其添加到本地 Maven 存储库中。如果尚未将连接器添加到存储库中，则会显示类似下面的编译错误：
```
Could not resolve dependencies for project your project name: Failure to find org.apache.flink:flink-connector-kinesis_2.11:jar:1.8.2 in https://repo.maven.apache.org/maven2 was cached in the local repository, resolution will not be reattempted until the update interval of central has elapsed or updates are forced
```
要解决此错误，必须下载连接器的 Apache Flink 源代码（版本为 1.8.2 [https://flink.apache.org/downloads.html](https://flink.apache.org/downloads.html)）。有关如何下载、编译和安装 Apache Flink 源代码的说明，请参阅[将 Apache Flink Kinesis Streams 连接器与之前的 Apache Flink 版本一起使用](earlier.md#how-creating-apps-building-kinesis)。
# 无效的选项：“kinesisanalyticsv2”
要使用适用于 Managed Service for Apache Flink API 的 v2，你需要最新版本的 AWS Command Line Interface (AWS CLI)。
有关升级的信息 AWS CLI，请参阅[《*AWS Command Line Interface 用户指南*》 AWS Command Line Interface中的安装](https://docs.aws.amazon.com/cli/latest/userguide/installing.html)。
# UpdateApplication 操作不会重新加载应用程序代码
如果未指定 S3 对象版本，则该[UpdateApplication](https://docs.aws.amazon.com/managed-flink/latest/apiv2/API_UpdateApplication.html)操作不会重新加载具有相同文件名的应用程序代码。要使用相同的文件名重新加载应用程序代码，请在 S3 存储桶上启用版本控制，并使用 `ObjectVersionUpdate` 参数指定新的对象版本。有关在 S3 存储桶中启用对象版本控制的更多信息，请参阅[启用或禁用版本控制](https://docs.aws.amazon.com/AmazonS3/latest/user-guide/enable-versioning.html)。
# S3 StreamingFileSink FileNotFoundExceptions
如果缺少由其保存点引用的正在处理的部件文件，则从快照启动`FileNotFoundException`时，Managed Service for Apache Flink 应用程序可能会遇到正在进行的部分文件。出现这种故障模式时，Managed Service for Apache Flink 应用程序的操作员状态通常是不可恢复的，必须在不使用快照的情况下重新启动。`SKIP_RESTORE_FROM_SNAPSHOT`参见以下示例堆栈跟踪：
```
java.io.FileNotFoundException: No such file or directory: s3://amzn-s3-demo-bucket/pathj/INSERT/2023/4/19/7/_part-2-1234_tmp_12345678-1234-1234-1234-123456789012
at org.apache.hadoop.fs.s3a.S3AFileSystem.s3GetFileStatus(S3AFileSystem.java:2231)
at org.apache.hadoop.fs.s3a.S3AFileSystem.innerGetFileStatus(S3AFileSystem.java:2149)
at org.apache.hadoop.fs.s3a.S3AFileSystem.getFileStatus(S3AFileSystem.java:2088)
at org.apache.hadoop.fs.s3a.S3AFileSystem.open(S3AFileSystem.java:699)
at org.apache.hadoop.fs.FileSystem.open(FileSystem.java:950)
at org.apache.flink.fs.s3hadoop.HadoopS3AccessHelper.getObject(HadoopS3AccessHelper.java:98)
at org.apache.flink.fs.s3.common.writer.S3RecoverableMultipartUploadFactory.recoverInProgressPart(S3RecoverableMultipartUploadFactory.java:97)
...
```
Flink `StreamingFileSink` 将记录写入[文件系统](https://nightlies.apache.org/flink/flink-docs-release-1.15/docs/deployment/filesystems/overview/)支持的文件系统。鉴于传入的流可以不受限制，因此将数据组织成大小有限的部分文件，并在写入数据时添加新文件。零件生命周期和展期政策决定了零件文件的时间、大小和命名。
在检查点和保存指向（快照）期间，所有待处理文件都将被重命名并提交。但是，正在处理的部分文件不会提交，而是会重命名，其引用保留在检查点或保存点元数据中，以便在恢复任务时使用。这些正在处理的零件文件最终将变为 “待处理”，由随后的检查点或保存点重命名并提交。
以下是缺少正在处理的零件文件的根本原因和缓解措施：
+ 用于启动适用于 Apache Flink 的托管服务应用程序的陈旧快照 — 只有在应用程序停止或更新时拍摄的最新系统快照才能用于通过 Amazon S3 启动适用于 Amazon Flink 的托管服务应用程序。 StreamingFileSink为避免此类故障，请使用最新的系统快照。
+ 例如，当您在停止或更新期间选择使用创建的快照`CreateSnapshot`而不是系统触发的快照时，就会发生这种情况。旧快照的保存点保留了对正在进行的部分文件的 out-of-date引用，该文件已被后续检查点或保存点重命名并提交。
+ 当从非最新 Stop/Update 事件中选择系统触发的快照时，也会发生这种情况。例如，已禁用系统快照但已`RESTORE_FROM_LATEST_SNAPSHOT`配置的应用程序。通常，使用 Amazon S3 的 Apache Flink 应用程序的托管服务 StreamingFileSink 应始终启用和`RESTORE_FROM_LATEST_SNAPSHOT`配置系统快照。
+ 已移除正在处理的部分文件 — 由于正在处理的部分文件位于 S3 存储桶中，因此其他有权访问该存储桶的组件或参与者可以将其删除。
+ 当您停止应用程序的时间过长，并且您的应用程序的 savepoint 引用的正在处理的部分文件已被 [S3 存储桶 MultiPartUpload](https://docs.aws.amazon.com/AmazonS3/latest/userguide/mpu-abort-incomplete-mpu-lifecycle-config.html)生命周期策略删除时，就会发生这种情况。为避免此类故障，请确保您的 S3 Bucket MPU 生命周期策略涵盖的期限足够长，足以满足您的用例。
+ 当正在处理的零件文件被手动删除或被系统的另一个组件删除时，也可能发生这种情况。为避免此类故障，请确保正在处理的零件文件不会被其他参与者或组件删除。
+ 在 savepoint 之后触发自动检查点的争用条件 — 这会影响 Managed Service for Apache Flink 1.13 及以下版本。Managed Service for Apache Flink 1.15 版本中已修复此问题。将您的应用程序迁移到最新版本的 Managed Service for Apache Flink 以防止问题再次发生。我们还建议从迁移 StreamingFileSink 到[FileSink](https://nightlies.apache.org/flink/flink-docs-release-1.19/docs/connectors/datastream/filesystem/#file-sink)。
+ 当应用程序停止或更新时，适用于 Managed Service for Apache Flink 会触发保存点并分两步停止应用程序。如果在这两个步骤之间触发了自动检查点，则该保存点将无法使用，因为其正在处理的部分文件将被重命名并可能被提交。
# FlinkKafkaConsumer 使用 savepoint 停止时出现问题
如果您启用了系统快照，则在使用旧版 FlinkKafkaConsumer 时，您的应用程序可能会陷入更新、停止或缩放的困境。没有针对此[问题的](https://issues.apache.org/jira/browse/FLINK-28758)已发布修复程序，因此我们建议您升级到新版本[KafkaSource](https://nightlies.apache.org/flink/flink-docs-master/docs/connectors/datastream/kafka/#kafka-source)以缓解此问题。
如果您在`FlinkKafkaConsumer`启用快照的情况下使用的，则当 Flink 任务在处理 savepoint API 请求时停止时，`FlinkKafkaConsumer`可能会失败，并报告运行时系统错误。`ClosedException`在这种情况下，Flink 应用程序会卡住，表现为失败的检查点。
# Flink 1.15 异步接收器死锁
Apache Flink 实现 AWS AsyncSink 接口的连接器存在一个[已知问题](https://issues.apache.org/jira/browse/FLINK-32230)。这会影响使用带有以下连接器的 Flink 1.15 的应用程序：
+ 对于 Java 应用程序：
+ KinesisStreamsSink – `org.apache.flink:flink-connector-kinesis`
+ KinesisStreamsSink – `org.apache.flink:flink-connector-aws-kinesis-streams`
+ KinesisFirehoseSink – `org.apache.flink:flink-connector-aws-kinesis-firehose`
+ DynamoDbSink – `org.apache.flink:flink-connector-dynamodb`
+ Flink SQL/TableAPI/Python 应用程序：
+ kinesis – `org.apache.flink:flink-sql-connector-kinesis`
+ kinesis – `org.apache.flink:flink-sql-connector-aws-kinesis-streams`
+ firehose – `org.apache.flink:flink-sql-connector-aws-kinesis-firehose`
+ dynamodb – `org.apache.flink:flink-sql-connector-dynamodb`
受影响的应用程序将出现以下症状：
+ Flink 任务处于`RUNNING`状态，但未处理数据；
+ 没有任务重启；
+ 检查点正在超时。
该问题是由 AWS SDK中的一个[错误](https://github.com/aws/aws-sdk-java-v2/issues/4354)引起的，该错误导致它在使用异步HTTP客户端时不会向调用者显示某些错误。这会导致接收器无限期地等待 “飞行中请求” 在检查点刷新操作期间完成。
从版本 **2.20.** 144 开始，此问题已在 AWS SDK 中得到修复。
以下是有关如何更新受影响的连接器以在应用程序中使用新版本的 AWS SDK 的说明：
**Topics**
+ [更新 Java 应用程序](troubleshooting-async-deadlock-update-java-apps.md)
+ [更新 Python 应用程序](troubleshooting-async-deadlock-update-python-apps.md)
# 更新 Java 应用程序
按照以下步骤更新 Java 应用程序：
## flink-connector-kinesis
如果应用程序使用 `flink-connector-kinesis`。
Kinesis 连接器使用阴影将一些依赖项（包括 AWS SDK）打包到连接器 jar 中。要更新 S AWS DK 版本，请按以下步骤替换这些阴影类：
------
#### [ Maven ]
1. 将 Kinesis 连接器和所需的 AWS SDK 模块添加为项目依赖项。
1. 配置`maven-shade-plugin`：
1. 在复制 Kinesis 连接器 jar 的内容时，添加过滤器以排除阴影的 AWS SDK 类。
1. 按照 Kinesis 连接器的预期，添加重新定位规则，将更新的 AWS SDK 类移到包中。
**pom.xml**
```
...
...
org.apache.flink
flink-connector-kinesis
1.15.4
software.amazon.awssdk
kinesis
2.20.144
software.amazon.awssdk
netty-nio-client
2.20.144
software.amazon.awssdk
sts
2.20.144
...
...
...
...
org.apache.maven.plugins
maven-shade-plugin
3.1.1
package
shade
...
...
org.apache.flink:flink-connector-kinesis
org/apache/flink/kinesis/shaded/software/amazon/awssdk/**
org/apache/flink/kinesis/shaded/org/reactivestreams/**
org/apache/flink/kinesis/shaded/io/netty/**
org/apache/flink/kinesis/shaded/com/typesafe/netty/**
...
...
software.amazon.awssdk
org.apache.flink.kinesis.shaded.software.amazon.awssdk
org.reactivestreams
org.apache.flink.kinesis.shaded.org.reactivestreams
io.netty
org.apache.flink.kinesis.shaded.io.netty
com.typesafe.netty
org.apache.flink.kinesis.shaded.com.typesafe.netty
...
...
...
...
```
------
#### [ Gradle ]
1. 将 Kinesis 连接器和所需的 AWS SDK 模块添加为项目依赖项。
1. 调整 ShadowJar 配置：
1. 复制 Kinesis 连接器 jar 的内容时，请排除阴影的 AWS SDK 类。
1. 将更新后的 AWS SDK 类重新定位到 Kinesis 连接器预期的软件包中。
**build.gradle**
```
...
dependencies {
...
flinkShadowJar("org.apache.flink:flink-connector-kinesis:1.15.4")
flinkShadowJar("software.amazon.awssdk:kinesis:2.20.144")
flinkShadowJar("software.amazon.awssdk:sts:2.20.144")
flinkShadowJar("software.amazon.awssdk:netty-nio-client:2.20.144")
...
}
...
shadowJar {
configurations = [project.configurations.flinkShadowJar]
exclude("software/amazon/kinesis/shaded/software/amazon/awssdk/**/*")
exclude("org/apache/flink/kinesis/shaded/org/reactivestreams/**/*.class")
exclude("org/apache/flink/kinesis/shaded/io/netty/**/*.class")
exclude("org/apache/flink/kinesis/shaded/com/typesafe/netty/**/*.class")
relocate("software.amazon.awssdk", "org.apache.flink.kinesis.shaded.software.amazon.awssdk")
relocate("org.reactivestreams", "org.apache.flink.kinesis.shaded.org.reactivestreams")
relocate("io.netty", "org.apache.flink.kinesis.shaded.io.netty")
relocate("com.typesafe.netty", "org.apache.flink.kinesis.shaded.com.typesafe.netty")
}
...
```
------
## 其他受影响的连接器
如果应用程序使用其他受影响的连接器：
要更新 AWS SDK 版本，应在项目构建配置中强制执行 SDK 版本。
------
#### [ Maven ]
将 AWS SDK 物料清单 (BOM) 添加到`pom.xml`文件的依赖项管理部分，以强制执行项目的 SDK 版本。
**pom.xml**
```
...
...
software.amazon.awssdk
bom
2.20.144
import
pom
...
...
```
------
#### [ Gradle ]
添加对 AWS SDK 物料清单 (BOM) 的平台依赖，以强制执行项目的 SDK 版本。这需要 Gradle 5.0 或更高版本：
**build.gradle**
```
...
dependencies {
...
flinkShadowJar(platform("software.amazon.awssdk:bom:2.20.144"))
...
}
...
```
------
# 更新 Python 应用程序
Python 应用程序可以通过两种不同的方式使用连接器：将连接器和其他 Java 依赖项打包为单个 uber-jar 的一部分，或者直接使用连接器 jar。要修复受 Async Sink 死锁影响的应用程序，请执行以下操作：
+ 如果应用程序使用 uber jar，请按照中的说明进行操作[更新 Java 应用程序](troubleshooting-async-deadlock-update-java-apps.md)。
+ 要从源代码重建连接器 jar，请使用以下步骤：
**从源头构建连接器：**
先决条件，类似于 Flink [编译要求](https://nightlies.apache.org/flink/flink-docs-release-1.15/docs/flinkdev/building/#build-flink)：
+ Java 11
+ Maven 3.2.5
## flink-sql-connector-kinesis
1. 下载 Flink 1.15.4 源代码：
```
wget https://archive.apache.org/dist/flink/flink-1.15.4/flink-1.15.4-src.tgz
```
1. 解压缩源代码：
```
tar -xvf flink-1.15.4-src.tgz
```
1. 导航到 kinesis 连接器目录
```
cd flink-1.15.4/flink-connectors/flink-connector-kinesis/
```
1. 编译并安装连接器 jar，指定所需的 AWS SDK 版本。要加快构建速度`-DskipTests`，请使用跳过测试执行和`-Dfast`跳过其他源代码检查：
```
mvn clean install -DskipTests -Dfast -Daws.sdkv2.version=2.20.144
```
1. 导航到 kinesis 连接器目录
```
cd ../flink-sql-connector-kinesis
```
1. 编译并安装 sql 连接器 jar：
```
mvn clean install -DskipTests -Dfast
```
1. 生成的罐子将在以下网址发售：
```
target/flink-sql-connector-kinesis-1.15.4.jar
```
## flink-sql-connector-aws-kinesis-streams
1. 下载 Flink 1.15.4 源代码：
```
wget https://archive.apache.org/dist/flink/flink-1.15.4/flink-1.15.4-src.tgz
```
1. 解压缩源代码：
```
tar -xvf flink-1.15.4-src.tgz
```
1. 导航到 kinesis 连接器目录
```
cd flink-1.15.4/flink-connectors/flink-connector-aws-kinesis-streams/
```
1. 编译并安装连接器 jar，指定所需的 AWS SDK 版本。要加快构建速度`-DskipTests`，请使用跳过测试执行和`-Dfast`跳过其他源代码检查：
```
mvn clean install -DskipTests -Dfast -Daws.sdk.version=2.20.144
```
1. 导航到 kinesis 连接器目录
```
cd ../flink-sql-connector-aws-kinesis-streams
```
1. 编译并安装 sql 连接器 jar：
```
mvn clean install -DskipTests -Dfast
```
1. 生成的罐子将在以下网址发售：
```
target/flink-sql-connector-aws-kinesis-streams-1.15.4.jar
```
## flink-sql-connector-aws-kinesis-firehose
1. 下载 Flink 1.15.4 源代码：
```
wget https://archive.apache.org/dist/flink/flink-1.15.4/flink-1.15.4-src.tgz
```
1. 解压缩源代码：
```
tar -xvf flink-1.15.4-src.tgz
```
1. 导航到连接器目录
```
cd flink-1.15.4/flink-connectors/flink-connector-aws-kinesis-firehose/
```
1. 编译并安装连接器 jar，指定所需的 AWS SDK 版本。要加快构建速度`-DskipTests`，请使用跳过测试执行和`-Dfast`跳过其他源代码检查：
```
mvn clean install -DskipTests -Dfast -Daws.sdk.version=2.20.144
```
1. 导航到 sql 连接器目录
```
cd ../flink-sql-connector-aws-kinesis-firehose
```
1. 编译并安装 sql 连接器 jar：
```
mvn clean install -DskipTests -Dfast
```
1. 生成的罐子将在以下网址发售：
```
target/flink-sql-connector-aws-kinesis-firehose-1.15.4.jar
```
## flink-sql-connector-dynamodb
1. 下载 Flink 1.15.4 源代码：
```
wget https://archive.apache.org/dist/flink/flink-connector-aws-3.0.0/flink-connector-aws-3.0.0-src.tgz
```
1. 解压缩源代码：
```
tar -xvf flink-connector-aws-3.0.0-src.tgz
```
1. 导航到连接器目录
```
cd flink-connector-aws-3.0.0
```
1. 编译并安装连接器 jar，指定所需的 AWS SDK 版本。要加快构建速度`-DskipTests`，请使用跳过测试执行和`-Dfast`跳过其他源代码检查：
```
mvn clean install -DskipTests -Dfast -Dflink.version=1.15.4 -Daws.sdk.version=2.20.144
```
1. 生成的罐子将在以下网址发售：
```
flink-sql-connector-dynamodb/target/flink-sql-connector-dynamodb-3.0.0.jar
```
# Amazon Kinesis Data Streams 源在重新分片期间处理失序
当前的 FlinkKinesisConsumer 实现并未在 Kinesis 分片之间提供强有力的排序保证。这可能会导致在重新分片 Kinesis Stream 期间进行 out-of-order处理，对于遇到处理延迟的 Flink 应用程序尤其如此。在某些情况下，例如基于事件时间的窗口运算符，事件可能会因为由此产生的延迟而被丢弃。
![\[Diagram showing shards and shard consumers with time progression and trim horizon.\]](http://docs.aws.amazon.com/zh_cn/managed-flink/latest/java/images/flink-ts.png)
这是开源 Flink 中的一个[已知问题](https://issues.apache.org/jira/browse/FLINK-6349)。在连接器修复可用之前，请确保您的 Flink 应用程序在重新分区期间不会落后于 Kinesis Data Streams。通过确保 Flink 应用程序能够容忍处理延迟，您可以最大限度地减少 out-of-order处理的影响和数据丢失的风险。
# 实时向量嵌入蓝图常见问题解答和故障排除
查看以下常见问题解答和故障排除部分，对实时向量嵌入蓝图问题进行故障排除。有关实时向量嵌入蓝图的更多信息，请参阅[实时向量嵌入蓝图](https://docs.aws.amazon.com/msk/latest/developerguide/ai-vector-embedding-integration-learn-more.html)。
[有关 Apache Flink 应用程序的一般托管服务疑难解答，请参阅 https://docs.aws.amazon.com/managed-flink/ latest/java/troubleshooting-runtime.html。](https://docs.aws.amazon.com/managed-flink/latest/java/troubleshooting-runtime.html)
**Topics**
+ [实时向量嵌入蓝图 - 常见问题解答](troubleshooting-blueprints-FAQ.md)
+ [实时向量嵌入蓝图 - 故障排除](troubleshooting-blueprints-TS.md)
# 实时向量嵌入蓝图 - 常见问题解答
查看以下有关实时向量嵌入蓝图的常见问题解答。有关实时向量嵌入蓝图的更多信息，请参阅[实时向量嵌入蓝图](https://docs.aws.amazon.com/msk/latest/developerguide/ai-vector-embedding-integration-learn-more.html)。
**Topics**
+ [这个蓝图创造了哪些 AWS 资源？](#troubleshooting-blueprints-1)
+ [AWS CloudFormation 堆栈部署完成后我的操作是什么？](#troubleshooting-blueprints-2)
+ [源 Amazon MSK 主题中的数据结构应该是什么？](#troubleshooting-blueprints-3)
+ [我能否指定要嵌入的部分消息？](#troubleshooting-blueprints-4)
+ [我是否可以从多个 Amazon MSK 主题中读取数据？](#troubleshooting-blueprints-5)
+ [我能否使用正则表达式来配置 Amazon MSK 主题名称？](#troubleshooting-blueprints-6)
+ [可以从 Amazon MSK 主题中读取的消息的最大大小是多少？](#troubleshooting-blueprints-7)
+ [支持 OpenSearch 哪种类型？](#troubleshooting-blueprints-8)
+ [为什么我需要使用矢量搜索集合、向量索引，并在我的 OpenSearch 无服务器集合中添加向量字段？](#troubleshooting-blueprints-9)
+ [我应该将哪些方面设置为向量字段的维度？](#troubleshooting-blueprints-10)
+ [配置的 OpenSearch 索引中的输出是什么样子？](#troubleshooting-blueprints-11)
+ [我能否指定要添加到存储在 OpenSearch 索引中的文档的元数据字段？](#troubleshooting-blueprints-12)
+ [我应该指望 OpenSearch 索引中有重复的条目吗？](#troubleshooting-blueprints-13)
+ [我可以向多个 OpenSearch 索引发送数据吗？](#troubleshooting-blueprints-14)
+ [我能否在单个 AWS 账户中部署多个实时向量嵌入应用程序？](#troubleshooting-blueprints-15)
+ [多个实时向量嵌入应用程序能否使用相同的数据来源或接收器？](#troubleshooting-blueprints-16)
+ [该应用程序是否支持跨账户连接？](#troubleshooting-blueprints-17)
+ [该应用程序是否支持跨区域连接？](#troubleshooting-blueprints-18)
+ [我的 Amazon MSK 集群和 OpenSearch 集合能否位于不同的 VPCs 或子网中？](#troubleshooting-blueprints-19)
+ [该应用程序支持哪些嵌入模型？](#troubleshooting-blueprints-20)
+ [我能否根据自己的工作负载微调应用程序的性能？](#troubleshooting-blueprints-21)
+ [支持哪些 Amazon MSK 身份验证类型？](#troubleshooting-blueprints-22)
+ [什么是 `sink.os.bulkFlushIntervalMillis`，我该如何设置？](#troubleshooting-blueprints-23)
+ [当我部署 Managed Service for Apache Flink 应用程序时，它将从 Amazon MSK 主题的哪一点开始读取消息？](#troubleshooting-blueprints-24)
+ [如何使用 `source.msk.starting.offset`？](#troubleshooting-blueprints-25)
+ [支持哪些分块策略？](#troubleshooting-blueprints-26)
+ [如何读取向量数据存储中的记录？](#troubleshooting-blueprints-27)
+ [可在何处找到源代码的新更新？](#troubleshooting-blueprints-28)
+ [我能否更改 AWS CloudFormation 模板并更新适用于 Apache Flink 的托管服务应用程序？](#troubleshooting-blueprints-29)
+ [会代表我 AWS 监控和维护应用程序吗？](#troubleshooting-blueprints-30)
+ [此应用程序是否会将我的数据移出我 AWS 账户？](#troubleshooting-blueprints-31)
## 这个蓝图创造了哪些 AWS 资源？
要查找部署在您的账户中的资源，请导航到 AWS CloudFormation 控制台并确定以您为 Apache Flink 托管服务应用程序提供的名称开头的堆栈名称。选择**资源**选项卡，查看作为堆栈一部分创建的资源。以下是堆栈创建的关键资源：
+ 实时向量嵌入 Managed Service for Apache Flink 应用程序
+ 用于存放实时向量嵌入应用程序源代码的 Amazon S3 存储桶
+ CloudWatch 用于存储日志的日志组和日志流
+ 用于获取和创建资源的 Lambda 函数
+ 适用于 Lambdas、适用于 Apache Flink 应用程序的托管服务以及访问亚马逊 Bedrock 和亚马逊服务的 IAM 角色和策略 OpenSearch
+ Amazon OpenSearch 服务的数据访问政策
+ 用于访问亚马逊 Bedrock 和亚马逊 OpenSearch 服务的 VPC 终端节点
## AWS CloudFormation 堆栈部署完成后我的操作是什么？
AWS CloudFormation 堆栈部署完成后，访问适用于 Apache Flink 的托管服务控制台，找到您的蓝图 Apache Flink 托管服务应用程序。选择**配置**选项卡并确认所有运行时属性的设置是否正确。这些设置可能会溢出到下一页。如果您对这些设置放心，请选择**运行**。该应用程序将开始从您的主题中摄取消息。
要查看新版本，请参阅 [https://github.com/awslabs/real-time-vectorization-of-streaming-](https://github.com/awslabs/real-time-vectorization-of-streaming-data/releases) data/release。
## 源 Amazon MSK 主题中的数据结构应该是什么？
我们目前支持结构化和非结构化源数据。
+ 非结构化数据以 `source.msk.data.type` 中的 `STRING` 表示。从传入的消息中按原样读取数据。
+ 我们目前支持结构化的 JSON 数据（以 `source.msk.data.type` 中的 `JSON` 表示）。数据必须始终采用 JSON 格式。如果应用程序收到格式错误的 JSON，则应用程序将失败。
+ 使用 JSON 作为源数据类型时，请确保所有源主题中的每条消息都是有效的 JSON。如果您使用此设置订阅一个或多个不包含 JSON 对象的主题，则应用程序将失败。如果一个或多个主题混合了结构化和非结构化数据，我们建议您在 Managed Service for Apache Flink 应用程序中将源数据配置为非结构化数据。
## 我能否指定要嵌入的部分消息？
+ 对于非结构化输入数据（如果`source.msk.data.type`是）`STRING`，应用程序将始终嵌入整条消息并将整条消息存储在配置的 OpenSearch 索引中。
+ 对于结构化输入数据（其中 `source.msk.data.type` 为 `JSON`），您可以配置 `embed.input.config.json.fieldsToEmbed` 以指定应选择 JSON 对象中的哪个字段进行嵌入。这仅适用于顶级 JSON 字段，不适用于嵌套 JSONs 和包含 JSON 数组的消息。使用 .\$1 嵌入整个 JSON。
## 我是否可以从多个 Amazon MSK 主题中读取数据？
是，您可以使用此应用程序从多个 Amazon MSK 主题中读取数据。来自所有主题的数据均必须是相同的类型（STRING 或 JSON），否则可能会导致应用程序失败。来自所有主题的数据始终存储在单个 OpenSearch索引中。
## 我能否使用正则表达式来配置 Amazon MSK 主题名称？
`source.msk.topic.names` 不支持正则表达式列表。我们支持以逗号分隔的主题名称列表或包含所有主题的 `.*` 正则表达式。
## 可以从 Amazon MSK 主题中读取的消息的最大大小是多少？
可以处理的消息的最大大小受目前设置为 25,000,000 的 Amazon Bedrock InvokeModel 正文限制的限制。有关更多信息，请参阅 [InvokeModel](https://docs.aws.amazon.com/bedrock/latest/APIReference/API_runtime_InvokeModel.html#API_runtime_InvokeModel_RequestBody)。
## 支持 OpenSearch 哪种类型？
我们同时支持 OpenSearch 域名和集合。如果您使用的是 OpenSearch集合，请确保使用矢量集合并创建用于此应用程序的向量索引。这将允许您使用 OpenSearch 矢量数据库功能来查询数据。要了解更多信息，请参阅 [Amazon S OpenSearch ervice 的矢量数据库功能说明](https://aws.amazon.com/blogs/big-data/amazon-opensearch-services-vector-database-capabilities-explained/)。
## 为什么我需要使用矢量搜索集合、向量索引，并在我的 OpenSearch 无服务器集合中添加向量字段？
OpenSearchServerless 中的*向量搜索*集合类型提供了可扩展且性能高的相似度搜索功能。它简化了现代机器学习（ML）增强搜索体验和生成式人工智能（AI）应用程序的构建。有关更多信息，请参阅[使用向量搜索集合](https://docs.aws.amazon.com/opensearch-service/latest/developerguide/serverless-vector-search.html?icmpid=docs_console_unmapped)。
## 我应该将哪些方面设置为向量字段的维度？
根据要使用的嵌入模型设置向量字段的维度。请参阅下表，并确认相应文档中的这些值。
**向量字段维度**
| Amazon Bedrock 向量嵌入模型名称 | 模型提供的输出维度支持 |
| --- | --- |
| Amazon Titan 文本嵌入 V1 | 1,536 |
| Amazon Titan 文本嵌入 V2 | 1,024（默认）、384、256 |
| Amazon Titan Multimodal Embeddings G1 | 1,024（默认）、384、256 |
| Cohere Embed（英文版） | 1024 |
| Cohere Embed（多语版） | 1024 |
## 配置的 OpenSearch 索引中的输出是什么样子？
OpenSearch 索引中的每个文档都包含以下字段：
+ **original\$1data**：用于生成嵌入的数据。对于 STRING 类型，它是整条消息。对于 JSON 对象，它是用于嵌入的 JSON 对象。它可以是消息中的整个 JSON，也可以是 JSON 中的指定字段。例如，如果选择从传入的消息中嵌入名称，则输出将如下所示：
```
"original_data": "{\"name\":\"John Doe\"}"
```
+ **embedded\$1data**：由 Amazon Bedrock 生成的向量浮动嵌入数组
+ **日期**：存储文档的 UTC 时间戳 OpenSearch
## 我能否指定要添加到存储在 OpenSearch 索引中的文档的元数据字段？
不，目前，我们不支持向存储在 OpenSearch 索引中的最终文档中添加其他字段。
## 我应该指望 OpenSearch 索引中有重复的条目吗？
根据您配置应用程序的方式，您可能会在索引中看到重复的消息。一种常见的原因是应用程序重新启动。默认情况下，应用程序配置为从源主题中最早的消息开始读取。当您更改配置时，应用程序会重新启动并再次处理主题中的所有消息。为避免重新处理，请参阅有关如何使用 source.msk.starting.offset 并正确设置应用程序的起始偏移量的文档。
## 我可以向多个 OpenSearch 索引发送数据吗？
不是，该应用程序支持将数据存储到单个 OpenSearch 索引中。要将向量化输出设置为多个索引，必须部署单独的 Managed Service for Apache Flink 应用程序。
## 我能否在单个 AWS 账户中部署多个实时向量嵌入应用程序？
可以，如果每个应用程序都有唯一的名称，则可以在单个 AWS 账户 中部署多个实时向量嵌入 Managed Service for Apache Flink 应用程序。
## 多个实时向量嵌入应用程序能否使用相同的数据来源或接收器？
可以，您可以创建多个实时向量嵌入 Managed Service for Apache Flink 应用程序，这些应用程序从同一主题读取数据或将数据存储在同一个索引中。
## 该应用程序是否支持跨账户连接？
不，要使应用程序成功运行，Amazon MSK 集群和集 OpenSearch合必须与您尝试设置 Apache Flink 托管服务应用程序的 AWS 账户 位置相同。
## 该应用程序是否支持跨区域连接？
不可以，该应用程序仅允许您在Apache Flink托管服务应用程序的同一区域部署带有 Amazon MSK 集群和集 OpenSearch 合的托管服务 Flink 应用程序。
## 我的 Amazon MSK 集群和 OpenSearch 集合能否位于不同的 VPCs 或子网中？
是的，我们支持 Amazon MSK 集群 VPCs 和不同子网中的集 OpenSearch 合，前提是它们位于同一子网中。 AWS 账户请参阅（一般 MSF 故障排除），以确保您的设置正确。
## 该应用程序支持哪些嵌入模型？
目前，该应用程序支持 Bedrock 支持的所有模型。这些方法包括：
+ Amazon Titan Embeddings G1 - Text
+ Amazon Titan 文本嵌入 V2
+ Amazon Titan Multimodal Embeddings G1
+ Cohere Embed（英文版）
+ Cohere Embed（多语版）
## 我能否根据自己的工作负载微调应用程序的性能？
可以。应用程序的吞吐量取决于许多因素，所有这些因素都可以由客户控制：
1. **AWS MSF KPUs**：应用程序采用默认并行度系数 2 和每 KPU 1 的并行度进行部署，并行度已开启自动缩放。但是，建议您根据工作负载配置 Managed Service for Apache Flink 应用程序的扩展。有关更多信息，请参阅[审核 Managed Service for Apache Flink 应用程序资源](https://docs.aws.amazon.com/managed-flink/latest/java/how-resources.html)。
1. **Amazon Bedrock**：根据所选的 Amazon Bedrock 按需模式，可能适用不同的配额。在 Bedrock 中查看服务配额，以了解该服务能够处理的工作负载。有关更多信息，请参阅 [Amazon Bedrock 的配额](https://docs.aws.amazon.com/bedrock/latest/userguide/quotas.html)。
1. **Amazon S OpenSearch ervic** e：此外，在某些情况下，您可能会注意到 OpenSearch 这是您的管道中的瓶颈。有关扩展的信息，请参见 OpenSearch [缩放 Amazon OpenSearch 服务域](https://docs.aws.amazon.com/opensearch-service/latest/developerguide/sizing-domains.html)的大小。
## 支持哪些 Amazon MSK 身份验证类型？
我们仅支持 IAM MSK 身份验证类型。
## 什么是 `sink.os.bulkFlushIntervalMillis`，我该如何设置？
向 Amazon Serv OpenSearch ice 发送数据时，批量刷新间隔是批量请求的运行间隔，无论操作的数量或请求的大小如何。默认值设置为 1 毫秒。
虽然设置刷新间隔有助于确保及时为数据编制索引，但如果将该值设置得过低，也会导致开销增加。在选择刷新间隔时，请考虑您的使用案例以及及时编制索引的重要性。
## 当我部署 Managed Service for Apache Flink 应用程序时，它将从 Amazon MSK 主题的哪一点开始读取消息？
应用程序将开始读取 Amazon MSK 主题中的消息，偏移量由应用程序运行时配置中设置的 `source.msk.starting.offset` 配置指定。如果未明确设置 `source.msk.starting.offset`，则应用程序的默认行为是从主题中最早的可用消息开始读取。
## 如何使用 `source.msk.starting.offset`？
根据所需的行为，将 `ource.msk.starting.offset` 显式设置为以下值之一：
+ 最早：默认设置，即从分区中最早的偏移量读取。这是合适的选项，尤其是在以下情况下：
+ 您已经新创建 Amazon MSK 主题和使用者应用程序。
+ 您需要重播数据，这样才能构建或重建状态。在实现事件来源模式或初始化需要完整数据历史记录视图的新服务时，这一点很重要。
+ 最新：Managed Service for Apache Flink 应用程序将从分区末尾读取消息。如果您只关心生成新消息，并且不需要处理历史数据，我们建议您使用此选项。在此设置中，使用者将忽略现有消息，只读取上游创建者发布的新消息。
+ 已提交：Managed Service for Apache Flink 应用程序将开始使用来自使用组已提交偏移量的消息。如果提交的偏移量不存在，则将使用最早的重置策略。
## 支持哪些分块策略？
我们正在使用 [langchain](https://js.langchain.com/v0.1/docs/get_started/introduction/) 库对输入进行分块。只有当输入的长度大于所选的 `maxSegmentSizeInChars` 时，才会应用分块。我们支持以下五种分块类型：
+ `SPLIT_BY_CHARACTER`: 将尽可能多的字符放入每个区块中，其中每个区块的长度不大于。 maxSegmentSize InChars不关心空格，因此它可以截断单词。
+ `SPLIT_BY_WORD`: 将查找作为分块依据的空格字符。任何单词都不会被截断。
+ `SPLIT_BY_SENTENCE`: 使用带有英语语句模型的 Apache OpenNLP 库来检测语句边界。
+ `SPLIT_BY_LINE`: 将查找作为分块依据的换行字符。
+ `SPLIT_BY_PARAGRAPH`: 将查找作为分块依据的连续换行字符。
拆分策略会根据前面的顺序回退，其中较大的分块策略（例如 `SPLIT_BY_PARAGRAPH`）回退到 `SPLIT_BY_CHARACTER`。例如，在使用 `SPLIT_BY_LINE` 时，如果一行太长，则该行将按语句进行子分块，其中每个分块将尽可能多地放入语句。如果有任何语句太长，那么它将在单词层面上进行分块。如果一个单词太长，那么它将按字符拆分。
## 如何读取向量数据存储中的记录？
1. 当 `source.msk.data.type` 为 `STRING` 时
+ **original\$1data**：Amazon MSK 消息中的完整原始字符串。
+ **embedded\$1data**：如果嵌入向量不为空（已应用分块），则从 `chunk_data` 中创建该向量；如果未应用分块，则从 `original_data` 中创建该向量。
+ **chunk\$1data：**仅在对原始数据进行分块时才会出现。包含用于在 `embedded_data` 中创建嵌入的原始消息块。
1. 当 `source.msk.data.type` 为 `JSON` 时
+ **original\$1data**：应用 JSON 密钥筛选*后*来自 Amazon MSK 消息的完整原始 JSON。
+ **embedded\$1data**：如果嵌入向量不为空（已应用分块），则从 `chunk_data` 中创建该向量；如果未应用分块，则从 `original_data` 中创建该向量。
+ **chunk\$1key**：仅在对原始数据进行分块时才会出现。包含 `original_data` 中区块所属的 JSON 键。例如，在 `original_data` 的示例中，它可能看起来类似于嵌套键的 `jsonKey1.nestedJsonKeyA` 或*元数据*。
+ **chunk\$1data：**仅在对原始数据进行分块时才会出现。包含用于在 `embedded_data` 中创建嵌入的原始消息块。
是，您可以使用此应用程序从多个 Amazon MSK 主题中读取数据。来自所有主题的数据均必须是相同的类型（STRING 或 JSON），否则可能会导致应用程序失败。来自所有主题的数据始终存储在单个 OpenSearch索引中。
## 可在何处找到源代码的新更新？
前往 [https://github.com/awslabs/real-time-vectorization-of-streaming-data/release 查看新版本](https://github.com/awslabs/real-time-vectorization-of-streaming-data/releases)。
## 我能否更改 AWS CloudFormation 模板并更新适用于 Apache Flink 的托管服务应用程序？
不，更改 AWS CloudFormation 模板不会更新适用于 Apache Flink 的托管服务应用程序。中的任何新变化都 AWS CloudFormation 意味着需要部署新的堆栈。
## 会代表我 AWS 监控和维护应用程序吗？
不， AWS 不会代表您监控、扩展、更新或修补此应用程序。
## 此应用程序是否会将我的数据移出我 AWS 账户？
Apache Flink 托管服务应用程序读取和存储的所有数据都保留在您的 AWS 账户 账户中，永远不会离开您的帐户。
# 实时向量嵌入蓝图 - 故障排除
查看以下有关实时向量嵌入蓝图的故障排除主题。有关实时向量嵌入蓝图的更多信息，请参阅[实时向量嵌入蓝图](https://docs.aws.amazon.com/msk/latest/developerguide/ai-vector-embedding-integration-learn-more.html)。
**Topics**
+ [我的 CloudFormation 堆栈部署失败或已回滚。我可以采取哪些行动来修复此问题？](#troubleshooting-blueprints-deployment)
+ [我不想让自己的应用程序从 Amazon MSK 主题的开头开始读取消息。我应该怎么办？](#troubleshooting-blueprints-beginning)
+ [如何知道 Managed Service for Apache Flink 应用程序是否存在问题，以及如何对其进行调试？](#troubleshooting-blueprints-debug)
+ [我应该为自己的 Managed Service for Apache Flink 应用程序监控哪些关键指标？](#troubleshooting-blueprints-metrics)
## 我的 CloudFormation 堆栈部署失败或已回滚。我可以采取哪些行动来修复此问题？
+ 转到您的 CFN 堆栈并找到堆栈失败的原因。这可能与权限缺失、 AWS 资源名称冲突等原因有关。修复部署失败的根本原因。有关更多信息，请参阅[ CloudWatch 疑难解答指南](https://docs.aws.amazon.com/AWSCloudFormation/latest/UserGuide/troubleshooting.html#basic-ts-guide)。
+ [可选]每个 VPC 中的每个服务只能有一个 VPC 端点。如果您部署了多个实时矢量嵌入蓝图来写入同一 VPC 中的 Amazon Serv OpenSearch ice 集合，则它们可能会共享 VPC 终端节点。它们可能已经存在于您的 VPC 账户中，或者第一个实时矢量嵌入蓝图堆栈将为 Amazon Bedrock 和 Amazon Serv OpenSearch ice 创建 VPC 终端节点，供您账户中部署的所有其他堆栈使用。如果堆栈出现故障，请检查该堆栈是否为 Amazon Bedrock 和 Amazon Serv OpenSearch ice 创建了 VPC 终端节点，如果这些终端节点未在您的账户中的其他任何地方使用，则将其删除。有关删除 VPC 终端节点的步骤，请参阅有关如何安全删除应用程序的文档。
+ 您的账户中可能还有其他使用该 VPC 端点的服务或应用程序。删除该端点可能会导致其他服务的网络中断。删除这些端点时务必小心。
## 我不想让自己的应用程序从 Amazon MSK 主题的开头开始读取消息。我应该怎么办？
根据所需的行为，必须将 `source.msk.starting.offset` 显式设置为以下值之一：
+ **最早偏移量**：分区中最早的偏移量。
+ **最新偏移量**：使用者将从分区末端读取消息。
+ **提交的偏移量**：从使用者在分区内处理的最后一条消息中读取。
## 如何知道 Managed Service for Apache Flink 应用程序是否存在问题，以及如何对其进行调试？
使用 [Managed Service for Apache Flink 故障排除指南](https://docs.aws.amazon.com/managed-flink/latest/java/troubleshooting-runtime.html)，调试与应用程序有关的 Managed Service for Apache Flink 问题。
## 我应该为自己的 Managed Service for Apache Flink 应用程序监控哪些关键指标？
+ 常规 Managed Service for Apache Flink 应用程序的所有可用指标都可以帮助您监控应用程序。有关更多信息，请参阅 [Managed Service for Apache Flink 中的指标和维度](https://docs.aws.amazon.com/managed-flink/latest/java/metrics-dimensions.html)。
+ 要监控亚马逊 Bedrock 指标，请参阅[亚马逊 Bedrock 的亚马逊 CloudWatch 指标](https://docs.aws.amazon.com/bedrock/latest/userguide/monitoring.html#runtime-cloudwatch-metrics)。
+ 我们添加两个用于监控生成嵌入的性能的新指标。在中的`EmbeddingGeneration`操作名称下找到它们 CloudWatch。这两个指标是：
+ **BedrockTitanEmbeddingTokenCount**：向 Amazon Bedrock 发出的单次请求中存在的代币数量。
+ **BedrockEmbeddingGenerationLatencyMs**：报告发送和接收来自 Amazon Bedrock 的生成嵌入的响应所花费的时间（以毫秒为单位）。
+ 对于 Amazon Serv OpenSearch ice 无服务器集合，您可以使用诸如`IngestionDataRate``IngestionDocumentErrors`、之类的指标。有关更多信息，请参阅[使用 Amazon CloudWatch 监控 OpenSearch 无服务器](https://docs.aws.amazon.com/opensearch-service/latest/developerguide/monitoring-cloudwatch.html)。
+ 有关 OpenSearch 预配置的指标，请参阅[使用 Amazon CloudWatch 监控 OpenSearch 集群指标](https://docs.aws.amazon.com/opensearch-service/latest/developerguide/managedomains-cloudwatchmetrics.html)。
# 运行时问题排查
本节包含有关诊断和修复 Managed Service for Apache Flink 应用程序的运行时系统问题的信息。
**Topics**
+ [故障排除工具](#troubleshooting-tools)
+ [应用程序问题](troubleshooting-symptoms.md)
+ [应用程序正在重新启动](troubleshooting-rt-restarts.md)
+ [吞吐量太慢](troubleshooting-rt-throughput.md)
+ [无限制的状态增长](troubleshooting-rt-stateleaks.md)
+ [I/O 绑定运算符](troubleshooting-io-bound-operators.md)
+ [来自 Kinesis 数据流的上游或源限制](troubleshooting-source-throttling.md)
+ [检查点](troubleshooting-checkpoints.md)
+ [检查点操作已超时](troubleshooting-chk-timeout.md)
+ [Apache Beam 应用程序出现检查点故障](troubleshooting-chk-failure-beam.md)
+ [背压](troubleshooting-backpressure.md)
+ [数据偏斜](troubleshooting-data-skew.md)
+ [状态偏斜](troubleshooting-state-skew.md)
+ [集成不同区域中的资源](troubleshooting-resources-in-different-regions.md)
## 故障排除工具
检测应用程序问题的主要工具是 CloudWatch 警报。使用 CloudWatch 警报，您可以为指示应用程序中存在错误或瓶颈情况的 CloudWatch 指标设置阈值。有关推荐 CloudWatch 警报的信息，请参阅[在适用于 Apache Flink 的亚马逊托管服务中使用 CloudWatch 警报](monitoring-metrics-alarms.md)。
# 应用程序问题
本节包含 Managed Service for Apache Flink 应用程序中可能遇到的错误情况的解决方案。
**Topics**
+ [应用程序停留在临时状态](#troubleshooting-rt-stuck)
+ [快照创建失败](#troubleshooting-rt-snapshots)
+ [无法访问 VPC 中的资源](#troubleshooting-rt-vpc)
+ [在写入到 Amazon S3 存储桶时丢失数据](#troubleshooting-rt-s3)
+ [应用程序处于“运行”状态，但未处理数据](#troubleshooting-rt-processing)
+ [快照、应用程序更新或应用程序停止错误： InvalidApplicationConfigurationException](#troubleshooting-rt-appconfigexception)
+ [java.nio.file。 NoSuchFileException:/usr/local/openjdk-8/lib/security/cacerts](#troubleshooting-rt-fnf)
## 应用程序停留在临时状态
如果您的应用程序处于临时状态（`STARTING``UPDATING`、`STOPPING`、或`AUTOSCALING`），则可以使用`Force`参数设置为的[StopApplication](https://docs.aws.amazon.com/managed-flink/latest/apiv2/API_StopApplication.html)操作来停止应用程序`true`。您不能强制停止处于该`DELETING`状态的应用程序。或者，如果应用程序处于`UPDATING`或`AUTOSCALING`状态，则可以将其回滚到之前运行的版本。回滚应用程序时，它会从上次成功的快照中加载状态数据。如果应用程序没有快照，则 Managed Service for Apache Flink会拒绝回滚请求。有关回滚应用程序的更多信息，请参阅[RollbackApplication](https://docs.aws.amazon.com/managed-flink/latest/apiv2/API_RollbackApplication.html)操作。
**注意**
强制停止应用程序可能会导致数据丢失或重复。为了防止在应用程序重启期间丢失数据或重复处理数据，我们建议您经常为应用程序拍摄快照。
应用程序卡住的原因包括：
+ **应用程序状态太大：**应用程序状态过大或过于持久可能会导致应用程序在检查点或快照操作期间卡住。检查您的应用程序`lastCheckpointDuration`和`lastCheckpointSize`指标是否有稳步增加的值或异常高的值。
+ **应用程序代码太大：**验证应用程序 JAR 文件小于 512 MB。不支持超过 512 MB 的 JAR 文件。
+ **应用程序快照创建失败：**Managed Service for Apache Flink 在[https://docs.aws.amazon.com/managed-service-for-apache-flink/latest/apiv2/API_UpdateApplication.html](https://docs.aws.amazon.com/managed-service-for-apache-flink/latest/apiv2/API_UpdateApplication.html)或[https://docs.aws.amazon.com/managed-service-for-apache-flink/latest/apiv2/API_StopApplication.html](https://docs.aws.amazon.com/managed-service-for-apache-flink/latest/apiv2/API_StopApplication.html)请求期间拍摄应用程序的快照。然后，服务使用此快照状态并使用更新的应用程序配置恢复应用程序，以提供*一次性*处理语义。如果自动快照创建失败，请参阅[快照创建失败](#troubleshooting-rt-snapshots)以下内容。
+ **从快照恢复失败：**如果在应用程序更新中删除或更改一个操作符并尝试从快照中还原，默认情况下，如果快照包含缺少的操作符的状态数据，还原将失败。此外，应用程序将停滞在 `STOPPED` 或 `UPDATING` 状态。要更改此行为并允许恢复成功，请将应用程序的*AllowNonRestoredState*参数更改[FlinkRunConfiguration](https://docs.aws.amazon.com/managed-flink/latest/apiv2/API_FlinkRunConfiguration.html)为`true`。这样，恢复操作就可以跳过无法映射到新程序的状态数据。
+ **应用程序初始化需要更长的时间：**Managed Service for Apache Flink 在等待 Flink 任务启动时使用 5 分钟的内部超时（软设置）。如果您的作业未能在此超时时间内启动，您将看到如下 CloudWatch 日志：
```
Flink job did not start within a total timeout of 5 minutes for application: %s under account: %s
```
如果您遇到上述错误，则表示您在 Flink 任务的`main`方法下定义的操作花费的时间超过 5 分钟，从而导致 Managed Service for Apache Flink 端的 Flink 任务创建超时。我们建议你查看 Flink **JobManager**日志和应用程序代码，看看`main`方法是否会出现这种延迟。如果没有，则需要采取措施解决问题，以便在 5 分钟内完成。
您可以使用 [https://docs.aws.amazon.com/managed-service-for-apache-flink/latest/apiv2/API_ListApplications.html](https://docs.aws.amazon.com/managed-service-for-apache-flink/latest/apiv2/API_ListApplications.html) 或 [https://docs.aws.amazon.com/managed-service-for-apache-flink/latest/apiv2/API_DescribeApplication.html](https://docs.aws.amazon.com/managed-service-for-apache-flink/latest/apiv2/API_DescribeApplication.html) 操作检查应用程序状态。
## 快照创建失败
在以下情况下，Managed Service for Apache Flink 无法拍摄快照：
+ 应用程序超过快照限制。快照限制为 1,000 个。有关更多信息，请参阅 [使用快照管理应用程序备份](how-snapshots.md)。
+ 应用程序无权访问其源或接收器。
+ 应用程序代码无法正常工作。
+ 应用程序遇到其他配置问题。
如果在应用程序更新期间或停止应用程序时拍摄快照时遇到异常，请将应用程序的 [https://docs.aws.amazon.com/managed-service-for-apache-flink/latest/apiv2/API_ApplicationSnapshotConfiguration.html](https://docs.aws.amazon.com/managed-service-for-apache-flink/latest/apiv2/API_ApplicationSnapshotConfiguration.html)属性设置`SnapshotsEnabled`为`false`，然后重试请求。
如果未正确配置应用程序的操作员，则快照可能会失败。有关调整运算符性能的信息，请参见[运算符扩展](performance-improving.md#performance-improving-scaling-op)。
在应用程序恢复正常状态后，我们建议您将应用程序的`SnapshotsEnabled` 属性设置为 `true`。
## 无法访问 VPC 中的资源
如果应用程序使用在 Amazon VPC 上运行的 VPC，请执行以下操作以验证应用程序是否有权访问其资源：
+ 检查您的 CloudWatch 日志中是否存在以下错误。该错误表明应用程序无法访问 VPC 中的资源：
```
org.apache.kafka.common.errors.TimeoutException: Failed to update metadata after 60000 ms.
```
如果看到该错误，请确认正确设置了路由表，并且连接器具有正确的连接设置。
有关设置和分析 CloudWatch 日志的信息，请参阅[Amazon Managed Service for Apache Flink 的日志记录和监控](monitoring-overview.md)。
## 在写入到 Amazon S3 存储桶时丢失数据
使用 Apache Flink 版本 1.6.2 将输出写入 Amazon S3 存储桶时，可能会发生一些数据丢失。在直接使用 Amazon S3 存储输出时，我们建议使用最新支持的 Apache Flink 版本。要使用 Apache Flink 1.6.2 写入到 Amazon S3 存储桶，我们建议使用 Firehose。有关将 Firehose 与 Managed Service for Apache Flink 结合使用的更多信息，请参阅 [Firehose 接收器](earlier.md#get-started-exercise-fh)。
## 应用程序处于“运行”状态，但未处理数据
您可以使用 [https://docs.aws.amazon.com/managed-service-for-apache-flink/latest/apiv2/API_ListApplications.html](https://docs.aws.amazon.com/managed-service-for-apache-flink/latest/apiv2/API_ListApplications.html) 或 [https://docs.aws.amazon.com/managed-service-for-apache-flink/latest/apiv2/API_DescribeApplication.html](https://docs.aws.amazon.com/managed-service-for-apache-flink/latest/apiv2/API_DescribeApplication.html) 操作检查应用程序状态。如果您的应用程序进入`RUNNING`状态但未向接收器写入数据，则可以通过向应用程序添加 Amazon CloudWatch 日志流来解决问题。有关更多信息，请参阅 [使用应用程序 CloudWatch 日志选项](cloudwatch-logs.md#adding_cloudwatch)。日志流包含可用于解决应用程序问题的消息。
## 快照、应用程序更新或应用程序停止错误： InvalidApplicationConfigurationException
在快照操作期间或创建快照的操作（例如更新或停止应用程序）期间，可能会出现类似下面的错误：
```
An error occurred (InvalidApplicationConfigurationException) when calling the UpdateApplication operation:
Failed to take snapshot for the application xxxx at this moment. The application is currently experiencing downtime.
Please check the application's CloudWatch metrics or CloudWatch logs for any possible errors and retry the request.
You can also retry the request after disabling the snapshots in the Managed Service for Apache Flink console or by updating
the ApplicationSnapshotConfiguration through the AWS SDK
```
在应用程序无法创建快照时，将会出现该错误。
如果在快照操作期间或创建快照的操作期间遇到该错误，请执行以下操作：
+ 为应用程序禁用快照。您可以在适用于 Apache Flink 的托管服务控制台中执行此操作，也可以使用操作的`SnapshotsEnabledUpdate`参数来执行此操作。[UpdateApplication](https://docs.aws.amazon.com/managed-flink/latest/apiv2/API_UpdateApplication.html)
+ 调查无法创建快照的原因。有关更多信息，请参阅 [应用程序停留在临时状态](#troubleshooting-rt-stuck)。
+ 在应用程序恢复正常状态时，重新启用快照。
## java.nio.file。 NoSuchFileException:/usr/local/openjdk-8/lib/security/cacerts
在以前的部署中更新了 SSL 信任存储库位置。请在 `ssl.truststore.location` 参数中改用以下值：
```
/usr/lib/jvm/java-11-amazon-corretto/lib/security/cacerts
```
# 应用程序正在重新启动
如果您的应用程序运行状况不佳，则其 Apache Flink 任务会持续失败并重新启动。本节介绍这种情况的症状和故障排除步骤。
## 症状
这种情况可能具有以下症状：
+ 该 `FullRestarts` 指标不为零。此指标表示自您启动应用程序以来应用程序任务重新启动的次数。
+ 该 `Downtime` 指标不为零。此指标表示应用程序处于 `FAILING` 或 `RESTARTING` 状态的毫秒数。
+ 应用程序日志包含对 `RESTARTING` 或 `FAILED` 的状态更改。您可以使用以下 Logs Insights 查询来查询应用程序 CloudWatch 日志以了解这些状态变化:[分析错误：应用程序的任务相关故障](cloudwatch-logs-reading.md#cloudwatch-logs-reading-apps).
## 原因和解决方案
以下情况可能会导致您的应用程序变得不稳定并反复重启：
+ **运算符抛出异常：**如果应用程序中的运算符中的任何异常未得到处理，则应用程序会（解读为运算符无法处理故障）进行故障转移。应用程序从最新的检查点重新启动，以保持“只执行一次”的处理语义。因此，`Downtime` 在这些重启期间不为零。为了防止发生这种情况，我们建议您在应用程序代码中处理任何可重试的异常。
您可以查询应用程序日志以确定应用程序状态是否从 `RUNNING` 变为 `FAILED`，以调查发生这种情况的原因。有关更多信息，请参阅 [分析错误：应用程序的任务相关故障](cloudwatch-logs-reading.md#cloudwatch-logs-reading-apps)。
+ **未正确配置 Kinesis Data Streams：**如果您的应用程序的源或接收器是 Kinesis 数据流，请检查该流的[指标](https://docs.aws.amazon.com/streams/latest/dev/monitoring-with-cloudwatch.html)是否存在 `ReadProvisionedThroughputExceeded` 或 `WriteProvisionedThroughputExceeded` 错误。
如果您看到这些错误，则可以通过增加 Kinesis 流的分片数量来增加 Kinesis 流的可用吞吐量。有关重新分片的更多信息，请参阅[如何更改 Kinesis Data Streams 中打开的分片数？](https://aws.amazon.com/premiumsupport/knowledge-center/kinesis-data-streams-open-shards/)
+ **其他源或接收器未正确配置或不可用：**验证您的应用程序是否正确配置了源和接收器。检查应用程序中使用的任何源或接收器（例如其他 AWS 服务、外部源或目标）是否配置良好，没有遇到读取或写入限制，或者定期不可用。
如果您的依赖服务遇到吞吐量相关的问题，请增加这些服务的可用资源，或者调查任何错误或不可用的原因。
+ **未正确配置运算符：**如果应用程序中其中一个运算符的线程上的工作负载分配不正确，则该运算符可能会过载，应用程序可能会崩溃。有关调整运算符并行度的信息，请参见 [正确管理运算符扩展](performance-improving.md#performance-improving-scaling-op)。
+ **应用程序失败 DaemonException：**如果您使用的是 1.11 之前的 Apache Flink 版本，则此错误会出现在应用程序日志中。您可能需要升级到更高版本的 Apache Flink，这样才能使用 KPL 0.14 或更高版本。
+ **应用程序失败 TimeoutException，并出现 FlinkException、或 RemoteTransportException：**如果任务管理器崩溃，这些错误可能会出现在应用程序日志中。如果您的应用程序过载，您的任务管理器可能会承受 CPU 或内存资源压力，从而导致它们故障。
这些错误可能与以下内容相似：
+ `java.util.concurrent.TimeoutException: The heartbeat of JobManager with id xxx timed out`
+ `org.apache.flink.util.FlinkException: The assigned slot xxx was removed`
+ `org.apache.flink.runtime.io.network.netty.exception.RemoteTransportException: Connection unexpectedly closed by remote task manager`
若要对这种情况进行故障排除，请检查以下内容：
+ 检查您的 CloudWatch 指标，看看 CPU 或内存使用率是否出现异常峰值。
+ 检查您的应用程序是否存在吞吐量问题。有关更多信息，请参阅 [性能问题排查](performance-troubleshooting.md)。
+ 检查您的应用程序日志，了解您的应用程序代码引发的未处理异常。
+ **应用程序失败并出现 “ JaxbAnnotationModule 未找到” 错误：**如果您的应用程序使用 Apache Beam，但没有正确的依赖项或依赖项版本，则会发生此错误。使用 Apache Beam 的 Managed Service for Apache Flink 应用程序必须使用以下版本的依赖项：
```
2.10.2
...
com.fasterxml.jackson.module
jackson-module-jaxb-annotations
2.10.2
```
如果您没有提供正确的 `jackson-module-jaxb-annotations` 版本作为显式依赖项，则您的应用程序会从环境依赖项中加载该版本，并且由于版本不匹配，应用程序会在运行时系统崩溃。
有关将 Apache Beam 与 Managed Service for Apache Flink结合使用的更多信息，请参阅[使用 CloudFormation使用 Apache Beam 创建应用程序](examples-beam.md)。
+ **应用程序因使用 java.io 而失败。 IOException: 网络缓冲区数量不足**
当应用程序没有为网络缓冲区分配足够的内存时，就会发生这种情况。网络缓冲区便于子任务之间的通信。它们用于在通过网络传输之前存储记录，也用于存储传入的数据，然后再将其解析成记录并交给子任务。所需的网络缓冲区数量直接随着任务图的并行度和复杂性而变化。有多种方法可以缓解此问题：
+ 您可以配置较低的配置，`parallelismPerKpu`以便为每个子任务和网络缓冲区分配更多的内存。请注意，降低`parallelismPerKpu`会增加 KPU，从而增加成本。为避免这种情况，您可以通过将并行度降低相同的系数来保持相同数量的 KPU。
+ 您可以通过减少运算符的数量或链接运算符来简化任务图，从而减少所需的缓冲区。
+ 否则，您可以联系以 https://aws.amazon.com/premiumsupport/获取自定义网络缓冲区配置。
# 吞吐量太慢
如果您的应用程序处理传入的流数据速度不够快，则其性能将不佳并变得不稳定。本节介绍这种情况的症状和故障排除步骤。
## 症状
这种情况可能具有以下症状：
+ 如果您的应用程序的数据源是 Kinesis 流，则该流的`millisbehindLatest`指标会不断增加。
+ 如果您的应用程序的数据源是 Amazon MSK 集群，则该集群的使用者延迟指标会不断增加。有关更多信息，请参阅 [Amazon MSK 开发人员指南](https://docs.aws.amazon.com/msk/latest/developerguide/what-is-msk.html)中的[消费端延迟监控](https://docs.aws.amazon.com/msk/latest/developerguide/consumer-lag.html)。
+ 如果您的应用程序的数据源是不同的服务或来源，请检查所有可用的消费者延迟指标或可用数据。
## 原因和解决方案
导致应用程序吞吐量缓慢的原因可能有很多。如果您的应用程序无法跟上输入的速度，请检查以下内容：
+ 如果吞吐量延迟激增然后逐渐减弱，请检查应用程序是否正在重新启动。您的应用程序将在重新启动时停止处理输入，从而导致延迟激增。有关应用程序故障的更多信息，请参阅[应用程序正在重新启动](troubleshooting-rt-restarts.md)。
+ 如果吞吐量延迟一致，请检查您的应用程序是否针对性能进行了优化。有关优化应用程序性能的信息，请参阅[性能问题排查](performance-troubleshooting.md)。
+ 如果吞吐量延迟不是激增而是持续增加，并且您的应用程序已针对性能进行了优化，则必须增加应用程序资源。有关增加应用程序资源的信息，请参阅[实施应用程序扩展](how-scaling.md)。
+ 如果您的应用程序从不同区域的 Kafka 集群读取数据，`FlinkKafkaConsumer`或者尽管`KafkaSource`使用者延迟很高，但大部分时间处于空闲状态（高`idleTimeMsPerSecond`或低`CPUUtilization`），则可以增加的值`receive.buffer.byte`，例如 2097152。有关更多信息，请参阅[自定义 MSK 配置](https://docs.aws.amazon.com/msk/latest/developerguide/msk-configuration-properties.html)中的高延迟环境部分。
有关应用程序源中吞吐量缓慢或消费者延迟增加的故障排除步骤，请参阅[性能问题排查](performance-troubleshooting.md)。
# 无限制的状态增长
如果您的应用程序未正确处理过时的状态信息，则这些信息将不断积累并导致应用程序性能或稳定性出现问题。本节介绍这种情况的症状和故障排除步骤。
## 症状
这种情况可能具有以下症状：
+ 该`lastCheckpointDuration`指标正在逐渐增加或激增。
+ 该`lastCheckpointSize`指标正在逐渐增加或激增。
## 原因和解决方案
以下情况可能会导致您的应用程序积累状态数据：
+ 您的应用程序保留状态数据的时间超过了所需的时间。
+ 您的应用程序使用持续时间过长的窗口查询。
+ 您没有为状态数据设置 TTL。有关更多信息，请参阅 Apache Flink [文档中的状态 Time-To-Live (TTL)](https://nightlies.apache.org/flink/flink-docs-release-1.18/docs/dev/datastream/fault-tolerance/state/#state-time-to-live-ttl)。
+ 您正在运行的应用程序依赖于 Apache Beam 版本 2.25.0 或更高版本。你可以选择退出新版本的读取转换，方法是 BeamApplicationProperties使用关键实验和值来[扩展你的](https://docs.aws.amazon.com/managed-flink/latest/java/examples-beam.html#examples-beam-configure)读取转换`use_deprecated_read`。有关更多信息，请参阅 [Apache Beam 文档](https://beam.apache.org/blog/beam-2.25.0/#highlights)。
有时，应用程序会面临不断增加的状态规模增长，从长远来看，这是不可持续的（毕竟 Flink 应用程序可以无限期运行）。有时，这可以追溯到应用程序将数据存储在状态中，而旧信息没有正确过期。但是有时候人们对 Flink 能提供的功能抱有不合理的期望。应用程序可以在跨几天甚至几周的大时间窗口内使用聚合。[AggregateFunctions](https://nightlies.apache.org/flink/flink-docs-stable/docs/dev/datastream/operators/windows/#aggregatefunction)除非使用允许增量聚合，否则 Flink 需要保持整个窗口的事件处于状态。
此外，在使用流程函数实现自定义运算符时，应用程序需要从状态中删除业务逻辑不再需要的数据。在这种情况下， time-to-live可以使用 st [at](https://nightlies.apache.org/flink/flink-docs-stable/docs/dev/datastream/fault-tolerance/state/#state-time-to-live-ttl) e 根据处理时间自动使数据过时。Managed Service for Apache Flink使用增量检查点，因此状态 ttl 基于 [RocksDB 压缩](https://github.com/facebook/rocksdb/wiki/Compaction)。只有在压缩操作发生后，您才能观察到状态大小的实际缩小（由检查点大小表示）。特别是对于小于 200 MB 的检查点大小，您不太可能观察到由于状态过期而导致检查点大小缩小。但是，保存点基于不包含旧数据的状态的干净副本，因此您可以在 Managed Service for Apache Flink中触发快照，以强制删除过期的状态。
出于调试目的，禁用增量检查点以更快地验证检查点大小是否确实减小或稳定下来（并避免 RockSB 中压缩的影响）将很有用。不过，这需要向服务团队创建工单。
# I/O 绑定运算符
最好避免在数据路径上依赖外部系统。保持参考数据集的状态通常比查询外部系统来丰富单个事件的性能要高得多。但是，有时会有一些依赖关系无法轻易移动到状态，例如，如果您想使用托管在 Amazon Sagemaker 上的机器学习模型来丰富事件。
通过网络与外部系统连接的运营商可能会成为瓶颈并造成背压。强烈建议使用 [AsyncIO](https://nightlies.apache.org/flink/flink-docs-stable/docs/dev/datastream/operators/asyncio/) 来实现该功能，以减少单个呼叫的等待时间并避免整个应用程序变慢。
此外，对于具有 I/O 绑定运算符的应用程序，增加适用于 Apache Flink 的托管服务应用程序的 [ParallelismPerKPU](https://docs.aws.amazon.com/managed-flink/latest/apiv2/API_ParallelismConfiguration.html) 设置也是有意义的。此配置描述应用程序在其每个 Kinesis 处理单元 (KPU) 可以执行的并行子任务数。通过将值从默认值 1 增加到（比如）4，应用程序可以利用相同的资源（且成本相同），但可以扩展到并行度的 4 倍。这对于 I/O 绑定的应用程序来说效果很好，但它会给未 I/O 绑定的应用程序带来额外的开销。
# 来自 Kinesis 数据流的上游或源限制
**症状**：应用程序遇到 `LimitExceededExceptions` 来自其上游源 Kinesis 数据流的问题。
**潜在原因**：Apache Flink 库 Kinesis 连接器的默认设置为从 Kinesis 数据流源读取，对于每次 `GetRecords` 调用获取的最大记录数，默认设置非常激进。默认情况下，Apache Flink 配置为每次 `GetRecords` 调用获取 10,000 条记录（默认情况下，此调用每 200 毫秒进行一次），尽管每个分片的限制只有 1,000 条记录。
当尝试从 Kinesis 数据流中使用时，此默认行为可能会导致限制，从而影响应用程序的性能和稳定性。
您可以通过检查 CloudWatch `ReadProvisionedThroughputExceeded`指标并查看该指标大于零的长期或持续时间来确认这一点。
通过观察持续`LimitExceededException`存在的错误，您还可以在适用于 Apache Flink 的亚马逊托管服务 Flink 应用程序的 CloudWatch 日志中看到这一点。
**解决方案**：您可以采取以下两项措施之一来解决这种情况：
+ 降低每次 `GetRecords` 调用提取的记录数的默认限制
+ 在 Amazon Managed Service for Apache Flink 应用程序中启用自适应读取。有关自适应读取功能的更多信息，请参阅 [SHARD\$1USE\$1ADAPTIVE\$1READS](https://nightlies.apache.org/flink/flink-docs-release-1.10/api/java/org/apache/flink/streaming/connectors/kinesis/config/ConsumerConfigConstants.html#SHARD_USE_ADAPTIVE_READS)
# 检查点
检查点是 Flink 的机制，用于确保应用程序的状态具有容错性。该机制允许 Flink 在任务失败时恢复运算符的状态，并为应用程序提供与无故障执行相同的语义。使用适用于 Apache Flink 的托管服务，应用程序的状态存储在 RocksDB 中，RocksDB 是一个嵌入式存 key/value 储，可将其工作状态保持在磁盘上。当使用检查点时，状态也会上传到 Amazon S3，因此，即使磁盘丢失，也可以使用该检查点来恢复应用程序状态。
有关更多信息，请参阅[状态快照的工作原理？](https://nightlies.apache.org/flink/flink-docs-master/docs/learn-flink/fault_tolerance/#how-does-state-snapshotting-work) 。
## 通过检查点检验阶段
对于 Flink 中的检查点操作员子任务，有 5 个主要阶段：
+ 等待 [**开始延迟**] — Flink 使用插入到直播中的检查点屏障，因此此阶段的时间是操作员等待检查点屏障到达的时间。
+ 对**齐 [对齐持续时间**]-在此阶段，子任务已到达一个屏障，但它正在等待来自其他输入流的屏障。
+ Sync checkpointing [**同步持续时间**] — 此阶段是子任务实际捕捉操作员的状态并阻止子任务上的所有其他活动的时候。
+ 异步检查点 [**异步持续时间**] — 此阶段的大部分时间是将状态上传到 Amazon S3 的子任务。在此阶段，子任务不再被阻止，可以处理记录。
+ 确认 — 这通常是一个很短的阶段，只是子任务向发送确认 JobManager 并执行任何提交消息（例如使用 Kafka sinks）。
每个阶段（确认除外）都映射到 Flink WebUI 中提供的检查点的持续时间指标，这可以帮助找出长检查点的原因。
要查看检查点上每个可用指标的确切定义，请转到 [“历史记录” 选项卡](https://nightlies.apache.org/flink/flink-docs-release-1.13/docs/ops/monitoring/checkpoint_monitoring/#history-tab)。
## 正在调查
在调查较长的检查点持续时间时，要确定的最重要因素是检查点的瓶颈，即哪个操作员和子任务到达检查点的时间最长，以及该子任务的哪个阶段需要较长的时间。这可以通过任务检查点任务下的 Flink WebUI 来确定。Flink 的 Web 界面提供了有助于调查检查点问题的数据和信息。有关完整细分，请参阅[监控检查点。](https://nightlies.apache.org/flink/flink-docs-release-1.13/docs/ops/monitoring/checkpoint_monitoring/)
首先要看的是 Job 图表中每个操作员的**端到端持续时间**，以确定哪个操作员需要很长时间才能完成检查点并值得进一步调查。根据 Flink 文档，时长的定义是：
*从触发时间戳到最近一次确认（或者 n/a 如果尚未收到确认）的持续时间。完整检查点的端到端持续时间由最后一个确认该检查点的子任务决定。这个时间通常比单个子任务实际检查状态所需的时间还要长。*
检查点的其他持续时间也提供了有关时间花在何处的更精细的信息。
如果 “**同步持续时间**” 较高，则表示快照期间发生了某些事情。在此阶段`snapshotState()`，将调用实现 SnapshotState 接口的类；这可能是用户代码，因此线程转储可用于对此进行调查。
如果**异步持续时间**过长，则表明需要花费大量时间将状态上传到 Amazon S3。如果状态很大，或者正在上传的状态文件很多，则可能会发生这种情况。如果是这样的话，那么值得研究一下应用程序是如何使用状态的，并确保尽可能使用 Flink 原生数据结构（[使用键控状态](https://nightlies.apache.org/flink/flink-docs-master/docs/dev/datastream/fault-tolerance/state/#using-keyed-state)）。Managed Service for Apache Flink 配置 Flink 的方式可以最大限度地减少 Amazon S3 的调用次数，从而确保调用时间不会太长。以下是操作员的检查点统计信息示例。它表明，与前面的操作员检查点统计数据相比，**异步持续时间**相对较长。
![\[调查检查点检验\]](http://docs.aws.amazon.com/zh_cn/managed-flink/latest/java/images/checkpoint.png)
如果**启动延迟**过高，则表明大部分时间都花在等待检查点屏障到达操作员身上。这表明应用程序需要一段时间来处理记录，这意味着屏障正在缓慢地流过任务图。如果 Job 受到反压或者操作员经常忙碌，通常会出现这种情况。以下是第二个 KeyedProcess 操作员忙碌的示例。 JobGraph
![\[调查检查点检验\]](http://docs.aws.amazon.com/zh_cn/managed-flink/latest/java/images/checkpoint2.png)
你可以使用 Flink Flame Graphs 或 TaskManager 话题转储来调查花了这么长时间的事情。一旦确定了瓶颈，就可以使用 Flame-Graphs 或 thread-dumps 对其进行进一步研究。
## 线程转储
线程转储是另一种调试工具，其级别略低于火焰图。线程转储输出所有线程在某个时间点的执行状态。Flink 采用 JVM 线程转储，这是 Flink 进程中所有线程的执行状态。线程的状态由线程的堆栈跟踪以及一些其他信息来呈现。火焰图实际上是使用快速连续采集的多个堆栈轨迹来构建的。该图表是由这些轨迹制成的可视化效果，便于识别常见的代码路径。
```
"KeyedProcess (1/3)#0" prio=5 Id=1423 RUNNABLE
at app//scala.collection.immutable.Range.foreach$mVc$sp(Range.scala:154)
at $line33.$read$$iw$$iw$ExpensiveFunction.processElement(>19)
at $line33.$read$$iw$$iw$ExpensiveFunction.processElement(:14)
at app//org.apache.flink.streaming.api.operators.KeyedProcessOperator.processElement(KeyedProcessOperator.java:83)
at app//org.apache.flink.streaming.runtime.tasks.OneInputStreamTask$StreamTaskNetworkOutput.emitRecord(OneInputStreamTask.java:205)
at app//org.apache.flink.streaming.runtime.io.AbstractStreamTaskNetworkInput.processElement(AbstractStreamTaskNetworkInput.java:134)
at app//org.apache.flink.streaming.runtime.io.AbstractStreamTaskNetworkInput.emitNext(AbstractStreamTaskNetworkInput.java:105)
at app//org.apache.flink.streaming.runtime.io.StreamOneInputProcessor.processInput(StreamOneInputProcessor.java:66)
...
```
上面是从 Flink 用户界面中获取的单个线程的线程转储片段。第一行包含有关此线程的一些一般信息，包括：
+ 话题名称 *KeyedProcess (1/3) \$10*
+ 线程的优先级 *prio=5*
+ 一个唯一的线程Id *Id =1423*
+ 线程状态可*运行*
线程的名称通常提供有关线程一般用途的信息。操作员线程可以通过其名称来识别，因为操作员线程与操作员线程具有相同的名称，并且可以指示它与哪个子任务有关，例如，*KeyedProcess (1/3) \$10* 线程来自*KeyedProcess*操作员，来自第 1 个（共 3 个）子任务。
线程可能处于以下几种状态之一：
+ 新增-线程已创建但尚未处理
+ RUNNABLE — 线程在 CPU 上执行
+ BLOCKED — 该线程正在等待另一个线程释放其锁定
+ 等待-线程正在使用`wait()``join()`、或`park()`方法等待
+ TIMED\$1WAITING — 线程正在使用睡眠、等待、加入或暂留方法等待，但等待时间最长。
**注意**
在 Flink 1.13 中，线程转储中单个堆栈跟踪的最大深度限制为 8。
**注意**
线程转储应该是调试 Flink 应用程序中性能问题的最后手段，因为线程转储可能难以读取，需要采集多个样本并进行手动分析。如果可能的话，最好使用火焰图。
### Flink 中的线程转储了
在 Flink 中，可以通过选择 Flink 用户界面左侧导航栏上的 “**任务管理**器” 选项，选择特定的任务管理器，然后导航到 “线程转储” 选项卡来进行**线程**转储。线程转储可以下载、复制到你最喜欢的文本编辑器（或线程转储分析器），也可以直接在 Flink Web UI 的文本视图中进行分析（但是，最后一个选项可能有点笨拙。
要确定要使用哪个任务管理器，可以在选择特定的运算符时使用该**TaskManagers**选项卡的线程转储。这表明操作员正在操作员的不同子任务上运行，并且可以在不同的任务管理器上运行。
![\[使用线程转储\]](http://docs.aws.amazon.com/zh_cn/managed-flink/latest/java/images/checkpoint4.png)
转储将由多个堆栈跟踪组成。但是，在调查垃圾场时，与操作员相关的内容是最重要的。这些可以很容易地找到，因为操作员线程与操作员线程具有相同的名称，并且可以指示它与哪个子任务有关。例如，以下堆栈跟踪来自*KeyedProcess*操作员，并且是第一个子任务。
```
"KeyedProcess (1/3)#0" prio=5 Id=595 RUNNABLE
at app//scala.collection.immutable.Range.foreach$mVc$sp(Range.scala:155)
at $line360.$read$$iw$$iw$ExpensiveFunction.processElement(:19)
at $line360.$read$$iw$$iw$ExpensiveFunction.processElement(:14)
at app//org.apache.flink.streaming.api.operators.KeyedProcessOperator.processElement(KeyedProcessOperator.java:83)
at app//org.apache.flink.streaming.runtime.tasks.OneInputStreamTask$StreamTaskNetworkOutput.emitRecord(OneInputStreamTask.java:205)
at app//org.apache.flink.streaming.runtime.io.AbstractStreamTaskNetworkInput.processElement(AbstractStreamTaskNetworkInput.java:134)
at app//org.apache.flink.streaming.runtime.io.AbstractStreamTaskNetworkInput.emitNext(AbstractStreamTaskNetworkInput.java:105)
at app//org.apache.flink.streaming.runtime.io.StreamOneInputProcessor.processInput(StreamOneInputProcessor.java:66)
...
```
如果有多个同名的运算符，这可能会变得令人困惑，但我们可以命名运算符来解决这个问题。例如：
```
....
.process(new ExpensiveFunction).name("Expensive function")
```
## [火焰图](https://nightlies.apache.org/flink/flink-docs-release-1.13/docs/ops/debugging/flame_graphs/)
Flame graphs 是一种有用的调试工具，可以可视化目标代码的堆栈轨迹，从而可以识别最常见的代码路径。它们是通过对堆栈轨迹进行多次采样来创建的。火焰图的 x 轴显示不同的堆栈配置文件，而 y 轴显示堆栈深度以及堆栈跟踪中的调用。火焰图中的单个矩形表示堆栈框架，框架的宽度表示它在堆栈中出现的频率。有关火焰图及其使用方法的更多详细信息，请参阅[火焰图](https://www.brendangregg.com/flamegraphs.html)。
在 Flink 中，可以通过 Web UI 访问操作员的火焰图，方法是选择运算符，然后选择**FlameGraph**选项卡。收集到足够的样本后，将显示火焰图。以下是 FlameGraph ProcessFunction 检查点花了很多时间的。
![\[使用火焰图\]](http://docs.aws.amazon.com/zh_cn/managed-flink/latest/java/images/checkpoint3.png)
这是一个非常简单的火焰图，显示所有的 CPU 时间都花在 ExpensiveFunction 操作员内部的 foreach `processElement` 视图中。您还可以获得行号，以帮助确定代码执行的哪个位置。
# 检查点操作已超时
如果您的应用程序未经过优化或未正确配置，则检查点可能会失败。本节介绍这种情况的症状和故障排除步骤。
## 症状
如果应用程序的检查点失败，则`numberOfFailedCheckpoints`将大于零。
检查点失败可能是由于直接故障（例如应用程序错误）或暂时性故障（例如应用程序资源耗尽）所致。检查您的应用程序日志和指标是否存在以下症状：
+ 您的代码中存在错误。
+ 访问应用程序的依赖服务时出错。
+ 序列化数据时出错。如果默认序列化程序无法序列化您的应用程序数据，则应用程序将失败。有关在应用程序中使用自定义序列化器的信息，请参阅 Apache Flink 文档中的[数据类型和序列化](https://nightlies.apache.org/flink/flink-docs-release-1.19/docs/dev/datastream/fault-tolerance/serialization/types_serialization/)。
+ 内存不足错误。
+ 以下指标出现峰值或稳定增长：
+ `heapMemoryUtilization`
+ `oldGenerationGCTime`
+ `oldGenerationGCCount`
+ `lastCheckpointSize`
+ `lastCheckpointDuration`
有关监控检查点的更多信息，请参阅 Apache Flink 文档中的[监控检查点](https://nightlies.apache.org/flink/flink-docs-release-1.19/docs/ops/monitoring/checkpoint_monitoring/)。
## 原因和解决方案
您的应用程序日志错误消息显示了直接失败的原因。暂时性故障可能有以下原因：
+ 您的应用程序的 KPU 配置不足。有关增加应用程序预置的信息，请参阅[实施应用程序扩展](how-scaling.md)。
+ 您的应用程序状态大小太大。您可以使用该`lastCheckpointSize`指标监控应用程序的状态大小。
+ 您的应用程序的状态数据在密钥之间的分布不均衡。如果您的应用程序使用`KeyBy`运算符，请确保传入的数据在密钥之间平均分配。如果将大部分数据分配给单个密钥，则会造成瓶颈，从而导致故障。
+ 您的应用程序面临内存或垃圾收集背压。监控应用程序的`heapMemoryUtilization`、`oldGenerationGCTime`、和`oldGenerationGCCount`是否出现峰值或稳步增加的值。
# Apache Beam 应用程序出现检查点故障
如果您的 Beam 应用程序配置为[shutdownSourcesAfterIdleMs](https://beam.apache.org/documentation/runners/flink/#:~:text=shutdownSourcesAfterIdleMs)设置为 0ms，则检查点可能无法触发，因为任务处于 “已完成” 状态。本节介绍这种情况的症状和解决方法。
## 症状
前往您的 Apache Flink 应用程序 CloudWatch 日志托管服务，检查是否记录了以下日志消息。以下日志消息表明，由于某些任务已经完成，检查点未能触发。
```
{
"locationInformation": "org.apache.flink.runtime.checkpoint.CheckpointCoordinator.onTriggerFailure(CheckpointCoordinator.java:888)",
"logger": "org.apache.flink.runtime.checkpoint.CheckpointCoordinator",
"message": "Failed to trigger checkpoint for job your job ID since some tasks of job your job ID has been finished, abort the checkpoint Failure reason: Not all required tasks are currently running.",
"threadName": "Checkpoint Timer",
"applicationARN": your application ARN,
"applicationVersionId": "5",
"messageSchemaVersion": "1",
"messageType": "INFO"
}
```
这也可以在 Flink 控制面板上找到，其中一些任务已进入 “已完成” 状态，并且无法再进行检查点操作了。
![\[处于 “已完成” 状态的任务\]](http://docs.aws.amazon.com/zh_cn/managed-flink/latest/java/images/beam_checkpoint_failure.png)
## 原因
shutdownSourcesAfterIdleMs 是一个 Beam 配置变量，用于关闭在配置的毫秒时间内处于空闲状态的信号源。一旦源被关闭，就无法再进行检查点检查了。这可能导致[检查点失败](https://issues.apache.org/jira/browse/FLINK-2491)。
任务进入 “已完成” 状态的原因之一 shutdownSourcesAfterIdleMs 是设置为 0 毫秒，这意味着空闲的任务将立即关闭。
## 解决方案
要防止任务立即进入 “已完成” 状态，请 shutdownSourcesAfterIdleMs 将其设置为 long.max\$1Value。这可以通过两种方式完成：
+ 选项 1：如果在 Apache Flink 托管服务应用程序配置页面中设置了 beam 配置，则可以添加新的密钥值对来设置 shutdpwnSourcesAfteridle Ms，如下所示：
![\[设置为 long. shutdownSourcesAfter IdleMs max_Value\]](http://docs.aws.amazon.com/zh_cn/managed-flink/latest/java/images/beam_checkpoint_failure_solution.png)
+ 选项 2：如果在 JAR 文件中设置了光束配置，则可以 shutdownSourcesAfterIdleMs 按以下方式进行设置：
```
FlinkPipelineOptions options = PipelineOptionsFactory.create().as(FlinkPipelineOptions.class); // Initialize Beam Options object
options.setShutdownSourcesAfterIdleMs(Long.MAX_VALUE); // set shutdownSourcesAfterIdleMs to Long.MAX_VALUE
options.setRunner(FlinkRunner.class);
Pipeline p = Pipeline.create(options); // attach specified options to Beam pipeline
```
# 背压
Flink 使用背压来调整各个操作员的处理速度。
出于多种原因，操作员可能难以继续处理收到的消息量。该操作需要的 CPU 资源可能超过操作员的可用资源，操作员可能会等待 I/O 操作完成。如果操作员无法足够快地处理事件，则会在向慢速运算符馈送的上游操作员中产生背压。这会导致上游操作员减速，从而进一步将背压传播到源，并通过减慢速度使源系统适应应用程序的总体吞吐量。你可以在[ Apache Flink™](https://www.ververica.com/blog/how-flink-handles-backpressure) 如何处理背压中找到对背压及其工作原理的更深入的描述。
了解应用程序中哪些运算符运行缓慢，可以为你提供重要信息，帮助你了解应用程序中性能问题的根本原因。背压信息[通过 Flink 控制面板公开](https://nightlies.apache.org/flink/flink-docs-stable/docs/ops/monitoring/back_pressure/)。要识别慢速操作员，请寻找背压值较高且最接近水槽的操作员（以下示例中的操作员 B）。因此，导致缓慢的运算符就是下游运算符之一（示例中为运算符 C）。B 可以更快地处理事件，但由于无法将输出转发给实际的慢速运算符 C，因此会受到反压
```
A (backpressured 93%) -> B (backpressured 85%) -> C (backpressured 11%) -> D (backpressured 0%)
```
识别出慢速运算符后，请尝试了解为什么它很慢。可能有多种原因，有时还不清楚出了什么问题，可能需要数天的调试和分析才能解决。以下是一些显而易见且更常见的原因，其中一些原因将在下面进一步解释：
+ 操作员正在执行缓慢的 I/O，例如网络调用（考虑改用 AsyncIO）。
+ 数据存在偏差，一个操作员收到的事件比其他操作员多（通过查看 Flink 控制面板中各个子任务（即同一运算符的实例）的消息 in/out 数量进行验证。
+ 这是一项资源密集型操作（如果没有数据偏差，可以考虑向外扩展以进行 CPU/memory 绑定工作，或者为绑定工作增加`ParallelismPerKPU`规模） I/O
+ 在操作员中进行大量日志记录（将生产应用程序的日志记录减少到最低限度，或者考虑改为将调试输出发送到数据流）。
## 使用丢弃接收器测试吞吐量
[丢弃接收器](https://nightlies.apache.org/flink/flink-docs-stable/api/java/org/apache/flink/streaming/api/functions/sink/DiscardingSink.html)只是忽略它在执行应用程序时收到的所有事件（没有任何接收器的应用程序无法执行）。这对于吞吐量测试、性能分析以及验证应用程序是否正常扩展非常有用。这也是一项非常实用的健全性检查，用于验证水槽是否造成了背压或应用（但是仅检查背压指标通常更容易、更简单）。
通过用丢弃的接收器替换应用程序的所有接收器，并创建一个生成类似于生产数据的数据的模拟源，您可以测量应用程序在特定并行度设置下的最大吞吐量。然后，您还可以增加并行度，以验证应用程序是否可以正常扩展，并且不会出现只有在更高的吞吐量下才会出现的瓶颈（例如，由于数据倾斜）。
# 数据偏斜
Flink 应用程序以分布式方式在集群上执行。为了扩展到多个节点，Flink 使用了密钥流的概念，这本质上意味着流的事件根据特定的密钥（例如客户 ID）进行分区，然后 Flink 可以在不同节点上处理不同的分区。然后，会根据这些分区对许多 Flink 运算符进行评估，例如 [Keyed Windows](https://nightlies.apache.org/flink/flink-docs-stable/docs/dev/datastream/operators/windows/)、[Process Functions](https://nightlies.apache.org/flink/flink-docs-stable/docs/dev/datastream/operators/process_function/) 和 [Async I/O](https://nightlies.apache.org/flink/flink-docs-stable/docs/dev/datastream/operators/asyncio/)。
选择分区键通常取决于业务逻辑。同时，D [ynamoDB ](https://aws.amazon.com/dynamodb/)和 Spark 等许多最佳实践同样适用于 Flink，包括：
+ 确保分区键的高基数
+ 避免分区之间的事件量出现偏差
您可以通过比较 Flink 控制面板中子任务（即同一运算符 received/sent 的实例）的记录来识别分区中的偏差。此外，还可将 Managed Service for Apache Flink 监控配置为公开子任务级别`numRecordsIn/Out`和`numRecordsInPerSecond/OutPerSecond`子任务级别的指标。
# 状态偏斜
对于有状态的运算符，即为其业务逻辑（例如窗口）保持状态的运算符，数据倾斜总是会导致状态偏差。由于数据偏差，某些子任务比其他子任务接收更多的事件，因此还会将更多的数据保留在状态中。但是，即使对于具有均衡分区的应用程序，在状态下保留的数据量也可能存在偏差。例如，对于会话窗口，某些用户和会话分别可能比其他用户和会话长得多。如果较长的会话恰好属于同一个分区，则可能导致同一操作员的不同子任务所保持的状态大小不平衡。
状态偏差不仅会增加单个子任务所需的更多内存和磁盘资源，还会降低应用程序的整体性能。当应用程序使用检查点或保存点时，操作员状态会保留到 Amazon S3 中，以保护该状态免受节点或集群故障的影响。在此过程中（尤其是在 Apache Flink 托管服务上默认启用一次语义的情况下），从外部角度来看，处理会停止，直到完成。 checkpoint/savepoint 如果存在数据偏差，则完成操作的时间可能受累积了特别多的状态的单个子任务的限制。在极端情况下，由于单个子任务无法保持状态，拍摄 checkpoints/savepoints 可能会失败。
与数据倾斜类似，状态偏差会大大降低应用程序的速度。
要识别状态偏差，您可以利用 Flink 控制面板。查找最近的检查点或保存点，并在详细信息中比较为各个子任务存储的数据量。
# 集成不同区域中的资源
通过在 Flink 配置中`StreamingFileSink`进行跨区域复制所需的设置，您可以启用使用写入与 Managed Service for Apache Flink 应用程序不同区域的 Amazon S3 存储桶。为此，请向[AWS 支持 中心提交支持请求](https://console.aws.amazon.com/support/home#/)。