问题标签 [checkpointing]

我们目前正在开发一个使用 kafka、spark 流和 Cassandra 作为 DB 的系统。我们正在使用基于此处内容的检查点 [ http://spark.apache.org/docs/latest/streaming-programming-guide.html#checkpointing]。在用于创建 Streaming 上下文的函数中，我们使用 createDirectStream 创建我们的 DStream，从这一点开始，我们在不同的 RDD 上执行从调用 saveToCassandra 派生的几个转换和操作

我们正在运行不同的测试来确定应用程序在发生故障时应该如何恢复。关于我们的场景的一些关键点是：

• 我们在 kafka 中使用固定数量的记录（在 1000 万到 2000 万之间）进行测试，这意味着我们从 kafka 消费一次，应用程序从 kafka 中获取所有记录。

• 我们在其中一名工作人员中以 --deploy-mode 'client' 执行应用程序，这意味着我们手动停止和启动驱动程序。

我们不确定在创建 DStreams 后如何处理异常，例如，如果在写入 cassandra 时所有节点都死了，我们会得到一个异常中止作业，但在重新提交应用程序后，该作业不会重新调度并且应用程序不断消耗来自 kafka 的多个“isEmpty”调用。

我们在重新分区的 RDD 上使用“缓存”进行了几次测试（在失败后与停止和启动驱动程序不同），并更改参数“query.retry.count”、“query.retry”。延迟”和“spark.task.maxFailures”没有成功，例如，作业在 x 次失败后中止。

在这一点上，我们对如何使用检查点在失败后重新安排作业感到困惑。

检查点对 Apache Spark 有什么作用，它是否会对 RAM 或 CPU 造成任何影响？

我正在阅读有关检查点的信息。根据我现在阅读的内容，有两个主要的检查点：

• 系统级检查点 (SLC) – core-dump 风格的计算快照

• 应用程序级检查点（ALC）——程序是自检查点和自重启的

我有兴趣在 C 中实现函数级别的检查点恢复算法。我想知道是否可以在“应用程序级”类别中考虑这一点。

其次，是否有可用的开源库。

我在这里举一个简单的添加函数作为例子：

添加（int a0，int a1，int b0，int b1，int * res0，int * res1）

算法策略如下：

有没有办法以一般格式编写“存储”部分（第一行）。如果函数获得不同的参数类型怎么样。

我在有效使用变量范围时遇到问题。我想为简单循环网络的权重、偏差和内部状态定义一些变量。我在定义默认图形后调用get_saver()一次。然后，我使用tf.scan.

我希望从操作get_variable内部检索到的变量与调用scan内部定义的变量相同get_saver。但是，如果我运行此示例代码，我会得到以下错误输出：

知道我在这个例子中做错了什么吗？

我有一个简单的循环网络示例，其中保存了andtf.Saver和weight变量。biasstate

当示例在没有选项的情况下运行时，它将初始化状态向量以包含零，但我想传递一个load_model选项并将其使用状态向量的最后一个值作为session.run调用的提要。

我看到的所有文档都坚持必须调用session.run以从变量中检索存储的值，但在这种情况下，我想检索这些值，以便我可以初始化状态变量。我需要做一个单独的图表来检索初始化值吗？

下面的示例代码：

请注意第 124-126 行的注释行，说明我尝试初始化提要字典值的方式。它们都不起作用。

我正在运行一个从 Kafka 读取数据的 Spark Streaming 应用程序。我已激活检查点以在发生故障时恢复作业。

问题是，如果应用程序失败，当它重新启动时，它会尝试仅在一个微批次中执行故障点的所有数据。 这意味着，如果一个微批处理通常从 Kafka 接收 10.000 个事件，如果它失败并在 10 分钟后重新启动，它将不得不处理一个包含 100.000 个事件的微批处理。

现在，如果我希望通过检查点恢复成功，我必须分配比正常情况更多的内存。

重新启动时，Spark Streaming 尝试从检查点一次执行所有过去的事件是否正常，还是我做错了什么？

非常感谢。

• 目标：通过 Spark 流从 Kinesis 读取数据并以 Parquet 格式将数据存储到 S3。
• 情况：应用程序最初运行良好，运行批处理 1 小时，处理时间平均不到 30 分钟。出于某种原因，假设应用程序崩溃了，我们尝试从检查点重新启动。现在处理需要永远并且不会向前推进。我们尝试以 1 分钟的批处理间隔测试相同的东西，处理运行良好，批处理完成需要 1.2 分钟。当我们从检查点恢复时，每批大约需要 15 分钟。
• 注意：我们使用 s3 作为检查点，使用 1 个执行器，每个执行器有 19g 内存和 3 个核心

附上截图：

首次运行 - 检查点恢复之前

试图从检查点恢复：

配置文件

S3Basin.scala

Kinesis.scala

有向无环图

我想将检查点测试保存在亚马逊 S3 上的某个位置，这是我在 DStream 上的 scala 代码的一部分，使用以下格式但出现错误..

线程“main”java.lang.IllegalArgumentException 中的异常：AWS 访问密钥 ID 和秘密访问密钥必须分别指定为 s3n URL 的用户名或密码，或者通过设置 fs.s3n.awsAccessKeyId 或 fs.s3n。 awsSecretAccessKey 属性（分别）。

代码：

我已经编写了一个 Spark Streaming 应用程序，它需要对具有底层转换的各种 Dstream 进行一些检查，如该线程中所建议的（启动 Spark 流式传输上下文时出错），我已经在定义的函数中完成了所有转换以创建上下文,

但是，当我的上下文从检查点目录加载时，我仍然会遇到异常

谁能帮我在我的代码中解决这个问题，我在 Spark 1.6 和 2.0 中都尝试过我的代码，但得到了同样的异常。

我创建了一个如下的RDD：

我的问题是如何从检查点目录中读取数据