如何在 PySpark 中删除 RDD 以释放资源？答案

【问题标题】：How to delete an RDD in PySpark for the purpose of releasing resources?如何在 PySpark 中删除 RDD 以释放资源？
【发布时间】：2015-03-15 10:49:09
【问题描述】：

如果我有一个不再需要的 RDD，如何从内存中删除它？以下是否足以完成这项工作：

del thisRDD

谢谢！

【问题讨论】：

对我来说，以下代码行成功了：for (id, rdd) in sc._jsc.getPersistentRDDs().items(): rdd.unpersist()

标签： python apache-spark pyspark

【解决方案1】：

不，del thisRDD 还不够，它只会删除指向 RDD 的指针。您应该调用thisRDD.unpersist() 来删除缓存的数据。

供您参考，Spark 使用惰性计算模型，这意味着当您运行此代码时：

>>> thisRDD = sc.parallelize(xrange(10),2).cache()

您不会真正缓存任何数据，它只会在 RDD 执行计划中标记为“待缓存”。你可以这样检查：

>>> print thisRDD.toDebugString()
(2) PythonRDD[6] at RDD at PythonRDD.scala:43 [Memory Serialized 1x Replicated]
 |  ParallelCollectionRDD[5] at parallelize at PythonRDD.scala:364 [Memory Serialized 1x Replicated]

但是当你在这个 RDD 之上至少调用一次操作时，它会被缓存：

>>> thisRDD.count()
10
>>> print thisRDD.toDebugString()
(2) PythonRDD[6] at RDD at PythonRDD.scala:43 [Memory Serialized 1x Replicated]
 |       CachedPartitions: 2; MemorySize: 174.0 B; TachyonSize: 0.0 B; DiskSize: 0.0 B
 |  ParallelCollectionRDD[5] at parallelize at PythonRDD.scala:364 [Memory Serialized 1x Replicated]

您可以使用地址http://<driver_node>:4040/storage 在 Spark UI 中轻松检查持久化数据和持久化级别。你会看到del thisRDD 不会改变这个 RDD 的持久性，但是thisRDD.unpersist() 会取消它，而你仍然可以在你的代码中使用这个RDD（虽然它不会再持久化并且会每次查询时重新计算）

【讨论】：

【解决方案2】：

简短回答：视情况而定。

根据pyspark v.1.3.0 source code，del thisRDD应该足够PipelinedRDD，这是Python mapper/reducer生成的RDD：

class PipelinedRDD(RDD):
    # ...
    def __del__(self):
        if self._broadcast:
            self._broadcast.unpersist()
            self._broadcast = None

另一方面，RDD 类没有__del__ 方法（虽然它可能应该），所以你应该自己调用unpersist 方法。

编辑： __del__ 方法在 this 提交中被删除。

【讨论】：

可以提供分支特定的 URL。根据日期，它建议您链接到 1.1 或 1.2，但看起来当时没有 __del__，现在肯定没有。
这是 v1.3.0 更新了链接。 This commit 杀死了它。
谢谢！我明白我是多么想念它了——它只存在于 1.2 中。
如果我理解正确的话，__del__ 没有在 Python 端实现也没关系。当最后一个对它的引用消失时，RDD 将在 Java 端取消持久化。如果 Python 端的所有引用都已删除，那么 Py4J 会确保 Java 端的引用也消失，从而执行 Java RDD 终结器。（我添加了一个解释我的推理的答案，但它可以使用评论。）

【解决方案3】：

仅供参考，我会在del 之后推荐gc.collect()（如果rdd 占用大量内存）。

【讨论】：

【解决方案4】：

简答：以下代码应该可以解决问题：

import gc
del thisRDD
gc.collect()

说明：

即使你使用的是 PySpark，你的 RDD 的数据也是在 Java 端管理的，所以首先让我们问同样的问题，但是对于 Java 而不是 Python：

如果我使用 Java，并且我只是释放对我的 RDD 的所有引用，是否足以自动取消持久化它？

对于 Java，答案是肯定的，根据this answer，RDD 在被垃圾回收时会自动取消持久化。（显然该功能已在 this PR 中添加到 Spark。）

好的，在 Python 中会发生什么？如果我在 Python 中删除对我的 RDD 的所有引用，是否会导致它们在 Java 端被删除？

PySpark 使用Py4J 将对象从 Python 发送到 Java，反之亦然。根据Py4J Memory Model Docs：

一旦对象在 Python VM 上被垃圾回收（引用计数 == 0），该引用就会在 Java VM 上被删除

但请注意：删除对 RDD 的 Python 引用不会导致它立即被删除。您必须等待 Python 垃圾收集器清理引用。您可以阅读 Py4J 的详细说明，他们推荐以下内容：

调用gc.collect() 通常也可以。

好的，现在回到你原来的问题：

以下是否足以完成这项工作：
del thisRDD

几乎。您应该删除对它的最后一个引用（即del thisRDD），然后，如果您确实需要立即取消持久化RDD**，请致电gc.collect()。

**嗯，从技术上讲，这将立即删除 Java 端的 reference，但是在 Java 的垃圾收集器真正执行 RDD 的终结器并因此取消持久化数据之前会有一点延迟。

【讨论】：