如果我缓存 Spark Dataframe，然后覆盖引用，原始数据帧是否仍会被缓存？

假设我有一个函数来生成 (py)spark 数据帧，并将数据帧作为最后一个操作缓存到内存中。

def gen_func(inputs):
   df = ... do stuff...
   df.cache()
   df.count()
   return df

根据我的理解，Spark的缓存工作原理如下：

1. When cache/persist加上一个动作（count()) 在数据上调用 帧，它是从它的 DAG 计算出来并缓存到内存中，贴在 指向引用它的对象。
2. 只要存在对该对象的引用（可能在其他函数/其他范围内），df 将继续被缓存，并且依赖于 df 的所有 DAG 将使用内存中缓存的数据作为起点。
3. 如果删除了对 df 的所有引用，Spark 会将缓存作为内存进行垃圾收集。它可能不会立即被垃圾收集，从而导致一些短期内存块（特别是，如果您生成缓存数据并太快地丢弃它们，则会导致内存泄漏），但最终它会被清除。

我的问题是，假设我使用gen_func生成数据框，但随后覆盖原始数据框引用（可能使用filter or a withColumn).

df=gen_func(inputs)
df=df.filter("some_col = some_val")

在Spark中，RDD/DF是不可变的，因此过滤器之后重新分配的df和过滤器之前的df引用了两个完全不同的对象。在这种情况下，对原始 df 的引用是cache/counted已被覆盖。这是否意味着缓存的数据帧不再可用并且将被垃圾收集？这是否意味着新的后置过滤器df是否会从头开始计算所有内容，尽管是从先前缓存的数据帧生成的？

我问这个问题是因为我最近正在修复代码中的一些内存不足问题，在我看来，缓存可能是问题所在。然而，我还不太了解使用缓存的安全方法是什么，以及如何意外地使缓存内存失效的全部细节。我的理解中缺少什么？我在执行上述操作时是否偏离了最佳实践？

我做了几个实验，如下所示。显然，数据帧一旦被缓存，仍然被缓存（如图所示getPersistentRDDs和查询计划 -InMemory等），即使所有 Python 引用都被覆盖或完全删除del，并显式调用垃圾收集。

实验一：

def func():
    data = spark.createDataFrame([[1],[2],[3]]).toDF('col1')
    data.cache()
    data.count()
    return data

sc._jsc.getPersistentRDDs()

df = func()
sc._jsc.getPersistentRDDs()

df2 = df.filter('col1 != 2')
del df
import gc
gc.collect()
sc._jvm.System.gc()
sc._jsc.getPersistentRDDs()

df2.select('*').explain()

del df2
gc.collect()
sc._jvm.System.gc()
sc._jsc.getPersistentRDDs()

Results:

>>> def func():
...     data = spark.createDataFrame([[1],[2],[3]]).toDF('col1')
...     data.cache()
...     data.count()
...     return data
...
>>> sc._jsc.getPersistentRDDs()
{}

>>> df = func()
>>> sc._jsc.getPersistentRDDs()
{71: JavaObject id=o234}

>>> df2 = df.filter('col1 != 2')
>>> del df
>>> import gc
>>> gc.collect()
93
>>> sc._jvm.System.gc()
>>> sc._jsc.getPersistentRDDs()
{71: JavaObject id=o240}

