Apa perbedaan antara RDD map dan mapPartitionsmetode? Dan apakah flatMapberperilaku suka mapatau suka mapPartitions? Terima kasih.

(Sunting) yaitu apa perbedaan (baik secara semantik atau dalam hal eksekusi) antara

  def map[A, B](rdd: RDD[A], fn: (A => B))
               (implicit a: Manifest[A], b: Manifest[B]): RDD[B] = {
    rdd.mapPartitions({ iter: Iterator[A] => for (i <- iter) yield fn(i) },
      preservesPartitioning = true)
  }

Dan:

  def map[A, B](rdd: RDD[A], fn: (A => B))
               (implicit a: Manifest[A], b: Manifest[B]): RDD[B] = {
    rdd.map(fn)
  }

Apa perbedaan antara peta RDD dan metode mapPartitions?

Peta metode mengubah setiap elemen dari RDD sumber menjadi elemen tunggal dari hasil RDD dengan menerapkan fungsi. mapPartitions mengubah setiap partisi RDD sumber menjadi beberapa elemen dari hasil (mungkin tidak ada).

Dan apakah flatMap berperilaku seperti map atau like mapPartitions?

Baik flatMap bekerja pada elemen tunggal (as map) dan menghasilkan banyak elemen dari hasilnya (as mapPartitions).

Imp. TIP:

Setiap kali Anda memiliki inisialisasi kelas berat yang harus dilakukan satu kali untuk banyak RDDelemen daripada satu kali per RDDelemen, dan jika inisialisasi ini, seperti pembuatan objek dari perpustakaan pihak ketiga, tidak dapat diserialisasi (sehingga Spark dapat mengirimkannya melintasi gugus ke node pekerja), gunakan mapPartitions()sebagai ganti map(). mapPartitions()menyediakan inisialisasi yang harus dilakukan sekali per tugas pekerja / utas / partisi bukannya sekali per RDDelemen data misalnya: lihat di bawah.

val newRd = myRdd.mapPartitions(partition => {
  val connection = new DbConnection /*creates a db connection per partition*/

  val newPartition = partition.map(record => {
    readMatchingFromDB(record, connection)
  }).toList // consumes the iterator, thus calls readMatchingFromDB 

  connection.close() // close dbconnection here
  newPartition.iterator // create a new iterator
})

Q2. apakah flatMapberperilaku seperti peta atau suka mapPartitions?

Iya. silakan lihat contoh 2 dari flatmap.. penjelasannya sendiri.

Q1. Apa perbedaan antara RDD mapdanmapPartitions

mapberfungsi dengan fungsi yang digunakan pada level per elemen sambil mapPartitionsmenjalankan fungsi di level partisi.

Skenario Contoh : jika kita memiliki elemen 100K diRDDpartisitertentumaka kita akan menjalankan fungsi yang digunakan oleh transformasi pemetaan 100K kali ketika kita gunakanmap.

Sebaliknya, jika kita menggunakan mapPartitionsmaka kita hanya akan memanggil fungsi tertentu satu kali, tetapi kita akan melewati semua catatan 100 ribu dan mendapatkan kembali semua tanggapan dalam satu panggilan fungsi.

Akan ada perolehan kinerja karena mapbekerja pada fungsi tertentu berkali-kali, terutama jika fungsi tersebut melakukan sesuatu yang mahal setiap kali itu tidak perlu dilakukan jika kita melewati semua elemen sekaligus (dalam kasus mappartitions).

peta

Menerapkan fungsi transformasi pada setiap item RDD dan mengembalikan hasilnya sebagai RDD baru.

Daftar Varian

def map [U: ClassTag] (f: T => U): RDD [U]

Contoh:

val a = sc.parallelize(List("dog", "salmon", "salmon", "rat", "elephant"), 3)
 val b = a.map(_.length)
 val c = a.zip(b)
 c.collect
 res0: Array[(String, Int)] = Array((dog,3), (salmon,6), (salmon,6), (rat,3), (elephant,8)) 

