Apache Spark - foreach-Vs foreachPartitions, Wann, Was?

foreach und foreachPartitions Aktionen.

foreach(Funktion): Einheit

Eine generische Funktion für das aufrufen der Operationen mit Nebenwirkungen. Für jeden
element in der RDD, ruft es die Funktion übergeben . Das ist
in der Regel verwendet für die Manipulation von Akkumulatoren oder schreiben auf externe
speichert.

Hinweis: ändern von Variablen in anderen als Akkumulatoren außerhalb des foreach() möglicherweise nicht definiertes Verhalten zur Folge. Sehen Verständnis Verschlüsse für mehr details.

Beispiel :

scala> val accum = sc.longAccumulator("My Accumulator")
accum: org.apache.spark.util.LongAccumulator = LongAccumulator(id: 0, name: Some(My Accumulator), value: 0)

scala> sc.parallelize(Array(1, 2, 3, 4)).foreach(x => accum.add(x))
...
10/09/29 18:41:08 INFO SparkContext: Tasks finished in 0.317106 s

scala> accum.value
res2: Long = 10

foreachPartition(Funktion): Einheit

Ähnlich foreach() , sondern Aufruf der Funktion für jede
element, er fordert es für jede partition. Die Funktion sollte in der Lage sein
zu akzeptieren einen iterator. Dies ist effizienter, als foreach() weil
es reduziert die Anzahl der Funktionsaufrufe (wie mapPartitions() ).

Nutzung von foreachPartition Beispiele:

• Beispiel1 : für jede partition eine Datenbank-Verbindung (Innerhalb der für jede partition-block), die Sie verwenden möchten, dann ist dies ein Beispiel, wie es getan werden kann, mit scala.

/** 
* Legen Sie in der Datenbank mit foreach-partition. 
* 
* @param sqlDatabaseConnectionString 
* @param sqlTableName 
*/
def insertToTable(sqlDatabaseConnectionString: String, sqlTableName: String): Unit = { 

//numPartitions = Anzahl gleichzeitiger DB-verbindungen können Sie planen, geben Sie 

datframe.neu partitionieren(numofpartitionsyouwant) 

val tableHeader: String = dataFrame.Spalten.mkString(",") 
dataFrame.foreachPartition { partition => 
//Hinweis : für Jede partition eine Verbindung (mehr bessere Weg ist die Verwendung von connection pools) 
val sqlExecutorConnection: Connection = DriverManager.getConnection(sqlDatabaseConnectionString) 
//Batch-Größe von 1000 verwendet, da einige Datenbanken verwenden können batch-Größe von mehr als 1000 für die ex : Azure sql 
partition.gruppiert(1000).foreach { 
group => 
val insertString: scala.Sammlung.veränderlich.StringBuilder = new scala.Sammlung.veränderlich.StringBuilder() 
Gruppe.foreach { 
Datensatz => insertString.append("('" + record.mkString(",") + "'),") 
} 

sqlExecutorConnection.createStatement() 
.executeUpdate(f"INSERT INTO [$sqlTableName] ($tableHeader) WERTE" 
+ insertString.stripSuffix(",")) 
} 


sqlExecutorConnection.close() //schließen der Verbindung, so dass verbindungen nicht erschöpfen. 
} 
} 

• Beispiel2 :

Nutzung von foreachPartition mit sparkstreaming (dstreams) und kafka producer

dstream.foreachRDD { rdd =>
  rdd.foreachPartition { partitionOfRecords =>
//only once per partition You can safely share a thread-safe Kafka //producer instance.
    val producer = createKafkaProducer()
    partitionOfRecords.foreach { message =>
      producer.send(message)
    }
    producer.close()
  }
}

Hinweis:, Wenn Sie es vermeiden möchten, diese Möglichkeit zu schaffen, Produzent einmal pro partition, betterway ist die broadcast-Produzent mit
sparkContext.broadcast seit Kafka producer ist asynchron und
Puffer Daten stark, bevor Sie senden.

Akku-Proben snippet zu spielen, um mit ihm durch die...
Sie können testen, die Leistung

 - test("Foreach - Spark") { 
import spark.implicits._ 
var accum = sc.longAccumulator 
sc.parallelisieren(Seq(1,2,3)).foreach(x => accum.add(x)) 
assert(accum.Wert == 6L) 
} 

test("Foreach-partition - Spark") { 
import spark.implicits._ 
var accum = sc.longAccumulator 
sc.parallelisieren(Seq(1,2,3)).foreachPartition(x => x.foreach(accum.add(_))) 
assert(accum.Wert == 6L) 
} 

Fazit :

foreachPartition Operationen auf Partitionen, so offensichtlich, es wäre
besser edge als foreach

Faustregel :

foreachPartition sollte verwendet werden, wenn Sie den Zugriff auf teure
Ressourcen wie Datenbankverbindungen oder kafka Produzent usw.. das würde initialisieren
eine pro-partition eher als ein pro-element(foreach). wenn es
kommt Akkumulatoren, können Sie Messen die Leistung durch die oben genannten test
Methoden, die sollten schneller arbeiten, im Fall von Akkumulatoren als auch..

Auch... siehe Karte vs mappartitions, die hat ein ähnliches Konzept, aber Sie sind Transformationen.

Den foreachPartition bedeutet nicht, es ist pro Knoten Aktivität, sondern es wird ausgeführt, für jede partition und es ist möglich, dass Sie möglicherweise eine große Anzahl von partition im Vergleich zur Anzahl der Knoten in diesem Fall Ihre Leistung kann beeinträchtigt werden. Wenn Sie beabsichtigen, tun eine Aktivität auf node-Ebene die Lösung erklärt hier kann nützlich sein, es ist zwar nicht von mir getesteten