В чем разница между RDD map и mapPartitionsметодом? И ведет flatMapсебя как mapили нравитсяmapPartitions ? Спасибо.

(править) то есть в чем разница (семантически или с точки зрения исполнения) между

  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)
  }

И:

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

В чем разница между картой RDD и методом mapPartitions?

Метод карта преобразует каждый элемент исходного RDD в единый элемент результата RDD путем применения функции. mapPartitions преобразует каждый раздел исходного RDD в несколько элементов результата (возможно, ни одного).

И flatMap ведет себя как map или как mapPartitions?

Также flatMap работает с одним элементом (as map) и производит несколько элементов результата (as mapPartitions).

Настоятельный НАКОНЕЧНИК :

Всякий раз, когда у вас есть тяжелая инициализация, которая должна выполняться один раз для многих RDDэлементов, а не один раз для каждого RDDэлемента, и если эта инициализация, такая как создание объектов из сторонней библиотеки, не может быть сериализована (чтобы Spark мог передать ее через кластер в рабочие узлы), используйте mapPartitions()вместо map(). mapPartitions()предусматривает, что инициализация выполняется один раз для каждой рабочей задачи / потока / раздела, а не один раз для каждого RDDэлемента данных, например: см. ниже.

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. ведет flatMapсебя как карта или как mapPartitions?

Да. см. пример 2 flatmap.. это не требует пояснений.

Q1. В чем разница между RDD mapиmapPartitions

mapвыполняет функцию, используемую на уровне каждого элемента, в то время как mapPartitionsвыполняет функцию на уровне раздела.

Пример сценария : если у нас есть 100 КБ элементов в конкретномRDDразделе, тогда мы 100 КБ запускаем функцию, используемую преобразованием отображения, когда используемmap.

И наоборот, если мы используем, mapPartitionsто мы вызовем конкретную функцию только один раз, но мы передадим все 100K записей и вернем все ответы за один вызов функции.

Произойдет прирост производительности, поскольку mapмного раз работает с определенной функцией, особенно если функция каждый раз делает что-то дорогостоящее, чего не нужно было бы делать, если бы мы передали все элементы сразу (в случае mappartitions).

карта

Применяет функцию преобразования к каждому элементу RDD и возвращает результат как новый RDD.

Варианты листинга

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

Пример :

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

Это специализированная карта, которая вызывается только один раз для каждого раздела. Все содержимое соответствующих разделов доступно в виде последовательного потока значений через входной аргумент (Iterarator [T]). Пользовательская функция должна возвращать еще один Iterator [U]. Объединенные итераторы результатов автоматически преобразуются в новый СДР. Обратите внимание, что кортежи (3,4) и (6,7) отсутствуют в следующем результате из-за выбранного нами разбиения.

preservesPartitioningуказывает, сохраняет ли функция ввода разделитель, что должно быть, falseесли только это не RDD пары и функция ввода не изменяет ключи.

Варианты листинга

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

Пример 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)) 

Пример 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) 

Вышеупомянутая программа также может быть написана с использованием flatMap следующим образом.

Пример 2 с использованием плоской карты

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) 

Вывод :

mapPartitionsпреобразование происходит быстрее, чем mapпоскольку оно вызывает вашу функцию один раз / раздел, а не один раз / элемент ..

Дополнительная литература: foreach против foreachPartitions Когда использовать Что?

Карта :

1. Он обрабатывает одну строку за раз, очень похоже на метод map () в MapReduce.
2. Вы возвращаетесь из трансформации после каждой строки.

MapPartitions

1. Он обрабатывает весь раздел за один раз.
2. Вы можете вернуться из функции только один раз после обработки всего раздела.
3. Все промежуточные результаты необходимо хранить в памяти, пока вы не обработаете весь раздел.
4. Предоставляет вам как функции setup () map () и cleanup () 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/