Dropping leere Datenrahmen Partitionen in Apache Spark

Question

ich versuche, einen Datenrahmen neu zu partitionieren nach einem columnm die der Datenrahmen hat N (lassen N=3 sagen) unterschiedliche Werte in der Partition-Spalte x, zB:

val myDF = sc.parallelize(Seq(1,1,2,2,3,3)).toDF("x") // create dummy data 

Was ich mag es zu erreichen ist, um myDF von x zu repartiton, ohne leere Partitionen zu produzieren. Gibt es einen besseren Weg als das?

val numParts = myDF.select($"x").distinct().count.toInt 
myDF.repartition(numParts,$"x") 

(Wenn ich numParts nicht angeben, in repartiton, sind die meisten meiner Partitionen leer (als repartition 200 Partitionen) schafft ...)

Answer 1

I der Lösung über df mit Iterieren denken würde, Partition und holen Datensatzanzahl in, um nicht leere Partitionen zu finden.

val nonEmptyPart = sparkContext.longAccumulator("nonEmptyPart") 

df.foreachPartition(partition => 
    if (partition.length > 0) nonEmptyPart.add(1)) 

Da wir nicht leere Partitionen (nonEmptyPart) erhalten haben, können wir leere Partitionen reinigen, indem Sie coalesce() (check coalesce() vs reparation()).

val finalDf = df.coalesce(nonEmptyPart.value.toInt) //coalesce() accepts only Int type 

Es kann oder auch nicht die beste sein, aber diese Lösung wird schlurfenden vermeiden, da wir nicht reparation() verwenden

---

Beispiel Kommentar

val df1 = sc.parallelize(Seq(1, 1, 2, 2, 3, 3)).toDF("x").repartition($"x") 
val nonEmptyPart = sc.longAccumulator("nonEmptyPart") 

df1.foreachPartition(partition => 
    if (partition.length > 0) nonEmptyPart.add(1)) 

val finalDf = df1.coalesce(nonEmptyPart.value.toInt) 

println(s"nonEmptyPart => ${nonEmptyPart.value.toInt}") 
println(s"df.rdd.partitions.length => ${df1.rdd.partitions.length}") 
println(s"finalDf.rdd.partitions.length => ${finalDf.rdd.partitions.length}") 

Output zu adressieren

nonEmptyPart => 3 
df.rdd.partitions.length => 200 
finalDf.rdd.partitions.length => 3