python - リソースを解放する目的でPySparkでRDDを削除するには?

Question

不要になった RDD がある場合、メモリから削除するにはどうすればよいですか? これを行うには、次のことで十分でしょうか。

del thisRDD

ありがとう！

Answer 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 の上で少なくとも 1 回アクションを呼び出すと、キャッシュされます。

>>> 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]

アドレスを使用して、Spark UI で永続化されたデータと永続化のレベルを簡単に確認できますhttp://<driver_node>:4040/storage。del thisRDDこの RDD の永続性を変更することはありませんが、永続thisRDD.unpersist()化を解除しますが、コードでこのRDD を使用することはできます (メモリには永続化されず、再計算されるたびに再計算されます)。問い合わせた）

Answer 2

簡単な答え: 場合によります。

pyspark v.1.3.0 ソース コードによると、Python マッパー/リデューサーによって生成された RDD である にはdel thisRDD十分なはずです。PipelinedRDD

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

RDD一方、クラスには__del__メソッドがありません (おそらく必要ですが) のでunpersist、自分でメソッドを呼び出す必要があります。

編集:メソッドは このコミット__del__で削除されました。

Answer 3

gc.collect()参考までに、後をお勧めしdelます（rddが大量のメモリを必要とする場合）。