问题标签 [checkpointing]

我正在尝试在 JAVA 中查找自定义检查点管理器的示例，它可以将检查点数据存储在本地文件夹中。

基本上，我正在构建一个 java 应用程序，它从具有多个消费者组的 azure 事件中心读取数据。以前，我使用存储帐户连接字符串和位于 azure blobs 中的存储容器来实例化 EventProcessorHost - 这工作正常。

POM 条目：

用于实例化主机的 Java 代码：

现在，要求是使用 Azure Databricks 群集中的本地文件夹（DBFS:/ 路径）来存储检查点数据。

我想我必须编写一个自定义检查点管理器来实现 ICheckpointManager。我能够在 SQL 数据库中找到执行此操作的示例，但我无法找到 CheckpointManager 将检查点数据存储在本地文件夹中的示例。

谁能帮忙，给我一个例子的链接或代码片段？

这篇文章来自一篇有趣的文章：http ://www.lifeisafile.com/Apache-Spark-Caching-Vs-Checkpointing/

" ... Checkpointing 将 rdd 物理存储到 hdfs 并销毁创建它的 lineage。即使在 Spark 应用程序终止后，检查点文件也不会被删除。检查点文件可用于后续作业运行或驱动程序。检查点 RDD导致双重计算，因为该操作将在执行计算和写入检查点目录的实际工作之前首先调用缓存。..."

我似乎记得在其他地方读过检查点文件仅适用于给定 Spark 应用程序中的作业或共享作业。

寻找澄清以及新应用程序如何使用检查点目录，因为我认为这是不可能的。

我正在使用 Apache Flink 设置分析管道来处理 IoT 数据流。在尝试配置系统时，我似乎找不到任何关于应该多久启动一次检查点的来源？是否有任何建议或严格的经验法则？例如 1 秒、10 秒、1 分钟等？

编辑：另外，有没有办法在运行时以编程方式配置检查点间隔？

我ModelCheckpoint用来在每个时代每 500 个批次保存检查点。它记录在这里https://www.tensorflow.org/api_docs/python/tf/keras/callbacks/ModelCheckpoint。

我将如何设置filepath以包含批号？我知道我可以{epoch}在logs.

我在一些代码示例中看到，人们使用 .pwf 作为模型文件保存格式。但在 PyTorch 文档中，推荐使用 .pt 和 .pth。我使用 .pwf 并在小型 1->16->16 卷积网络上运行良好。

我的问题是这些格式有什么区别？为什么在 PyTorch 文档中甚至不推荐 .pwf 扩展名，为什么人们仍然使用它？

要在崩溃后恢复训练，不仅必须恢复模型，还必须恢复进入model.fit(...)进程状态的所有对象和参数。

在我费心去分叉keras代码来实现一个fitting对象之前，例如训练数据，我想知道标准方法（如果有的话）是什么，用于崩溃恢复以从中断的地方恢复 TensorFlow 2.0 训练。

还是有人真的填补了 TensorFlow 对象模型中这个明显的漏洞？

我正在开发一个下沉到 Kafka 的 Flink 应用程序。我创建了一个默认池大小为 5 的 Kafka 生产者。我使用以下配置启用了检查点：

该应用程序有时会继续崩溃，并出现以下异常。这个问题与 kafka 生产者池大小或检查点有关吗？

我们通过FlinkRunner运行的Apache Beam作业之一遇到了检查点大小的奇怪行为。状态后端是基于文件的。该作业每天接收一次流量，持续一个小时，然后处于空闲状态，直到它接收到更多数据。

管道不使用任何窗口策略。它只是从源中读取，组合来自该源的不同值并将其写入接收器。我们只在无界源中记录状态。

随着我们处理更多数据，它的大小会慢慢增加，但是，一旦停止使用数据，检查点的大小不会显着减小。

我们认为它可能是数据库接收器的瓶颈，但是如果我们移除接收器并简单地转储数据，则会出现相同的行为。

该行为似乎类似于阶梯图，例如

• 检查点 = 120KB（起始大小检查点）
• checkpoint = 409MB（开始接收数据）
• checkpoint = 850MB（处理积压数据）
• checkpoint = 503MB（完成处理数据）
• checkpoint = 1.2GB（开始处理新数据和积压）
• checkpoint = 700MB（完成处理数据）
• 检查点 = 700MB（检查点的新起始大小）
• ...

有没有人见过这种行为？这是使用 Apache Beam 进行 Flink 检查点的已知问题吗？

在我的场景中，我将有一批事件同时进入，然后很长一段时间 EventHub 将处于空闲状态。在我的处理器客户端中，我想每 N 个事件或 N 分钟检查一次（以先到者为准）。

这是我设置 Azure.Messaging.EventHubs.EventProcessorClient 的方式：

在我的 ProcessEventHandler 中，我检查 eventsProcessedSinceLastCheckpoint 以及秒表上经过的时间。当任何一个达到最大值时，我都会重置两者并在我的控制台窗口中记下它：

对 eventsProcessedSinceLastCheckpoint 变量的检查非常有效，因为只要有新事件进入，就会触发 ProcessEventHandler。但是，当 EventHub 空闲时，不会调用 ProcessEventHandler，因此，如果 EventHub 安静了几分钟或几小时，我将永远不会检查经过的时间。

我知道我可以删除计时器，并且如果检查点之间发生崩溃，我的处理器应该能够处理重复事件。但在我的场景中（因为我会有很长的空闲时间）我想利用我拥有的时间并赶上，以避免在可能的时候出现额外的重复。因此，在空闲期间添加计时器作为后备。

我的问题是：如何在ProcessEventHandler之外调用UpdateCheckpointAsync()？该方法似乎只存在于ProcessEventArgs上。我不能直接在 EventProcessorClient 上调用它，这将是理想的，因为我可以将计时器检查移到 ProcessEventHandler 之外并进入我的 while 循环......

我有以下带有检查点的蛇文件。我正在尝试运行 2 个样本（定义为 RUNS）。然而，每次我尝试我都会得到一个额外的变量。关于如何解决这个问题的任何想法？谢谢..

我收到以下错误：

谢谢你的指点！