标签: checkpointing

我正在“本地”模式下运行 Spark 应用程序。它正确检查点到 checkpointFolder 配置中定义的目录。但是，我发现有两个问题导致了一些磁盘空间问题。

1）由于我们有多个用户运行该应用程序，服务器上的检查点文件夹是由第一个执行该应用程序的用户创建的，这会导致其他用户由于操作系统权限问题而运行失败。有没有办法在 checkpointFolder 中提供相对路径，例如 checkpointFolder=~/spark/checkpoint？

2）我已使用spark.worker.cleanup.enabled=true配置在运行后清理检查点文件夹，但没有看到这种情况发生。是否有其他方法可以通过应用程序清理它，而不是诉诸某些 cron 作业？

我习惯ModelCheckpoint在每个时期每 500 个批次保存一次检查点。它记录在此处https://www.tensorflow.org/api_docs/python/tf/keras/callbacks/ModelCheckpoint。

我该如何设置filepath以包含批次号？我知道我可以在 中使用{epoch}和 参数logs。

我是Apache Flink的新手，并且正在研究Apache Flink的示例。我发现，如果发生故障，Flink可以从检查点恢复流处理。

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.enableCheckpointing(10000L);

现在，我的问题是Flink在默认情况下将检查点保留在哪里？

任何帮助表示赞赏！

我正在查看Cloudbees 文档，该文档说：

正确的方法是始终将检查点步骤保持在任何节点块之外，不与代理或工作区相关联

给出的示例示例适用于脚本化管道。我试图在声明式管道中实现这一点，但不断出错。我可以让它工作的唯一方法是：

stage ('Promotion Checkpoint') {
    steps {
        checkpoint 'Ready for Manual intervention'
        timeout(time: 60, unit: 'SECONDS') {
            input message: 'Do you want to proceed?'
        }
    }
}

我的理解是声明式管道中的一个阶段类似于脚本管道中的节点。我不能在阶段或步骤之外进行检查点工作，这似乎是我对 Cloudbees 建议的解释。有人可以帮助在检查站之外正确使用吗？

我有一个WordCount用 Flink (Scala) 编写的示例流示例。在其中，我想使用外部化检查点来在发生故障时进行恢复。但它没有按预期工作。

我的代码如下：

object WordCount {
  def main(args: Array[String]) {
    // set up the execution environment
    val env = StreamExecutionEnvironment
      .getExecutionEnvironment
      .setStateBackend(new RocksDBStateBackend("file:///path/to/checkpoint", true))

    // start a checkpoint every 1000 ms
    env.enableCheckpointing(1000)

    // set mode to exactly-once (this is the default)
    env.getCheckpointConfig.setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE)

    // make sure 500 ms of progress happen between checkpoints
    env.getCheckpointConfig.setMinPauseBetweenCheckpoints(500)

    // checkpoints have to complete within one minute, or are discarded
    env.getCheckpointConfig.setCheckpointTimeout(60000)

    // prevent the tasks from failing if an error happens in …

当我尝试加载 pytorch 检查点时：

checkpoint = torch.load(pathname)

我懂了：

RuntimeError: cuda runtime error (35) : CUDA 驱动程序版本对于 Torch/csrc/cuda/Module.cpp:51 中的 CUDA 运行时版本不足

我用可用的 GPU 创建了检查点，但现在只有 CPU 可用。

如何加载检查点？

Background

• I was planning to use S3 to store the Flink's checkpoints using the FsStateBackend. But somehow I was getting the following error.

Error

org.apache.flink.core.fs.UnsupportedFileSystemSchemeException: Could not find a file system implementation for scheme 's3'. The scheme is not directly supported by Flink and no Hadoop file system to support this scheme could be loaded.

Flink version: I am using Flink 1.10.0 version.