Tìm hiểu về bộ nhớ đệm của Spark

Question

Tôi đang cố gắng hiểu cách bộ nhớ cache của Spark hoạt động.

Đây là sự hiểu biết ngây thơ của tôi, xin vui lòng cho tôi biết nếu tôi là thiếu cái gì:

val rdd1 = sc.textFile("some data") 
rdd1.cache() //marks rdd1 as cached 
val rdd2 = rdd1.filter(...) 
val rdd3 = rdd1.map(...) 
rdd2.saveAsTextFile("...") 
rdd3.saveAsTextFile("...") 

Ở phía trên, rdd1 sẽ được nạp từ đĩa (ví dụ HDFS) chỉ một lần. (khi rdd2 được lưu tôi giả định) và sau đó từ bộ nhớ cache (giả sử có đủ RAM) khi rdd3 được lưu)

Bây giờ đây là câu hỏi của tôi. Giả sử tôi muốn cache rdd2 và rdd3 vì chúng sẽ được sử dụng sau này, nhưng tôi không cần rdd1 sau khi tạo chúng.

Về cơ bản có sự trùng lặp, phải không? Kể từ khi rdd2 và rdd3 được tính toán, tôi không cần rdd1 nữa, tôi có lẽ nên unpersist nó, phải không? câu hỏi là khi nào?

Điều này có hiệu quả không? (Tùy chọn A)

val rdd1 = sc.textFile("some data") 
rdd1.cache() // marks rdd as cached 
val rdd2 = rdd1.filter(...) 
val rdd3 = rdd1.map(...) 
rdd2.cache() 
rdd3.cache() 
rdd1.unpersist() 

Tia lửa có thêm lời gọi không đáng tin cậy vào DAG không? hoặc là nó được thực hiện ngay lập tức? nếu nó được thực hiện ngay lập tức, sau đó về cơ bản rdd1 sẽ không được lưu trữ khi tôi đọc từ rdd2 và rdd3, phải không?

Tôi có nên thực hiện theo cách này thay thế (Tùy chọn B) không?

val rdd1 = sc.textFile("some data") 
rdd1.cache() // marks rdd as cached 
val rdd2 = rdd1.filter(...) 
val rdd3 = rdd1.map(...) 

rdd2.cache() 
rdd3.cache() 

rdd2.saveAsTextFile("...") 
rdd3.saveAsTextFile("...") 

rdd1.unpersist() 

Vậy câu hỏi là thế này: là lựa chọn A đủ tốt? tức là rdd1 vẫn chỉ tải tệp một lần? Hoặc tôi có cần phải đi với Tùy chọn B không?

Answer 1

Dường như tùy chọn B là bắt buộc. Lý do có liên quan đến cách persist/cache và unpersist được thực hiện bởi Spark. Vì các phép biến đổi RDD chỉ xây dựng các mô tả DAG mà không thực hiện, trong Tùy chọn A vào thời điểm bạn gọi là unpersist, bạn vẫn chỉ có các mô tả công việc chứ không phải là thực thi đang chạy.

Điều này có liên quan bởi vì cuộc gọi cache hoặc persist chỉ thêm RDD vào Bản đồ RDD tự đánh dấu là vẫn tồn tại trong quá trình thực hiện công việc. Tuy nhiên, unpersist trực tiếp cho blockManager gỡ bỏ RDD khỏi bộ nhớ và loại bỏ tham chiếu trong Bản đồ các RDD liên tục.

persist function

unpersist function

Vì vậy, bạn sẽ cần phải gọi unpersist sau Spark thực sự thực hiện và lưu trữ các RDD với người quản lý khối.

Các ý kiến ​​cho gợi ý RDD.persist phương pháp hướng này: rdd.persist

Answer 2

Trong tùy chọn A, bạn đã không được hiển thị khi bạn đang gọi hành động (gọi điện để tiết kiệm)

val rdd1 = sc.textFile("some data") 
rdd.cache() //marks rdd as cached 
val rdd2 = rdd1.filter(...) 
val rdd3 = rdd1.map(...) 
rdd2.cache() 
rdd3.cache() 
rdd1.unpersist() 
rdd2.saveAsTextFile("...") 
rdd3.saveAsTextFile("...") 

Nếu dãy là như trên, Tùy chọn A nên sử dụng phiên bản được lưu trong bộ nhớ cache của rdd1 để tính cả rdd2 và rdd 3

Answer 3

Tùy chọn B là phương pháp tối ưu với tinh chỉnh nhỏ. Sử dụng các phương pháp hành động ít tốn kém hơn. Trong cách tiếp cận được đề cập bởi mã của bạn, saveAsTextFile là một hoạt động tốn kém, thay thế nó bằng phương pháp đếm.

Idea ở đây là để loại bỏ các rdd1 lớn từ DAG, nếu nó không có liên quan để tính toán tiếp theo (sau rdd2 và rdd3 được tạo ra)

phương pháp cập nhật từ mã

val rdd1 = sc.textFile("some data").cache() 
val rdd2 = rdd1.filter(...).cache() 
val rdd3 = rdd1.map(...).cache() 

rdd2.count 
rdd3.count 

rdd1.unpersist()