mapPartitions

Ini adalah peta khusus yang hanya dipanggil sekali untuk setiap partisi. Seluruh konten partisi masing-masing tersedia sebagai aliran nilai berurutan melalui argumen input (Iterarator [T]). Fungsi kustom harus mengembalikan Iterator [U] lainnya. Iterator hasil gabungan secara otomatis dikonversi menjadi RDD baru. Harap dicatat, bahwa tupel (3,4) dan (6,7) hilang dari hasil berikut karena partisi yang kami pilih.

preservesPartitioningmenunjukkan apakah fungsi input mempertahankan partisi, yang seharusnya falsekecuali ini adalah RDD pasangan dan fungsi input tidak mengubah kunci.

Daftar Varian

def mapPartitions [U: ClassTag] (f: Iterator [T] => Iterator [U], preservesPartitioning: Boolean = false): RDD [U]

Contoh 1

val a = sc.parallelize(1 to 9, 3)
 def myfunc[T](iter: Iterator[T]) : Iterator[(T, T)] = {
   var res = List[(T, T)]()
   var pre = iter.next
   while (iter.hasNext)
   {
     val cur = iter.next;
     res .::= (pre, cur)
     pre = cur;
   }
   res.iterator
 }
 a.mapPartitions(myfunc).collect
 res0: Array[(Int, Int)] = Array((2,3), (1,2), (5,6), (4,5), (8,9), (7,8)) 

Contoh 2

val x = sc.parallelize(List(1, 2, 3, 4, 5, 6, 7, 8, 9,10), 3)
 def myfunc(iter: Iterator[Int]) : Iterator[Int] = {
   var res = List[Int]()
   while (iter.hasNext) {
     val cur = iter.next;
     res = res ::: List.fill(scala.util.Random.nextInt(10))(cur)
   }
   res.iterator
 }
 x.mapPartitions(myfunc).collect
 // some of the number are not outputted at all. This is because the random number generated for it is zero.
 res8: Array[Int] = Array(1, 2, 2, 2, 2, 3, 3, 3, 3, 3, 3, 3, 3, 3, 4, 4, 4, 4, 4, 4, 4, 5, 7, 7, 7, 9, 9, 10) 

Program di atas juga dapat ditulis menggunakan flatMap sebagai berikut.

Contoh 2 menggunakan flatmap

val x  = sc.parallelize(1 to 10, 3)
 x.flatMap(List.fill(scala.util.Random.nextInt(10))(_)).collect

 res1: Array[Int] = Array(1, 2, 3, 3, 3, 4, 4, 4, 4, 4, 4, 4, 4, 4, 5, 5, 6, 6, 6, 6, 6, 6, 6, 6, 7, 7, 7, 8, 8, 8, 8, 8, 8, 8, 8, 9, 9, 9, 9, 9, 10, 10, 10, 10, 10, 10, 10, 10) 

Kesimpulan:

mapPartitionstransformasi lebih cepat daripada mapkarena memanggil fungsi Anda sekali / partisi, bukan sekali / elemen ..

Bacaan lebih lanjut: foreach Vs foreachPartitions Kapan menggunakan What?

Peta :

1. Ini memproses satu baris pada satu waktu, sangat mirip dengan metode map () dari MapReduce.
2. Anda kembali dari transformasi setelah setiap baris.

MapPartitions

1. Ini memproses partisi lengkap dalam sekali jalan.
2. Anda dapat kembali dari fungsi hanya sekali setelah memproses seluruh partisi.
3. Semua hasil antara harus disimpan dalam memori sampai Anda memproses seluruh partisi.
4. Memberi Anda fungsi pengaturan () peta () dan pembersihan () di MapReduce

Map Vs mapPartitions http://bytepadding.com/big-data/spark/spark-map-vs-mappartitions/

Spark Map http://bytepadding.com/big-data/spark/spark-map/

Spark mapPartitions http://bytepadding.com/big-data/spark/spark-mappartitions/