Was ist der Unterschied zwischen einem RDD map und einer mapPartitionsMethode? Und verhält flatMapsich wie mapoder wie mapPartitions? Vielen Dank.

(bearbeiten) dh was ist der Unterschied (entweder semantisch oder in Bezug auf die Ausführung) zwischen

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

Und:

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

Was ist der Unterschied zwischen der Karte eines RDD und der mapPartitions-Methode?

Das Verfahren Karte wandelt jedes Element der Quelle RDD in ein einzelnes Element des Ergebnisses RDD durch eine Funktion der Anwendung. mapPartitions konvertiert jede Partition der Quell-RDD in mehrere Elemente des Ergebnisses (möglicherweise keine).

Und verhält sich flatMap wie map oder wie mapPartitions?

Weder, flatMap arbeitet auf einem einzigen Element (wie map) und erzeugt mehrere Elemente des Ergebnisses (wie mapPartitions).

Imp. TRINKGELD :

Wenn Sie eine Schwergewichtsinitialisierung haben, die einmal für viele RDDElemente und nicht einmal pro RDDElement durchgeführt werden sollte, und wenn diese Initialisierung, z. B. das Erstellen von Objekten aus einer Bibliothek eines Drittanbieters, nicht serialisiert werden kann (damit Spark sie über den Cluster an übertragen kann) die Arbeiterknoten) verwenden mapPartitions()anstelle von map(). mapPartitions()sieht vor, dass die Initialisierung einmal pro Worker-Task / Thread / Partition statt einmal pro RDDDatenelement durchgeführt wird. Beispiel: siehe unten.

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. verhält flatMapsich wie eine Karte oder wie mapPartitions?

Ja. Bitte sehen Sie Beispiel 2 von flatmap.. es ist selbsterklärend.

Q1. Was ist der Unterschied zwischen einem RDD mapundmapPartitions

mapArbeitet die Funktion, die auf Elementebene verwendet wird, während mapPartitionsdie Funktion auf Partitionsebene ausgeführt wird.

Beispielszenario : Wenn eine bestimmteRDDPartition100.000 Elemente enthält, wird die von der Zuordnungstransformation verwendete Funktion bei Verwendung 100.000 Mal ausgelöstmap.

Wenn wir umgekehrt verwenden, mapPartitionsrufen wir die jeweilige Funktion nur einmal auf, übergeben jedoch alle 100.000 Datensätze und erhalten alle Antworten in einem Funktionsaufruf zurück.

Es wird einen Leistungsgewinn geben, da die mapArbeit an einer bestimmten Funktion so oft ausgeführt wird, insbesondere wenn die Funktion jedes Mal etwas Teueres tut, das nicht erforderlich wäre, wenn wir alle Elemente gleichzeitig übergeben würden (im Fall von mappartitions).

Karte

Wendet eine Transformationsfunktion auf jedes Element der RDD an und gibt das Ergebnis als neue RDD zurück.

Varianten auflisten

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

Beispiel:

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

Dies ist eine spezielle Zuordnung, die für jede Partition nur einmal aufgerufen wird. Der gesamte Inhalt der jeweiligen Partitionen steht über das Eingabeargument (Iterarator [T]) als sequentieller Wertestrom zur Verfügung. Die benutzerdefinierte Funktion muss noch einen weiteren Iterator [U] zurückgeben. Die kombinierten Ergebnisiteratoren werden automatisch in eine neue RDD konvertiert. Bitte beachten Sie, dass die Tupel (3,4) und (6,7) aufgrund der von uns gewählten Partitionierung im folgenden Ergebnis fehlen.

preservesPartitioningGibt an, ob die Eingabefunktion den Partitionierer falsebeibehält. Dies sollte der Fall sein, es sei denn, es handelt sich um ein RDD-Paar, und die Eingabefunktion ändert die Tasten nicht.

Varianten auflisten

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

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

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

Das obige Programm kann auch wie folgt mit flatMap geschrieben werden.

Beispiel 2 mit 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) 

Fazit :

mapPartitionsDie Transformation ist schneller als mapda sie Ihre Funktion einmal / Partition aufruft, nicht einmal / Element.

Weiterführende Literatur: foreach Vs foreachPartitions Wann was verwenden?

Karte :

1. Es verarbeitet jeweils eine Zeile, sehr ähnlich der map () -Methode von MapReduce.
2. Sie kehren nach jeder Zeile von der Transformation zurück.

MapPartitions

1. Es verarbeitet die gesamte Partition auf einmal.
2. Sie können nach der Verarbeitung der gesamten Partition nur einmal von der Funktion zurückkehren.
3. Alle Zwischenergebnisse müssen gespeichert werden, bis Sie die gesamte Partition verarbeiten.
4. Bietet Ihnen die Funktion setup () map () und cleanup () von 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/