执行顺序和缓存需求

Question

让我们考虑这样一个 python 伪代码的 sn-p，使用 spark。

    rdd1 = sc.textFile("...")
    rdd2 = rdd1.map().groupBy().filter()
    importantValue = rdd2.count()
    rdd3 = rdd1.map(lambda x : x / importantValue)

在spark的tasks的DAG中，有两个分支，在创建rdd1之后。两个分支都使用rdd1，但第二个分支（计算rdd3）也使用来自rdd2（importantValue）的聚合值。我假设DAG 看起来像这样： 我对吗？如果是，我们是否可以假设用于计算rdd3 的rdd1 仍在内存中处理？或者我们必须缓存rdd1 以防止重复加载？

更一般地说，如果DAG 看起来像这样： 我们可以假设两个分支都是并行计算的并使用rdd1 的相同副本吗？还是 Spark 驱动程序会一个接一个地计算这些分支，因为这是两个不同的阶段？我知道在执行之前 spark 驱动程序将 DAG 拆分为多个阶段和更详细的逻辑部分 - tasks。一个阶段内的任务可以并行计算，因为内部没有洗牌阶段，但是图像中的两个并行分支呢？我知道所有关于 rdd 抽象的直觉（惰性评估等），但这并没有让我更容易理解。请给我任何建议。

Answer 1

我假设 DAG 看起来像这样：我说的对吗？

是的。

如果是，我们是否可以假设计算 rdd3 时使用的 rdd1 仍然在内存中处理？

没有。 Spark 利用惰性求值来处理数据。这意味着在需要之前不会计算任何内容。除非有明确的声明，否则不会存储任何内容。

或者我们必须缓存 rdd1 以防止重复加载？

没错，你需要缓存 rdd1 以防止文本文件被读取两次。

更一般地说，如果 DAG 看起来像这样：我们是否可以假设两个分支都是并行计算的并使用相同的 rdd1 副本？还是 Spark 驱动程序会一个接一个地计算这些分支，因为这是两个不同的阶段？

这两个分支不会被并行处理，因为它们有不同的谱系。通常，在采取行动之前不会处理任何数据。每当需要一个结果（读取，调用一个动作）时，所有正在进行的转换和给定沿袭中的当前动作的数据处理都会发生。之后，除非调用cache，否则内存中将不存在任何数据。

请参阅此 deck 了解转换与操作的说明

Answer 2

我们可以假设用于计算 rdd3 的 rdd1 仍然在内存中处理吗？或者我们必须缓存 rdd1 以防止重复加载？

rdd1 未缓存。要缓存它，您需要显式缓存它，并确保.cache()返回的RDD存储在用于创建rdd2的变量中：

rdd1 = sc.textFile("...").cache()
rdd2 = rdd1.map().groupBy().filter()

如果要将其添加到您的图表中，rdd1 和 rdd2 之间将有另一个 RDD。

我们可以假设两个分支都是并行计算的并使用相同的 rdd1 副本吗？还是 Spark 驱动程序会一个接一个地计算这些分支，因为这是两个不同的阶段？

决定作业是否并行运行的不是（分支）沿袭，而是作业请求本身。假设你跑了：

rdd4.count()
rdd5.count()

这将执行rdd1->rdd2->rdd4 然后计数，然后 rdd1->rdd3->rdd5。此计算将是顺序的。

要并行运行这两个图，必须并行提交作业（异步调用）。有很多问题可以解决这个话题。检查this 和this。