蓄电池何时真正可靠？

Question

我想使用累加器来收集有关我在 Spark 作业中处理的数据的一些统计信息。理想情况下，我会在作业计算所需的转换时这样做，但由于 Spark 会在不同情况下重新计算任务，因此累加器不会反映真实的指标。以下是文档对此的描述：

对于仅在操作内部执行的累加器更新，Spark 保证每个任务对累加器的更新只会 应用一次，即重新启动的任务不会更新该值。在 转换，用户应该知道每个任务的更新可能 如果重新执行任务或作业阶段，则应用不止一次。

这很令人困惑，因为大多数 动作 不允许运行自定义代码（可以使用累加器），它们主要从以前的转换中获取结果（懒惰地）。该文档还显示了这一点：

val acc = sc.accumulator(0)
data.map(x => acc += x; f(x))
// Here, acc is still 0 because no actions have cause the `map` to be computed.

但是如果我们在末尾添加data.count()，这是否可以保证正确（没有重复）？显然acc 不是“仅用于内部操作”，因为 map 是一种转换。所以不能保证。

另一方面，有关 Jira 票证的讨论谈论的是“结果任务”而不是“行动”。例如here 和here。这似乎表明结果确实可以保证是正确的，因为我们在和 action 之前立即使用acc，因此应该作为单个阶段计算。

我猜这个“结果任务”的概念与所涉及的操作类型有关，它是最后一个包含操作的操作，就像在这个例子中一样，它显示了几个操作是如何分成阶段的（洋红色，图片取自here)：

所以假设，该链末端的count() 动作将是同一最终阶段的一部分，并且我可以保证最后一张地图上使用的累加器不会包含任何重复项？

澄清这个问题会很棒！谢谢。

Answer 1

回答“蓄电池什么时候真正可靠？”

答案：当它们出现在 Action 操作中时。

根据 Action Task 中的文档，即使存在任何重新启动的任务，它也只会更新 Accumulator 一次。

对于仅在操作内部执行的累加器更新，Spark 保证每个任务对累加器的更新只会应用一次，即重新启动的任务不会更新值。在转换中，用户应注意，如果重新执行任务或作业阶段，每个任务的更新可能会应用多次。

Action 确实允许运行自定义代码。

例如

val accNotEmpty = sc.accumulator(0)
ip.foreach(x=>{
  if(x!=""){
    accNotEmpty += 1
  }
})

但是，为什么地图+动作，即。结果任务操作对于累加器操作不可靠？

1. 由于代码中的一些异常，任务失败。 Spark 将尝试 4 次（默认尝试次数）。如果任务每次失败都会给出异常。如果偶然成功，则 Spark 将继续并更新成功状态的累加器值，而失败状态的累加器值将被忽略。判决：妥善处理
2. Stage Failure ：如果执行器节点崩溃，则不是用户故障，而是硬件故障 - 如果节点在 shuffle 阶段出现故障。由于 shuffle 输出存储在本地，如果节点出现故障，则该 shuffle 输出消失。所以 Spark 回到生成 shuffle 输出的阶段，查看哪些任务需要重新运行，并在其中一个还活着的节点上执行它们。在我们重新生成丢失的 shuffle 输出之后，生成 map 输出的阶段已多次执行其中的一些任务。Spark 计算所有这些任务的累加器更新。
   结论：未在结果任务中处理。累加器将给出错误的输出。
3. 如果某个任务运行缓慢，Spark 可以在另一个节点上启动该任务的推测副本。
   结论：未处理。累加器将给出错误的输出。
4. 缓存的 RDD 很大，不能驻留在内存中。因此，无论何时使用 RDD，它都会重新运行 Map 操作以获取 RDD，并再次更新累加器。
   结论：未处理.Accumulator 会给出错误的输出。

因此可能会发生相同的函数可能会在相同的数据上运行多次。因此 Spark 不提供任何保证累加器因为 Map 操作而得到更新的保证。

所以在 Spark 中最好使用 Accumulator in Action 操作。

要了解有关 Accumulator 及其问题的更多信息，请参阅 Blog Post - 作者：Imran Rashid。

Answer 2

累加器更新会在任务成功完成后发送回驱动程序。因此，当您确定每个任务都将执行一次并且每个任务都按照您的预期执行时，您的累加器结果就可以保证是正确的。

我更喜欢依赖 reduce 和 aggregate 而不是累加器，因为很难列举所有可以执行任务的方式。

• 动作启动任务。
• 如果某个操作依赖于较早的阶段，并且该阶段的结果未（完全）缓存，则将启动较早阶段的任务。
• 检测到少量慢速任务时，推测执行会启动重复任务。

也就是说，有很多简单的情况下可以完全信任累加器。

val acc = sc.accumulator(0)
val rdd = sc.parallelize(1 to 10, 2)
val accumulating = rdd.map { x => acc += 1; x }
accumulating.count
assert(acc == 10)

这是否可以保证是正确的（没有重复）？

是的，如果推测执行被禁用。 map 和 count 将是一个阶段，所以就像你说的，一个任务不可能多次成功执行。

但累加器会作为副作用更新。因此，在考虑如何执行代码时，您必须非常小心。考虑这个而不是accumulating.count：

// Same setup as before.
accumulating.mapPartitions(p => Iterator(p.next)).collect
assert(acc == 2)

这也会为每个分区创建一个任务，并且每个任务将保证只执行一次。但是map 中的代码不会在所有元素上执行，只会在每个分区中的第一个元素上执行。

累加器就像一个全局变量。如果您共享对可以递增累加器的 RDD 的引用，那么其他代码（其他线程）也可以导致累加器递增。

// Same setup as before.
val x = new X(accumulating) // We don't know what X does.
                            // It may trigger the calculation
                            // any number of times.
accumulating.count
assert(acc >= 10)

Answer 3

我认为 Matei 在参考文档中回答了这个问题：

正如在https://github.com/apache/spark/pull/2524 上讨论的那样，这是 在一般情况下很难提供良好的语义 （非结果阶段内的累加器更新），用于以下 原因：

• 一个 RDD 可以作为多个阶段的一部分进行计算。为了 例如，如果您更新 MappedRDD 中的累加器，然后 洗牌，那可能是一个阶段。但是如果你再调用 map() 在 MappedRDD 上，然后随机播放结果，你会得到第二个 该地图是管道的阶段。你要数这个吗 累加器是否更新两次？

• 如果出现以下情况，可以重新提交整个阶段 shuffle 文件被定期清理器删除或由于 节点故障，因此任何跟踪 RDD 的东西都需要这样做 很长一段时间（只要 RDD 在用户中是可引用的） 程序），这将是相当复杂的实现。

所以我要走了 暂时将其标记为“不会修复”，结果部分除外 在 SPARK-3628 中完成的阶段。