我们可以通过保存 parquet 文件来复制 Spark 的 .cache() 行为吗？答案

【问题标题】：Can we replicate Spark's .cache() behavior but by saving parquet file instead?我们可以通过保存 parquet 文件来复制 Spark 的 .cache() 行为吗？
【发布时间】：2019-10-28 17:56:23
【问题描述】：

假设我有 4 个 spark 指令块，分别表示为 Dataframe A、B、C 和 D。

数据框有依赖关系：

C 要求完成 A 和 B
D 要求 B 完成

现在我想同时保存 C 和 D 输出。为了不重新计算 B 两次，我使用了

B = B.cache()

事实上，如果我先保存 C，我会让 A 和 B 并行运行，然后是 C，然后当我保存 D 时，它只需要计算 D，这在某种程度上是足够好的调度。

现在的问题是：如果不是缓存 B，而是像这样将其保存在镶木地板中

B.write.mode("overwrite").parquet(CacheLocation)
B = spark.read.parquet(CacheLocation)

整个运行速度比使用 B.cache() 至少快 10 倍（比在 B 上完全不使用任何缓存快 30 倍）

上述方法的问题是现在我不能同时运行 A 和 B，除非我使用 Futures 并决定缓存 A。我的性能非常受限，我的真实代码有超过 50 个块/4 个输出要同步，除非我可以使用 Spark 的默认调度，否则这是一场噩梦。

我想知道是否有任何方法可以覆盖 Spark 的缓存方法以保存为镶木地板而不是内存？或者有什么方法可以在 Spark 自己的调度程序中添加这种 parquet 缓存？

伪代码说明：

场景 1：使用缓存，整个运行 300 秒，但 A 和 B 并行运行

val A = parquet.read(A).withColumn(NewA,FormulaA)
val B = parquet.read(B).withColumn(NewB,FormulaC).cache
val C = A.join(B, FormulaA===FormulaC,left)
val D = B.withColumn(NewD, FormulaD)
C.write.mode("overwrite").parquet(SaveCLocation)
D.write.mode("overwrite").parquet(SaveDLocation)

场景2：使用parquet作为缓存，整个运行30s，但A和B串行运行

val A = parquet(A).withColumn(NewA,FormulaA)
var B = parquet(B).withColumn(NewB,FormulaC)
B.write.parquet(CacheLocation)
B = spark.read.parquet(CacheLocation)
val C = A.join(B, FormulaA===FormulaC,left)
val D = B.withColumn(NewD, FormulaD)
C.write.parquet(SaveCLocation)
D.write.parquet(SaveDLocation)

【问题讨论】：

你写的东西对我来说没有意义。 IO 始终是最慢的解决方案。我怀疑你体验到它快 10 倍的原因是因为当你调用 C 和 D 时 B 数据帧实际上并没有被缓存。请记住，Spark 进行惰性评估，所以是否有动作（比如count）强制评估在 C 和 D 中使用 B 之前（并因此缓存）它？如果你能提供更多的代码，也许可以？
我可以看到缓存的行为就像我在 SparkWebUI 上提到的那样，所以我很确定这不是问题。当我保存 C 时，我会执行 C.write.mode("overwrite").parquet(SaveLocation) 并且它是第一次评估 B。 parquet 格式一定有一些东西可以让我的所有操作（过滤器、连接和 if/then 瀑布）比缓存的数据帧快得多。我的代码太复杂，无法放在这里，如果我的情况有希望，我可以尝试设置可以重现问题的虚拟代码。
write 将强制评估您的 B 数据框。如果您只使用cache() 并且在在 D 和 C 中使用它之前没有做任何事情来强制 B 被评估（并因此被缓存），Spark 在必须开始评估 D 和 C 之前不会评估（或缓存）B，这可以解释您的性能差异。
另外，如果你能提供你使用的代码，这真的很有帮助，这样我们就可以准确地看到你的评估顺序和你对cache和write的使用。 ..
我添加了代码说明

标签： scala apache-spark lazy-evaluation parquet

【解决方案1】：

从您在此处发布的内容来看，我的怀疑是正确的。

在此操作之前，您的cache 场景中不会评估（或缓存）B

C.write.mode("overwrite").parquet(SaveCLocation)

然而，在您的 parquet 场景中，B 在此操作上被评估（并存储）：

B.write.parquet(CacheLocation)

换句话说，您无法比较val C = A.join(B, FormulaA===FormulaC,left) 在两种情况下的执行时间，因为在一种情况下，B 已经被评估并且只需要从磁盘中读取，而在另一种情况下，它必须首先评估 B。

尝试在您的缓存场景中添加一个B.count - 它会强制评估B，就像B.write 在您的parquet 场景中所做的那样。然后比较val C = A.join(B, FormulaA===FormulaC,left) 在两个场景中的运行时间——我相信你会看到缓存场景有很大的优势:)

val A = parquet.read(A).withColumn(NewA,FormulaA)
val B = parquet.read(B).withColumn(NewB,FormulaC).cache
B.count // force evaluation of B
val C = A.join(B, FormulaA===FormulaC,left)
val D = B.withColumn(NewD, FormulaD)
C.write.mode("overwrite").parquet(SaveCLocation)
D.write.mode("overwrite").parquet(SaveDLocation)

【讨论】：

抱歉，我的措辞可能不正确，我比较的是在两种情况下运行 A+B+C+D 的时间。否则，是的，这不会是苹果对苹果。
您还可以尝试添加B.count 吗？（在执行join 时进行缓存 - 根据我的经验 - 不是一个好主意......）
通过添加计数，火花将首先计算 B，一旦完成，它将计算 A，然后是 C，然后是 D。很难相信这可能比并行计算 A 和 B 更快，同时缓存 B 然后计算 C 然后 D。也就是说，我会尝试这样做以了解速度有多慢只是为了好玩:)
我认为你误解了 Spark 的并行特性。无论有没有B.count，每个动作都将开始对导致该动作的阶段进行并行评估。但是这些动作本身并没有被并行评估。因此，在您的情况下，它甚至会在考虑 D.write 之前评估 - 并完成 - C.write。（顺便说一句，动作是 count、write、collect、take 和 first。在调用其中一个之前，不会评估任何内容）。
@Simon，我认为你应该随心所欲，但请阅读火花阶段和阶段边界。我显然没有很好地向您解释这一点，也许其他人会做得更好 - 我将通过强调仅使用 write 来缓存内容在 spark 中不是最佳的。