Spark - repartition () vs coalesce ()

253

Laut Learning Spark

Beachten Sie, dass die Neupartitionierung Ihrer Daten ein ziemlich teurer Vorgang ist. Spark hat auch eine optimierte Version von repartition()aufgerufen coalesce(), mit der Datenverschiebungen vermieden werden können, jedoch nur, wenn Sie die Anzahl der RDD-Partitionen verringern.

Ein Unterschied, den ich bekomme, ist, dass mit repartition()der Anzahl der Partitionen erhöht / verringert werden kann, aber mit coalesce()der Anzahl der Partitionen kann nur verringert werden.

coalesce()Wie kann eine Datenverschiebung vermieden werden, wenn die Partitionen auf mehrere Computer verteilt sind und ausgeführt werden?

Praveen Sripati
quelle

Antworten:

353

Es vermeidet ein vollständiges Mischen. Wenn bekannt ist, dass die Anzahl abnimmt, kann der Executor Daten auf der minimalen Anzahl von Partitionen sicher speichern und nur die Daten von den zusätzlichen Knoten auf die von uns beibehaltenen Knoten verschieben.

Also würde es ungefähr so ​​aussehen:

Node 1 = 1,2,3
Node 2 = 4,5,6
Node 3 = 7,8,9
Node 4 = 10,11,12

Dann coalescebis zu 2 Partitionen:

Node 1 = 1,2,3 + (10,11,12)
Node 3 = 7,8,9 + (4,5,6)

Beachten Sie, dass für Knoten 1 und Knoten 3 die ursprünglichen Daten nicht verschoben werden mussten.

Justin Pihony
quelle
115
Danke für die Antwort. Die Dokumentation hätte besser sagen sollen minimize data movementals avoiding data movement.
Praveen Sripati
12
Gibt es einen Fall, in dem statt repartitionverwendet werden sollte coalesce?
Niemand
21
@Niemand Ich denke, die aktuelle Dokumentation deckt dies ziemlich gut ab: github.com/apache/spark/blob/… Denken Sie daran, dass nur der Parameter true repartitionaufgerufen wird . Lassen Sie mich wissen, ob das hilft. coalesceshuffle
Justin Pihony
2
Ist es möglich, die Anzahl der vorhandenen Partitionsdateien zu reduzieren? Ich habe keine HDFS, aber Problem mit vielen Dateien.
2
Die Aufteilung wird statistisch langsamer sein, da sie nicht weiß, dass sie schrumpft ... obwohl sie das vielleicht optimieren könnten. Intern nennt es nur shuffle = true
Verschmelzung
170

Justins Antwort ist großartig und diese Antwort geht tiefer.

Der repartitionAlgorithmus mischt vollständig und erstellt neue Partitionen mit Daten, die gleichmäßig verteilt sind. Erstellen wir einen DataFrame mit den Zahlen 1 bis 12.

val x = (1 to 12).toList
val numbersDf = x.toDF("number")

numbersDf enthält 4 Partitionen auf meinem Computer.

numbersDf.rdd.partitions.size // => 4

So werden die Daten auf die Partitionen aufgeteilt:

Partition 00000: 1, 2, 3
Partition 00001: 4, 5, 6
Partition 00002: 7, 8, 9
Partition 00003: 10, 11, 12

Lassen Sie uns die repartitionMethode vollständig mischen und diese Daten auf zwei Knoten abrufen.

val numbersDfR = numbersDf.repartition(2)

So werden die numbersDfRDaten auf meinem Computer partitioniert:

Partition A: 1, 3, 4, 6, 7, 9, 10, 12
Partition B: 2, 5, 8, 11

Die repartitionMethode erstellt neue Partitionen und verteilt die Daten gleichmäßig in den neuen Partitionen (die Datenverteilung ist bei größeren Datenmengen gleichmäßiger).

Unterschied zwischen coalesceundrepartition

coalesceVerwendet vorhandene Partitionen, um die Datenmenge zu minimieren, die gemischt wird. repartitionErstellt neue Partitionen und mischt vollständig. coalesceDies führt zu Partitionen mit unterschiedlichen Datenmengen (manchmal Partitionen mit sehr unterschiedlichen Größen) und repartitionzu ungefähr gleich großen Partitionen.

