我如何理解 Spark 中使用了缓存？答案

【问题标题】：How do I understand that caching is used in Spark?我如何理解 Spark 中使用了缓存？
【发布时间】：2020-01-27 17:07:28
【问题描述】：

在我的 Scala/Spark 应用程序中，我创建了 DataFrame。我计划在整个程序中多次使用这个 Dataframe。这就是为什么我决定对那个 DataFrame 使用.cache() 方法。正如您在循环中看到的那样，我使用不同的值多次过滤 DataFrame。出于某种原因，.count() 方法返回给我的结果总是相同的。实际上，它必须返回两个不同的计数值。另外，我注意到 Mesos 中的奇怪行为。感觉就像 .cache() 方法没有被执行。创建DataFrame后，程序会转到这部分代码if (!df.head(1).isEmpty)，并执行了很长时间。我假设缓存进程会运行很长时间，其他进程会使用这个缓存并快速运行。你觉得问题出在哪里？

import org.apache.spark.sql.DataFrame

var df: DataFrame = spark
    .read
    .option("delimiter", "|")
    .csv("/path_to_the_files/")
    .filter(col("col5").isin("XXX", "YYY", "ZZZ"))

df.cache()

var array1 = Array("111", "222")

var array2 = Array("333")

var storage = Array(array1, array2)

if (!df.head(1).isEmpty) {
    for (item <- storage) {
        df.filter(
            col("col1").isin(item:_*)
        )

        println("count: " + df.count())
    }
}

【问题讨论】：

检查spark.apache.org/docs/latest/…: "你可以标记一个RDD使用persist()或cache()方法持久化。第一个在动作中计算的时间，它将保存在节点的内存中。 "
一个绝对可以加快工作速度的更改是将isin 替换为left join，因为您正在计算col1 上存在多少次存储项。虽然我不确定这是否适合您的情况。

标签： scala dataframe apache-spark mesos

【解决方案1】：

其实它必须返回两个不同的计数值。

为什么？您在同一个df 上调用它。也许你的意思是

val df1 = df.filter(...)
println("count: " + df1.count())

我假设缓存进程会运行很长时间，其他进程会使用这个缓存并快速运行。

确实如此，但仅在执行依赖于此数据帧的第一个 action 时，并且 head 是该操作。所以你应该完全期待

程序转到这部分代码if (!df.head(1).isEmpty)并执行了很长时间

如果不使用缓存，您还将获得相同的 df.count() 调用时间，除非 Spark 检测到它并自行启用缓存。

【讨论】：

感谢您的回复。就我而言，我有一个巨大的 DataFrame，在本例中称为 df。我正在尝试在循环内使用不同的值过滤此 Dataframe。据我了解，在循环的每次迭代中，我都必须创建一个新的 DataFrame 并检查它的计数，对吗？在您看来，我的代码的什么顺序会加快它的工作速度？您认为在一定级别上使用.persist() 而不是.cache() 来加快速度是否有意义？
“除非 Spark 检测到它并自行启用缓存。” -> 有趣，你能说清楚这一点吗？
@EnzoBnl 我实际上不知道它是否会这样做，这是一个警告以防万一。