Apache Spark: carte vs mapPartitions?

Imp. ASTUCE :

Chaque fois que vous avez des poids lourds de l'initialisation qui devrait être fait une fois
pour de nombreux RDD éléments plutôt qu'une fois par RDD élément, et si ce
d'initialisation, comme la création d'objets à partir d'un tiers
bibliothèque, ne peut pas être sérialisé (de sorte que l'Étincelle peut transmettre à travers
le cluster de nœuds de travail), l'utilisation mapPartitions() au lieu de
map(). mapPartitions() fournit pour l'initialisation à faire
une fois par travailleur tâche/thread/partition au lieu d'une fois par RDD de données
élément pour exemple : voir ci-dessous.

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

---

T2. ne flatMap se comporter comme une carte ou comme mapPartitions?

Oui. voir l'exemple 2 de flatmap.. son auto-explicatif.

T1. Quelle est la différence entre un CA de map et mapPartitions

map fonctionne de la fonction utilisée par l'élément de niveau tout en
mapPartitions exerce la fonction au niveau de la partition.

Exemple de Scénario : si nous avons 100 K éléments dans un RDD partition, puis nous mettrons le feu au large de la fonction en cours d'utilisation par la cartographie de la transformation de 100K fois lorsque nous utilisons map.

À l'inverse, si nous utilisons mapPartitions alors nous ne appel à une fonction particulière d'un moment, mais nous allons passer dans tous les 100K dossiers et obtenir toutes les réponses en un seul appel de fonction.

Il y aura gain de performance depuis map travaille sur une fonction en particulier de nombreuses fois, en particulier si la fonction est en train de faire quelque chose de cher à chaque fois qu'il n'aurait pas besoin de le faire si nous avons passé dans tous les éléments à la fois(dans le cas de mappartitions).

carte

Applique une transformation de la fonction sur chaque élément de la RDD et les retours
le résultat en tant que nouvelle RDD.

Liste Des Variantes

def de la carte[U: ClassTag](f: T => U): CA[U]

Exemple :

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

C'est une institution spécialisée de la carte qui est appelé une seule fois pour chaque partition.
L'ensemble du contenu de chaque partitions est disponible en tant que
flux séquentiel de valeurs via l'argument d'entrée (Iterarator[T]).
La fonction doit retourner encore une autre Itérateur[U]. Le combiné
résultat des itérateurs sont automatiquement converties dans un nouveau CA. Merci
notez que le n-uplets (3,4) et (6,7) sont manquantes à partir de la suite
grâce à la partition que nous avons choisi.

preservesPartitioning indique si la fonction d'entrée préserve l'
programme de partitionnement, qui devrait être false sauf si c'est une paire de RDD et de l'entrée
la fonction ne doit pas modifier les touches.

Liste Des Variantes

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

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

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

Le programme ci-dessus peut également être écrite à l'aide de flatMap comme suit.

Exemple 2 en utilisant 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) 

Conclusion :

mapPartitions transformation est plus rapide que map depuis il appelle votre fonction une fois la partition/, pas une seule fois/élément..

Pour en savoir plus : foreach Vs foreachPartitions Quand utiliser Quoi?