Bản đồ tia lửaPhân tích vs valient lười biếng thoáng qua

Question

Tôi đã tự hỏi điều gì khác biệt khi sử dụng tia lửa mapPartitions chức năng so với val lười biếng thoáng qua.
Vì mỗi phân vùng cơ bản chạy trên một nút khác nhau, một trường hợp đơn giản của val lười biếng thoáng qua sẽ được tạo trên mỗi đầu (giả sử nó trong một đối tượng).

Ví dụ:

class NotSerializable(v: Int) { 
    def foo(a: Int) = ??? 
} 

object OnePerPartition { 
    @transient lazy val obj: NotSerializable = new NotSerializable(10) 
} 

object Test extends App{ 
    val conf = new SparkConf().setMaster("local[2]").setAppName("test") 
    val sc = new SparkContext(conf) 

    val rdd: RDD[Int] = sc.parallelize(1 to 100000) 

    rdd.map(OnePerPartition.obj.foo) 

    // ---------- VS ---------- 

    rdd.mapPartitions(itr => { 
     val obj = new NotSerializable(10) 
     itr.map(obj.foo) 
    }) 
} 

Người ta có thể hỏi tại sao bạn sẽ thậm chí còn muốn nó ...
Tôi muốn tạo ra một khái niệm chứa chung của chạy logic của tôi trên bất kỳ thực hiện bộ sưu tập chung (RDD, List , scalding pipe, v.v.)
Tất cả đều có khái niệm "bản đồ", nhưng mapPartition là duy nhất cho spark.

Answer 1

Trước hết bạn không cần transientlazy tại đây. Sử dụng object wrapper là đủ để làm cho công việc này và bạn thực sự có thể viết những dòng này như:

object OnePerExecutor { 
    val obj: NotSerializable = new NotSerializable(10) 
} 

Có một sự khác biệt cơ bản giữa các wrapper đối tượng và khởi tạo NotSerializable bên mapPartitions. Điều này:

rdd.mapPartitions(iter => { 
    val ns = NotSerializable(1) 
    ??? 
}) 

tạo một trường hợp NotSerializable đơn lẻ cho mỗi phân vùng. Mặt khác,

Trình bao bọc đối tượng, tạo một cá thể NotSerializable đơn lẻ cho mỗi JVM thực thi. Kết quả là trường hợp này:

• Có thể được sử dụng để xử lý nhiều phân đoạn.
• Có thể được truy cập đồng thời bởi nhiều chuỗi trình xử lý.
• Có tuổi thọ vượt quá cuộc gọi hàm khi được sử dụng.

Có nghĩa là nó phải là chủ đề an toàn và mọi cuộc gọi phương pháp đều phải là tác dụng phụ miễn phí.