问题标签 [checkpointing]

当火花流作业直接提交给火花时实施检查点似乎很简单。当使用 Spark 作业服务器提交流式作业时，当我们需要相同的情况时，我们面临着相当多的复杂性。任何指针/引用都可能很有帮助。有人可以帮忙吗？

当从检查点恢复失败的作业时，应用程序逻辑被正确调用并且 RDD 被重新实例化，但是对 RDD.map 的调用会导致 NullPointerException。

编辑：更新通知我正在使用 Kinesis 并且 WAL 已启用。S3 是否支持 WAL 检查点？我正在阅读其他没有得到很好支持的地方。 https://issues.apache.org/jira/browse/SPARK-9215

编辑：我在 HDFS 上遇到了类似的结果。

我有一个问题要让我的代码更自愈。例如：我执行方法 1 将数据从 CSV 加载到 Vertica 数据库中。我有另一种方法 2 来检查数据库中的行数和 CSV 文件中的行数是否相同。如果行数不匹配，那么我正在考虑从调用查询的位置调用方法 2，以将数据从 CSV 加载到数据库中。

我正在考虑针对此问题的检查点策略。例如，在代码中维护通常发生错误的一些点并在其他点调用它们。

我已经尝试在 python 中使用 pickle 模块，但后来知道 pickle 只能保存对象、类、变量等。不能保存我实际执行方法的点。

我提供了一些演示代码：

我希望我的代码能够执行，以便当 mango 方法调用 apple 方法时，它必须从仅打印“mapple”的点开始执行。它不能执行整个方法。

请给我一些关于如何解决这个问题的见解。

提前致谢

我正在尝试导出“模型”的权重和偏差，其中我最初没有使用“export_weights_and_biases = TRUE”训练模型

因此，我想尝试检查模型并尝试在新的“model2”中导出_weights_and_biases = TRUE。

然而，尽管没有更改任何参数 - 并确保nfolds=10就像在原始“模型”中一样，检查点模型几乎立即继续返回参数更改错误（h2o 版本 3.10.4.6）：

任何想法如何解决这个问题？我要做的就是获得我原来的“模型”的所有权重和偏差。

谢谢！

在我的 spark 应用程序中，我正在读取 spark rdd 中的几个 hive 表，然后稍后对这些 rdds 执行一些转换。rdd.cache()为了避免重新计算，我使用orrdd.persist()和方法缓存了这些 rdds rdd.checkpoint()。

根据 spark 文档和在线参考资料，我认为检查点操作比缓存更昂贵。尽管缓存保持 rdd 沿袭，并且检查点会破坏它，但检查点会从 HDFS 写入和读取。

在我的案例中观察到的奇怪的事情是，我看到检查点阶段比缓存/持久化（仅内存）更快（近 2 倍）。我跑了多次，结果仍然相似。

我无法理解为什么会这样。任何帮助都会有所帮助。

我已经尝试过Apache Spark (Structured Streaming) 中给出的建议：S3 Checkpoint support

我仍然面临这个问题。以下是我得到的错误

我有这样的东西作为我的代码的一部分

然后像这样使用检查点目录：

任何帮助表示赞赏。提前致谢！

Spark 2.2.0 检查点的工作方式与版本略有不同。有一个被创建的提交文件夹，在每批完成后，一个文件被写入该文件夹。

我面临这样一种情况，我有大约 10k 条记录，并且说我的流式传输作业在处理 5k 条记录后大约在中间失败，没有文件写入检查点目录中的提交文件夹，因此当我重新启动作业时，它从头开始，有 5k 条记录的数据重复。

据我了解，当有提交文件被写入时，当你重新启动它时，它会从最新的偏移量中获取，否则它会重新处理最后一个成功批处理中的数据。

如何设置批量处理的单据数量？

我正面临火花流作业的问题，我试图在火花中同时使用广播、mapWithState和检查点。

以下是用法：

• 由于我必须将一些连接对象（不可序列化）传递给执行程序，因此我使用的是org.apache.spark.broadcast.Broadcast
• 由于我们必须维护一些缓存信息，因此我使用带有 mapWithState 的状态流
• 我也在使用我的流上下文的检查点

我还需要将广播的连接对象传递到 mapWithState 中，以便从外部源获取一些数据。

当新创建上下文时，流程工作得很好。但是，当我使应用程序崩溃并尝试从检查点恢复时，我得到了 ClassCastException。

我根据asyncified.io 中的示例放置了一个小代码片段，以在github中重现该问题：

• 我的广播逻辑是yuvalitzchakov.utils.KafkaWriter.scala
• 应用程序的虚拟逻辑是yuvalitzchakov.stateful.SparkStatefulRunnerWithBroadcast.scala

代码的虚拟片段：

我收到以下错误时

kafkaWriter.value.someMethodCall()

被执行：

基本上kafkaWriter是广播变量，kafkaWriter.value应该返回我们广播的变量，但它返回的SerializableConiguration没有被强制转换为所需的对象

提前感谢您的帮助！

在独立 Spark 集群上部署我的 Spark Streaming 作业后，我遇到了一些检查点问题。控制台日志产生一个提示：

我正在使用默认的 HashPartitioner 将数据划分为两个分区。我将我的 HDFS 检查点目录设置为我的 Spark 主服务器和 HDFS 端口，如下所示：

在我的工作中，我自己从不手动调用.checkpoint(duration)任何 DStream。但是我从PairDStreams mapWithState()调用中得到了许多有状态的流。捕获异常的代码可以在ReliableCheckpointRDD 行 209ff中找到。不幸的是，我在网上找不到有关此错误的任何参考。

在我的工作中，每当触发检查点时，每个有状态的 DStream 都会引发异常。

任何帮助表示赞赏！

编辑#1

这不影响结果的正确性。然而，我想知道在我进行一些性能分析时性能是否会变得更糟。

我启用了将日志保存到 S3 的检查点。如果检查点目录中没有文件，火花流工作正常，我可以看到检查点目录中出现日志文件。然后我杀死火花流并重新启动它。这一次，我开始为 Spark 会话获取 NullPointerException。简而言之，如果检查点目录中没有日志文件，则火花流可以正常工作。但是，一旦我在检查点目录中重新启动带有日志文件的火花流式传输，我就会开始在火花会话上收到空指针异常。下面是代码：

再一次，我第一次运行这段代码时（检查点目录中没有日志文件），我可以看到数据框被打印出来。如果我在检查点目录中运行日志文件，我什至看不到

打印出来，第一次打印出来。错误：

错误发生在：

编辑：我添加了这一行：

它仍然没有任何区别，仍然得到 NullPointerException。