了解 Spark 的缓存

Question

我试图了解 Spark 的缓存是如何工作的。

这是我的幼稚理解，如果我遗漏了什么，请告诉我：

val rdd1 = sc.textFile("some data")
rdd1.cache() //marks rdd1 as cached
val rdd2 = rdd1.filter(...)
val rdd3 = rdd1.map(...)
rdd2.saveAsTextFile("...")
rdd3.saveAsTextFile("...")

在上面，rdd1 只会从磁盘（例如 HDFS）加载一次。 （我假设保存 rdd2 时），然后在保存 rdd3 时从缓存（假设有足够的 RAM）中）

现在这是我的问题。假设我想缓存 rdd2 和 rdd3 因为它们都将在以后使用，但在创建它们之后我不需要 rdd1。

基本上有重复，不是吗？由于一旦计算了 rdd2 和 rdd3 ，我就不再需要 rdd1 了，我可能应该取消它，对吧？问题是什么时候？

这行得通吗？ （选项 A）

val rdd1 = sc.textFile("some data")
rdd1.cache()   // marks rdd as cached
val rdd2 = rdd1.filter(...)
val rdd3 = rdd1.map(...)
rdd2.cache()
rdd3.cache()
rdd1.unpersist()

spark 是否将 unpersist 调用添加到 DAG？还是立即完成？如果它立即完成，那么当我从 rdd2 和 rdd3 读取时，基本上 rdd1 将不会被缓存，对吗？

我应该这样做（选项 B）吗？

val rdd1 = sc.textFile("some data")
rdd1.cache()   // marks rdd as cached
val rdd2 = rdd1.filter(...)
val rdd3 = rdd1.map(...)

rdd2.cache()
rdd3.cache()

rdd2.saveAsTextFile("...")
rdd3.saveAsTextFile("...")

rdd1.unpersist()

所以问题是这样的： 选项A是否足够好？即rdd1 是否仍然只加载文件一次？ 还是我需要选择选项 B？

Answer 1

似乎需要选项 B。原因与Spark如何执行persist/cache和unpersist有关。由于 RDD 转换仅构建 DAG 描述而不执行，因此在选项 A 中，当您调用 unpersist 时，您仍然只有作业描述，而不是正在运行的执行。

这是相关的，因为 cache 或 persist 调用只是将 RDD 添加到 RDD 的 Map 中，这些 RDD 将自己标记为在作业执行期间持久化。但是，unpersist 直接告诉 blockManager 将 RDD 从存储中逐出，并删除持久 RDD 的 Map 中的引用。

persist function

unpersist function

因此，您需要在 Spark 实际执行并使用块管理器存储 RDD 后调用 unpersist。

RDD.persist 方法的 cmets 暗示了这一点： rdd.persist

Answer 2

在选项A中，您在调用操作时没有显示（调用保存）

val rdd1 = sc.textFile("some data")
rdd.cache() //marks rdd as cached
val rdd2 = rdd1.filter(...)
val rdd3 = rdd1.map(...)
rdd2.cache()
rdd3.cache()
rdd1.unpersist()
rdd2.saveAsTextFile("...")
rdd3.saveAsTextFile("...")

如果顺序如上，选项 A 应该使用 rdd1 的缓存版本来计算 rdd2 和 rdd 3

Answer 3

选项 B 是一种带有小幅调整的最佳方法。使用成本较低的行动方法。在您的代码中提到的方法中， saveAsTextFile 是一项昂贵的操作，请用 count 方法替换它。

这里的想法是从 DAG 中删除大的 rdd1，如果它与进一步的计算无关（在创建 rdd2 和 rdd3 之后）

从代码更新方法

val rdd1 = sc.textFile("some data").cache()
val rdd2 = rdd1.filter(...).cache() 
val rdd3 = rdd1.map(...).cache()

rdd2.count
rdd3.count

rdd1.unpersist()