caching - 将缓存的 Spark 数据帧与其他数据帧连接并再次缓存的有效方法

Question

我有一个已缓存的大型数据框

val largeDf = someLargeDataframe.cache

现在我需要将它与一个微小的结合起来并再次缓存

val tinyDf = someTinyDataframe.cache
val newDataframe = largeDf.union(tinyDf).cached
tinyDf.unpersist()
largeDf.unpersist()

效率非常低，因为需要重新缓存所有数据。有没有有效的方法可以将少量数据添加到大型缓存数据帧中？

<小时/>

阅读 Teodors 的解释后，我知道在对新数据帧执行某些操作之前，我无法取消旧数据帧的持久化。但是如果我需要做这样的事情怎么办？

def myProcess(df1: Dataframe, df2: Dataframe): Dataframe{
    val df1_trans = df1.map(....).cache
    val df2_trans = df2.map(....).cache

    doSomeAction(df1_trans, df2_trans)

    val finalDf = df1_trans.union(df2_trans).map(....).cache
    // df1_trans.unpersist()
    // df2_trans.unpersist()
    finalDf
}

我希望缓存我的 df1_trans 和 df2_trans 以提高函数内部的性能，因为它们将被多次调用，但我最终需要返回的数据帧也是由 df1_trans 和 df2_trans 构造的，如果可以的话不要在离开函数之前取消持久化它们，我永远找不到其他地方来执行此操作，但是，如果我取消持久化它们，我的 FinalDf 将不会从缓存中受益。

遇到这种情况我能做什么？谢谢!

Answer 1

val largeDf = someLargeDataframe.cache
val tinyDf = someTinyDataframe.cache
val newDataframe = largeDf.union(tinyDf).cache

如果您现在在执行所有 largeDf 数据帧的任何操作之前调用 unpersist()，您将不会从缓存这两个数据帧中受益。

tinyDf.unpersist()
largeDf.unpersist()

只要其他两个数据帧已经缓存，我就不会担心缓存联合数据帧，您不太可能会看到性能下降。

对以下内容进行基准测试:

========= now? ============
val largeDf = someLargeDataframe.cache
val tinyDf = someTinyDataframe.cache
val newDataframe = largeDf.union(tinyDf).cache
tinyDf.unpersist()
largeDf.unpersist()
#force evaluation
newDataframe.count()

========= alternative 1 ============
val largeDf = someLargeDataframe.cache
val tinyDf = someTinyDataframe.cache
val newDataframe = largeDf.union(tinyDf).cache

#force evaluation
newDataframe.count()
tinyDf.unpersist()
largeDf.unpersist()

======== alternative 2 ==============
val largeDf = someLargeDataframe.cache
val tinyDf = someTinyDataframe.cache
val newDataframe = largeDf.union(tinyDf)

newDataframe.count()


======== alternative 3 ==============
val largeDf = someLargeDataframe
val tinyDf = someTinyDataframe
val newDataframe = largeDf.union(tinyDf).cache

#force evaluation
newDataframe.count()