问题标签 [checkpointing]

我使用以下代码在训练模型的循环之外创建了一个检查点管理器：

然后在训练模型时，我ckpt_save_path = ckpt_manager.save()会在每个 epoch 之后保存变量。

鉴于我想实现一种提前停止方法，我需要在特定时期之后恢复所有变量并使用这些变量进行预测。如果我使用上面的代码保存变量（希望保存过程是正确的？），那么在假设 epoch e 之后如何恢复变量。我知道我可以先创建相同的变量和对象，然后使用下面的代码来恢复最新的检查点，但不知道如何恢复特定的检查点（如 epoch number e 之后的变量）而不是最新的。

ckpt.restore(ckpt_manager.latest_checkpoint).assert_consumed()

谢谢，

我正在使用 flink 1.9 和 REST API/jobs/:jobid/savepoints来触发保存点并取消作业（优雅地停止作业以便稍后从保存点运行）。

我在源函数中使用了两阶段提交，所以我的源实现了CheckpointedFunction和CheckpointListener接口。在snapshotState()方法调用中，我将内部状态和notifyCheckpointComplete()检查点状态快照到第 3 方系统。

从源代码中我可以看到，只有snapshotState()部分是同步的CheckpointCoordinator-

检查点确认和完成通知在AsyncCheckpointRunnable.

话虽如此，当savepointwith cancel-jobset totrue被触发时，在拍摄快照后，一些任务管理器会在作业取消和执行之前继续接收完成通知notifyCheckpointComplete()，而有些则不会。

问题是是否有办法使用保存点取消作业，以便notifyCheckpointComplete()保证在作业取消之前由所有任务管理器调用，或者目前没有办法实现这一点？

我在 kubernetes 集群中使用带有作业管理器 pod 和两个任务管理器 pod 的 Flink 集群。当我将流式作业提交给作业管理器时，它会运行作业并将输出接收到接收器中。我还启用了检查点以从故障中恢复。现在，当我故意删除一个任务管理器 pod 以验证 flink 中的节点故障处理时，我看到一些假设到达接收器的记录没有收到。当 Kubernetes 自动重启 Pod 时，它会继续处理记录，但不会从检查点恢复。我正在使用以下命令提交作业

我在工作环境中有以下内容：

当任务管理器 pod 重新启动时，我有以下日志。

但是在检查点目录 2966c462794bf94523e9a53c1d9a2f13 我只有以下项目。

我在目录 2966c462794bf94523e9a53c1d9a2f13 中没有目录 p-421bdbdb924a09ddc017b854d52d9a9457109d43-7307729d1f0408de20cd29e352a2a655

根据文档，任务应该自动从检查点位置恢复。

请让我知道这可能是哪里的问题。

更新

进行了实际测试 -

以“t”秒间隔将记录连续插入到 flink 作业中。当任务管理器处理记录时，我杀死了一个任务管理器 pod。此时我停止将记录插入到 flink 作业中。在作业的输入端，我向其中插入了 1000 条记录。当任务管理器再次出现时，我在接收器中有 700 条记录。

现在我开始一次插入一条记录，发现 sink 中的记录突然增加到 940 条，然后开始增加 1，即任务管理器崩溃后插入的记录开始进入 sink。但是我从最初的 1000 条记录中丢失了 60 条记录，这些记录是在任务管理器崩溃之前插入的

背景

• 我打算使用 S3 来存储 Flink 的检查点，使用FsStateBackend. 但不知何故，我收到了以下错误。

错误

Flink 版本：我使用的是 Flink 1.10.0 版本。

如果我们在 EventHub conf 中指定起始位置，如下所示：

并且还要在 StreamWriter 中指定检查点位置

重新启动后，是否setStartingPosition因为检查点始终用作开始读取的点而变得无关紧要？

谢谢。

我有一个使用rocksdbjni 6.15.2 来管理和检查点状态的流管道。我正在尝试在单独的离线 Scala 进程中使用同一个库来读取检查点文件，并进行一些进一步的处理。

为了测试，我将其中一个检查点状态文件夹复制到本地文件夹：

然后运行这个简单的 Scala 程序：

并遇到此异常：org.rocksdb.RocksDBException: Corruption: IO error: No such file or directoryWhile open a file for random read: /mnt/tmp/rocksdb/010958.ldb: No such file or directory

有没有人遇到过这个异常？RocksDB 为什么要在这里找.ldb文件？

我正在尝试在 CPU 上运行类似 Resnet 的图像分类模型，并且想知道运行模型的每一层所需的时间细分。

我面临的问题是 github 链接https://github.com/facebookresearch/semi-supervised-ImageNet1K-models将模型保存为.pth文件。它非常大（100 MB），我不知道它与 pytorch 有什么不同，除了它是二进制的。我使用以下脚本从此文件加载模型。但是我看不到修改模型或t = time.time()在模型层之间插入变量/语句以分解每一层的时间的方法。

问题：

1. 在以下脚本中运行模型会正确估计在 CPU 上运行模型所需的端到端时间 (t2-t1)，还是还包括 pytorch 编译时间？

2. 如何在连续层之间插入时间语句以获得细分？

3. github 链接上没有推理/训练脚本，只有 .pth 文件。那么究竟应该如何进行推理或训练呢？如何在 .pth 模型的连续层之间插入额外的层并保存它们？

我启用了操作员检查点并为ProcessFunction操作员顺利工作。

在作业失败时，我可以看到操作员状态如何在snapshotState()挂钩上外部化，在恢复时，我可以看到状态如何在initializeState()挂钩上恢复。

但是，当我尝试CheckpointedFunction在 an 上实现接口和上述两种方法时AsyncFunction，它似乎不起作用。我所做的几乎与ProcessFunction...相同，但是当工作在失败后关闭时，它似乎并没有被snapshotState()钩住，并且在工作恢复时，context.isRestored()总是错误的。

为什么CheckpointedFunction.snapshotState()和CheckpointedFunction.initializeState()没有被执行AsyncFunction但是是的ProcessFunction？

编辑：由于某种原因，我的检查点需要很长时间。我相信我的配置非常标准，1 秒的间隔，500 毫秒的最小暂停，恰好一次。没有其他调音。
我从检查点协调员那里得到这些痕迹

如果我尝试设置 checkpointTimeout，我需要按顺序或 5 分钟左右设置一些内容。这么小的状态（它只是一个 Counter 和一个 Long）的检查点怎么需要 5 分钟？

我还读到 NFS 卷是一个麻烦的秘诀，但到目前为止我还没有在集群上运行它，我只是在我的本地文件系统上测试它

我们有一个以原生 K8S 部署模式运行的 Flink 流式作业 v1.14。当我们使用 AT_LEAST_ONCE 检查点模式时，无论我们分配多少内存，托管内存使用率都会达到 100%。任何想法可能是什么原因，或者这实际上是 Flink 如何管理内存的预期行为？

我正在开发一个大脚本，它的骨架如下所示：

1. 在每次起飞期间，首先在load_variables()函数中加载用户标志。
2. 然后脚本继续执行function_1() => function_2() => [...] => function_n()

我需要实现将存储在log.txt.
假设该脚本已在function_2().
我想在每个函数启动之前保存进度，将其存储在 中log.txt，当我再次重新运行脚本时，我想load_variables()然后跳转到存储在log.txt.

如何使用 bash 实现这一点？