Ist coalesceoder repartitionschneller?

coalesceläuft möglicherweise schneller als repartition, aber Partitionen mit ungleicher Größe arbeiten im Allgemeinen langsamer als Partitionen mit gleicher Größe. Normalerweise müssen Sie Datasets nach dem Filtern eines großen Datensatzes neu partitionieren. Ich habe festgestellt repartition, dass Spark insgesamt schneller ist, da Spark für Partitionen gleicher Größe ausgelegt ist.

NB Ich habe merkwürdigerweise beobachtet, dass eine Neupartitionierung die Größe der Daten auf der Festplatte erhöhen kann . Stellen Sie sicher, dass Sie Tests ausführen, wenn Sie Repartition / Coalesce für große Datenmengen verwenden.

Lesen Sie diesen Blog-Beitrag, wenn Sie noch mehr Details wünschen.

Wenn Sie in der Praxis Coalesce & Repartition verwenden

Befugnisse
quelle
8
Tolle Antwort @Powers, aber sind die Daten in Partition A und B nicht verzerrt? Wie ist es gleichmäßig verteilt?
anwartheravian
Was ist der beste Weg, um die Partitionsgröße ohne OOM-Fehler zu ermitteln? Ich benutze, rdd.glom().map(len).collect()aber es gibt viele OOM-Fehler.
anwartheravian
8
@anwartheravian - Partition A und Partition B haben unterschiedliche Größen, da der repartitionAlgorithmus Daten für sehr kleine Datensätze nicht so gleichmäßig verteilt. Früher habe ich repartition5 Millionen Datensätze in 13 Partitionen organisiert und jede Datei war zwischen 89,3 MB und 89,6 MB groß - das ist ziemlich gleich!
Powers
1
@Powers dies sieht besser aus Antwort mit Detail.
Grün
1
Dies erklärt den Unterschied viel besser. Vielen Dank!
Abhi
22

Ein weiterer Punkt, der hier zu beachten ist, ist, dass das Grundprinzip von Spark RDD die Unveränderlichkeit ist. Durch die Neupartitionierung oder Zusammenführung wird eine neue RDD erstellt. Das Basis-RDD bleibt mit seiner ursprünglichen Anzahl von Partitionen bestehen. Wenn der Anwendungsfall die Beibehaltung der RDD im Cache erfordert, muss dies auch für die neu erstellte RDD durchgeführt werden.

scala> pairMrkt.repartition(10)
res16: org.apache.spark.rdd.RDD[(String, Array[String])] =MapPartitionsRDD[11] at repartition at <console>:26

scala> res16.partitions.length
res17: Int = 10

scala>  pairMrkt.partitions.length
res20: Int = 2
Harikrishnan Ck
quelle
Schön! Dies ist kritisch und zumindest für diesen erfahrenen Scala-Entwickler nicht offensichtlich - dh weder eine Neupartitionierung noch ein Koaleszenzversuch , die Daten zu ändern, sondern nur, wie sie über die Knoten verteilt sind
Doug,
1
@ Harikrishnan Wenn ich also die anderen Antworten richtig verstanden habe, verwendet Spark im Falle einer Koaleszenz vorhandene Partitionen. Da RDD jedoch unveränderlich ist, können Sie beschreiben, wie Coalesce vorhandene Partitionen nutzt. Nach meinem Verständnis dachte ich, dass Spark neue Partitionen in Koaleszenz an die vorhandenen Partitionen anfügt.
Explorer
Aber wenn das "alte" RDD nicht mehr verwendet wird, wie es im Ausführungsdiagramm bekannt ist, wird es aus dem Speicher gelöscht, wenn es nicht beibehalten wird, nicht wahr?
Markus
15

repartition - Es wird empfohlen, es zu verwenden, während die Anzahl der Partitionen erhöht wird, da alle Daten gemischt werden müssen.

