Spark Parkett Partitionierung: Große Anzahl von Dateien

Question

Ich versuche, Funken Partitionierung zu nutzen. Ich habe versucht, so etwas wie

data.write.partitionBy("key").parquet("/location") 

Die Frage ist hier jede Partition erstellt große Anzahl von Parkett-Dateien zu tun, die langsame Lese führen, wenn ich aus dem Stammverzeichnis zu lesen versuchen.

Um zu vermeiden, dass ich

data.coalese(numPart).write.partitionBy("key").parquet("/location") 

versuchte Dies schafft jedoch numPart Anzahl von Parkett-Dateien in jeder Partition. Jetzt ist meine Partitionsgröße anders. SO würde ich idealerweise eine separate Koaleszenz pro Partition haben wollen. Dies ist jedoch keine leichte Sache. Ich muss alle Partitionen besuchen, die zu einer bestimmten Nummer koaleszieren und an einem anderen Ort speichern.

Wie sollte ich Partitionierung verwenden, um viele Dateien nach dem Schreiben zu vermeiden?

Answer 1

Zuerst würde ich wirklich coalesce vermeiden, da dies oft bis weiter in der Kette der Transformation gedrückt wird und kann die Parallelität Ihrer Arbeit zerstören (ich zu diesem Thema gefragt, hier: How to prevent Spark optimization)

Writing 1 Datei pro Parkett-Partition ist realtively leicht (siehe Spark dataframe write method writing many small files):

data.repartition($"key").write.partitionBy("key").parquet("/location") 

Wenn Sie eine beliebige Anzahl von Dateien (oder Dateien, die alle die gleiche Größe haben) festlegen möchten, müssen Sie Ihre Daten weiter partitionieren ein anderes Attribut mit der könnte verwendet werden (ich kann dir nicht sagen, woran das liegen könnte) Ihr Fall):

data.repartition($"key",$"another_key").write.partitionBy("key").parquet("/location") 

another_key könnte ein weiteres Attribut des Datensatzes oder ein abgeleitetes Attribut mit einigen Modulo oder Rundungsoperationen auf bestehende Attribute sein. Sie könnten sogar Fenster-Funktionen mit row_number über key und dann um diese durch so etwas wie

data.repartition($"key",floor($"row_number"/N)*N).write.partitionBy("key").parquet("/location") 

Diese Datei