Làm thế nào để xóa một RDD trong PySpark với mục đích giải phóng tài nguyên?

Question

Nếu tôi có RDD mà tôi không còn cần, làm thế nào để xóa nó khỏi bộ nhớ? Sau đây là đủ để thực hiện việc này:

del thisRDD 

Cảm ơn!

Answer 1

Không, del thisRDD là không đủ, nó sẽ chỉ xóa con trỏ tới RDD. Bạn nên gọi thisRDD.unpersist() để xóa dữ liệu được lưu trong bộ nhớ cache.

Đối với bạn thông tin, Spark sử dụng một mô hình tính toán lười biếng, có nghĩa là khi bạn chạy mã này:

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

bạn sẽ không có bất kỳ dữ liệu lưu trữ thực sự, nó sẽ chỉ được đánh dấu là 'được lưu trữ 'trong kế hoạch thực thi RDD. Bạn có thể kiểm tra xem nó theo cách này:

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

Nhưng khi bạn gọi một hành động trên đầu trang của RDD này ít nhất một lần, nó sẽ trở thành cache:

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

Bạn có thể dễ dàng kiểm tra các dữ liệu tiếp tục tồn tại và mức độ kiên trì trong giao diện người dùng Spark sử dụng địa chỉ http://<driver_node>:4040/storage. Bạn sẽ thấy ở đó rằng del thisRDD sẽ không thay đổi độ bền của RDD này, nhưng thisRDD.unpersist() sẽ không thay đổi nó, trong khi bạn vẫn có thể sử dụng RDD này trong mã của bạn (trong khi nó sẽ không tồn tại trong bộ nhớ nữa và sẽ được tính toán lại mỗi lần nó được truy vấn)

Answer 2

Câu trả lời ngắn gọn: nó phụ thuộc.

Theo pyspark v.1.3.0 source code, del thisRDD nên là đủ cho PipelinedRDD, mà là một RDD tạo ra bởi Python mapper/giảm tốc:

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

RDD lớp mặt khác, không có phương pháp __del__ (trong khi nó có thể nên), vì vậy bạn nên tự mình gọi phương thức unpersist.

Chỉnh sửa:__del__ phương pháp đã bị xóa trong this cam kết.

Answer 3

Chỉ cần FYI, Tôi muốn giới thiệu "gc.collect()" sau "del" (nếu rdd mất nhiều bộ nhớ).

Answer 4

Câu trả lời ngắn: Các mã sau đây nên làm như lừa:

import gc 
del thisRDD 
gc.collect() 

---

Giải thích:

Thậm chí nếu bạn đang sử dụng PySpark, dữ liệu của RDD bạn được quản lý trên phía Java, vì vậy trước tiên hãy hỏi cùng một câu hỏi, nhưng đối với Java thay vì Python:

Nếu tôi đang sử dụng Java, và tôi đơn giản giải phóng tất cả các tham chiếu đến RDD của tôi, điều đó có đủ để tự động gỡ bỏ nó không?

Đối với Java, câu trả lời là CÓ, RDD sẽ tự động không được nhận khi nó được thu thập rác, theo số this answer. (Dường như chức năng đó đã được thêm vào Spark trong this PR.)

OK, điều gì sẽ xảy ra trong Python? Nếu tôi loại bỏ tất cả các tham chiếu đến RDD của tôi trong Python, điều đó có khiến chúng bị loại bỏ ở phía Java không?

PySpark sử dụng Py4J để gửi các đối tượng từ Python sang Java và ngược lại. Theo Py4J Memory Model Docs:

Khi đối tượng được thu gom rác thải trên Python VM (tính tham khảo == 0), tài liệu tham khảo được lấy ra trên Java VM

Nhưng hãy lưu ý: Loại bỏ các Python các tham chiếu đến RDD của bạn sẽ không làm cho nó bị ngay lập tức bị xóa. Bạn phải chờ cho bộ thu gom rác Python dọn sạch các tham chiếu. Bạn có thể đọc giải thích Py4J để biết chi tiết, nơi họ đề xuất những điều sau:

Cuộc gọi đến gc.collect() cũng thường hoạt động.

OK, bây giờ trở lại câu hỏi ban đầu của bạn:

có những điều sau đây là đủ để có được điều này thực hiện:

del thisRDD 

Hầu. Bạn nên loại bỏ tham chiếu cuối cùng (ví dụ: del thisRDD), và sau đó, nếu bạn thực sự cần RDD không được nhận ngay lập tức **, hãy gọi gc.collect(). ** Vâng, về mặt kỹ thuật, điều này sẽ ngay lập tức xóa tham chiếu ở phía Java, nhưng sẽ có một chút chậm trễ cho đến khi bộ thu gom rác của Java thực sự thực thi trình xác nhận của RDD và do đó không cho phép dữ liệu.