coalesce- Es wird empfohlen, es zu verwenden und gleichzeitig die Anzahl der Partitionen zu verringern. Wenn Sie beispielsweise 3 Partitionen haben und diese auf 2 reduzieren möchten, coalescewerden die Daten der 3. Partition auf Partition 1 und 2 verschoben. Partition 1 und 2 verbleiben im selben Container. Auf der anderen Seite repartitionwerden Daten in allen Partitionen gemischt, daher ist die Netzwerknutzung zwischen den Ausführenden hoch und wirkt sich auf die Leistung aus.

coalesceLeistung besser als repartitionbeim Reduzieren der Anzahl der Partitionen.

Kamalesan C.
quelle
Nützliche Erklärung.
Narendra Maru
11

Was aus dem Code und den Codedokumenten folgt, coalesce(n)ist, dass es dasselbe ist wie coalesce(n, shuffle = false)und repartition(n)dasselbe wiecoalesce(n, shuffle = true)

Somit können beide coalesceund repartitionverwendet werden, um die Anzahl der Partitionen zu erhöhen

Mit shuffle = truekönnen Sie tatsächlich zu einer größeren Anzahl von Partitionen verschmelzen. Dies ist nützlich, wenn Sie eine kleine Anzahl von Partitionen haben, z. B. 100, wobei möglicherweise einige Partitionen ungewöhnlich groß sind.

Ein weiterer wichtiger Hinweis ist, dass Sie, wenn Sie die Anzahl der Partitionen drastisch verringern, die Verwendung einer gemischten Version von in Betracht ziehen sollten coalesce(wie repartitionin diesem Fall). Auf diese Weise können Ihre Berechnungen parallel auf übergeordneten Partitionen ausgeführt werden (mehrere Aufgaben).

Wenn Sie jedoch eine drastische Verschmelzung durchführen, z. B. zu numPartitions = 1, kann dies dazu führen, dass Ihre Berechnung auf weniger Knoten erfolgt, als Sie möchten (z. B. auf einem Knoten im Fall von numPartitions = 1). Um dies zu vermeiden, können Sie bestehen shuffle = true. Dies fügt einen Zufallsschritt hinzu, bedeutet jedoch, dass die aktuellen Upstream-Partitionen parallel ausgeführt werden (je nach aktueller Partitionierung).

Bitte beachten Sie auch die entsprechende Antwort hier

Kasur
quelle
10

Alle Antworten fügen dieser sehr oft gestellten Frage großes Wissen hinzu.

Nach der Tradition der Zeitachse dieser Frage sind hier meine 2 Cent.

Ich fand, dass die Aufteilung in einem ganz bestimmten Fall schneller als das Zusammenwachsen ist.

Wenn in meiner Anwendung die Anzahl der von uns geschätzten Dateien unter dem bestimmten Schwellenwert liegt, funktioniert die Neupartitionierung schneller.

Hier ist was ich meine

if(numFiles > 20)
    df.coalesce(numFiles).write.mode(SaveMode.Overwrite).parquet(dest)
else
    df.repartition(numFiles).write.mode(SaveMode.Overwrite).parquet(dest)

Wenn meine Dateien im obigen Snippet weniger als 20 waren, dauerte es ewig, bis die Koaleszenz abgeschlossen war, während die Neupartitionierung viel schneller war, und daher der obige Code.

Diese Anzahl (20) hängt natürlich von der Anzahl der Arbeitnehmer und der Datenmenge ab.

Hoffentlich hilft das.

Abhishek
quelle
6

Repartition : Mische die Daten in eine NEUE Anzahl von Partitionen.

Z.B. Der anfängliche Datenrahmen ist in 200 Partitionen unterteilt.

df.repartition(500): Die Daten werden von 200 Partitionen auf neue 500 Partitionen gemischt.

Zusammenführen : Mischt die Daten in die vorhandene Anzahl von Partitionen.

df.coalesce(5): Die Daten werden von den verbleibenden 195 Partitionen auf 5 vorhandene Partitionen gemischt.

Rahul
quelle
3

Ich möchte der Antwort von Justin und Power Folgendes hinzufügen:

repartitionignoriert vorhandene Partitionen und erstellt neue. Sie können es also verwenden, um Datenversatz zu beheben. Sie können Partitionsschlüssel erwähnen, um die Verteilung zu definieren. Datenversatz ist eines der größten Probleme im Problembereich "Big Data".