>>> df2.select('*').explain()
== Physical Plan ==
*(1) Filter (isnotnull(col1#174L) AND NOT (col1#174L = 2))
+- *(1) ColumnarToRow
   +- InMemoryTableScan [col1#174L], [isnotnull(col1#174L), NOT (col1#174L = 2)]
         +- InMemoryRelation [col1#174L], StorageLevel(disk, memory, deserialized, 1 replicas)
               +- *(1) Project [_1#172L AS col1#174L]
                  +- *(1) Scan ExistingRDD[_1#172L]

>>> del df2
>>> gc.collect()
85
>>> sc._jvm.System.gc()
>>> sc._jsc.getPersistentRDDs()
{71: JavaObject id=o250}

实验2：

def func():
    data = spark.createDataFrame([[1],[2],[3]]).toDF('col1')
    data.cache()
    data.count()
    return data

sc._jsc.getPersistentRDDs()

df = func()
sc._jsc.getPersistentRDDs()

df = df.filter('col1 != 2')
import gc
gc.collect()
sc._jvm.System.gc()
sc._jsc.getPersistentRDDs()

df.select('*').explain()

del df
gc.collect()
sc._jvm.System.gc()
sc._jsc.getPersistentRDDs()

Results:

>>> def func():
...     data = spark.createDataFrame([[1],[2],[3]]).toDF('col1')
...     data.cache()
...     data.count()
...     return data
...
>>> sc._jsc.getPersistentRDDs()
{}

>>> df = func()
>>> sc._jsc.getPersistentRDDs()
{86: JavaObject id=o317}

>>> df = df.filter('col1 != 2')
>>> import gc
>>> gc.collect()
244
>>> sc._jvm.System.gc()
>>> sc._jsc.getPersistentRDDs()
{86: JavaObject id=o323}

>>> df.select('*').explain()
== Physical Plan ==
*(1) Filter (isnotnull(col1#220L) AND NOT (col1#220L = 2))
+- *(1) ColumnarToRow
   +- InMemoryTableScan [col1#220L], [isnotnull(col1#220L), NOT (col1#220L = 2)]
         +- InMemoryRelation [col1#220L], StorageLevel(disk, memory, deserialized, 1 replicas)
               +- *(1) Project [_1#218L AS col1#220L]
                  +- *(1) Scan ExistingRDD[_1#218L]

>>> del df
>>> gc.collect()
85
>>> sc._jvm.System.gc()
>>> sc._jsc.getPersistentRDDs()
{86: JavaObject id=o333}

实验3（对照实验，证明unpersist works)

def func():
    data = spark.createDataFrame([[1],[2],[3]]).toDF('col1')
    data.cache()
    data.count()
    return data

sc._jsc.getPersistentRDDs()

df = func()
sc._jsc.getPersistentRDDs()

df2 = df.filter('col1 != 2')
df2.select('*').explain()

df.unpersist()
df2.select('*').explain()

Results:

>>> def func():
...     data = spark.createDataFrame([[1],[2],[3]]).toDF('col1')
...     data.cache()
...     data.count()
...     return data
...
>>> sc._jsc.getPersistentRDDs()
{}

>>> df = func()
>>> sc._jsc.getPersistentRDDs()
{116: JavaObject id=o398}

>>> df2 = df.filter('col1 != 2')
>>> df2.select('*').explain()
== Physical Plan ==
*(1) Filter (isnotnull(col1#312L) AND NOT (col1#312L = 2))
+- *(1) ColumnarToRow
   +- InMemoryTableScan [col1#312L], [isnotnull(col1#312L), NOT (col1#312L = 2)]
         +- InMemoryRelation [col1#312L], StorageLevel(disk, memory, deserialized, 1 replicas)
               +- *(1) Project [_1#310L AS col1#312L]
                  +- *(1) Scan ExistingRDD[_1#310L]

>>> df.unpersist()
DataFrame[col1: bigint]
>>> sc._jsc.getPersistentRDDs()
{}

>>> df2.select('*').explain()
== Physical Plan ==
*(1) Project [_1#310L AS col1#312L]
+- *(1) Filter (isnotnull(_1#310L) AND NOT (_1#310L = 2))
   +- *(1) Scan ExistingRDD[_1#310L]

回答OP的问题：

这是否意味着缓存的数据帧不再可用并且将被垃圾收集？这是否意味着新的后置过滤器 df 将从头开始计算所有内容，尽管是从先前缓存的数据帧生成的？

实验表明no对彼此而言。数据帧保持缓存状态，不会被垃圾收集，并且根据查询计划，使用缓存的（不可引用的）数据帧计算新的数据帧。

一些与缓存使用相关的有用功能（如果您不想通过 Spark UI 执行此操作）包括：

sc._jsc.getPersistentRDDs()，它显示了缓存的 RDD/数据帧的列表，以及

spark.catalog.clearCache()，这会清除所有缓存的 RDD/数据帧。

我在执行上述操作时是否偏离了最佳实践？

我无权就此评判你，但正如其中一条评论所建议的，避免重新分配给df因为数据帧是不可变的。尝试想象你正在用 scala 编码并且你定义了df as a val. Doing df = df.filter(...)是不可能的。 Python 本身无法强制执行这一点，但我认为最佳实践是避免覆盖任何数据帧变量，以便您始终可以调用df.unpersist()之后，如果您不再需要缓存的结果。