Apache Beam 应用程序的检查点失败 - Amazon Kinesis Data Analytics
症状原因解决方案
Amazon Web Services 文档中描述的 Amazon Web Services 服务或功能可能因区域而异。要查看适用于中国区域的差异，请参阅 中国的 Amazon Web Services 服务入门 (PDF)。

本文属于机器翻译版本。若本译文内容与英语原文存在差异，则一律以英文原文为准。

Apache Beam 应用程序的检查点失败

如果您的 Beam 应用程序配置为shutdownSourcesAfterIdleMs设置为 0ms，则检查点可能无法触发，因为任务处于 “完成” 状态。本节介绍这种情况的症状和解决方法。

症状

转到您的 Kinesis Data Analytics 应用程序 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 仪表板上找到，其中一些任务已进入 “完成” 状态，不再可能进行检查点了。

处于 “已完成” 状态的任务

原因

shutdownSourcesAfterIdleMs 是一个 Beam 配置变量，用于关闭在配置的毫秒时间内处于空闲状态的信号源。一旦信号源被关闭，就无法再进行检查点了。这可能导致检查点失败。

任务进入 “完成” 状态的原因之一 shutdownSourcesAfterIdleMs 是设置为 0 毫秒时，这意味着空闲的任务将立即关闭。

解决方案

要防止任务立即进入 “完成” 状态，请设置为 shutdownSourcesAfterIdleMs long.max_Value。可以通过以下两种方式执行此操作：

  • 选项 1：如果您的光束配置是在 Kinesis Data Analytics 应用程序配置页面中设置的，则可以添加新的键值对来设置 shutdpwnSourcesAfteridle Ms，如下所示：

    设置 shutdownSourcesAfterIdleMs 为 long.max_Value

  • 选项 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