本文属于机器翻译版本。若本译文内容与英语原文存在差异，则一律以英文原文为准。 # 运行时问题排查 本节包含有关诊断和修复 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#/)。