coalescearbeitet mit vorhandenen Partitionen und mischt eine Teilmenge davon. Es kann den Datenversatz nicht so stark beheben wie es der repartitionFall ist. Selbst wenn es weniger teuer ist, ist es möglicherweise nicht das, was Sie brauchen.

Salim
quelle
3

Zu all den tollen Antworten möchte ich hinzufügen, dass dies repartitioneine der besten Optionen ist, um die Vorteile der Datenparallelisierung zu nutzen. Währendcoalesce bietet eine kostengünstige Option zum Reduzieren der Partitionen und ist sehr nützlich, wenn Sie Daten in HDFS oder eine andere Senke schreiben, um große Schreibvorgänge zu nutzen.

Ich habe dies beim Schreiben von Daten im Parkettformat als nützlich empfunden, um den vollen Vorteil zu erzielen.

Ashkrit Sharma
quelle
2

Für jemanden, der Probleme beim Generieren einer einzelnen CSV-Datei aus PySpark (AWS EMR) als Ausgabe und beim Speichern auf s3 hatte, half die Verwendung der Neupartition. Der Grund dafür ist, dass das Zusammenwachsen nicht vollständig gemischt werden kann, die Partitionierung jedoch. Im Wesentlichen können Sie die Anzahl der Partitionen mithilfe der Neupartitionierung erhöhen oder verringern, aber nur die Anzahl der Partitionen (jedoch nicht 1) mithilfe der Koaleszenz verringern. Hier ist der Code für alle, die versuchen, eine CSV von AWS EMR nach s3 zu schreiben:

df.repartition(1).write.format('csv')\
.option("path", "s3a://my.bucket.name/location")\
.save(header = 'true')
rauben
quelle
0

Auf einfache Weise COALESCE: - dient nur dazu, die Anzahl der Partitionen zu verringern, kein Mischen von Daten, sondern nur die Partitionen zu komprimieren

REPARTITION: - dient sowohl zum Erhöhen als auch zum Verringern der Anzahl der Partitionen, es findet jedoch ein Mischen statt

Beispiel:-

val rdd = sc.textFile("path",7)
rdd.repartition(10)
rdd.repartition(2)

Beides funktioniert gut

Aber wir gehen im Allgemeinen auf diese beiden Dinge ein, wenn wir die Ausgabe in einem Cluster sehen müssen. Wir gehen damit um.

Bujuti Niranjan Reddy
quelle
9
Es wird auch Daten mit Coalese geben.
sun_dare
0

Sie sollten aber auch sicherstellen, dass die Daten, die zu Koaleszenzknoten kommen, hoch konfiguriert sind, wenn Sie mit großen Datenmengen arbeiten. Da alle Daten auf diese Knoten geladen werden, kann dies zu einer Speicherausnahme führen. Obwohl die Reparatur teuer ist, bevorzuge ich es, sie zu verwenden. Da mischt und verteilt die Daten gleichmäßig.

Seien Sie weise, zwischen Koaleszenz und Partitionierung zu wählen.

Arun Goudar
quelle
0

Der repartitionAlgorithmus mischt die Daten vollständig und erstellt gleich große Datenpartitionen. coalescekombiniert vorhandene Partitionen, um ein vollständiges Mischen zu vermeiden.

Coalesce eignet sich gut zum Erstellen einer RDD mit vielen Partitionen und zum Kombinieren von Partitionen auf einem einzelnen Worker-Knoten, um eine endgültige RDD mit weniger Partitionen zu erstellen.

Repartitionmischt die Daten in Ihrem RDD neu, um die endgültige Anzahl der von Ihnen angeforderten Partitionen zu erstellen. Die Partitionierung von DataFrames scheint ein Implementierungsdetail auf niedriger Ebene zu sein, das vom Framework verwaltet werden sollte, ist es aber nicht. Wenn Sie große DataFrames in kleinere filtern, sollten Sie die Daten fast immer neu partitionieren. Sie werden wahrscheinlich häufig große DataFrames in kleinere filtern. Gewöhnen Sie sich also an die Neupartitionierung.

Lesen Sie diesen Blog-Beitrag, wenn Sie noch mehr Details wünschen.

Sambhav Kumar
quelle