opuszczając puste przegródki DataFrame Apache Spark

Question

staram się ponownego dzielenia DataFrame według columnm The DataFrame ma N (powiedzmy N=3) różne wartości w partycji kolumnie x, np:

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

Co ja Chcąc to osiągnąć, należy ponownie rozdzielić myDF przez x bez tworzenia pustych partycji. Czy jest lepszy sposób niż to zrobić?

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

(Gdybym nie określają numParts w repartiton, większość moich partycji są puste (jak repartition tworzy 200 partycje) ...)

Answer 1

myślałem roztworu z iteracji po df partycja i pobieranie rekordu w nim znajdują się, aby znaleźć niepuste partycje.

val nonEmptyPart = sparkContext.longAccumulator("nonEmptyPart") 

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

Jak mamy niepuste partycje (nonEmptyPart), możemy oczyścić pustych partycji za pomocą coalesce() (check coalesce() vs reparation()).

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

To może lub nie może być najlepszy, ale to rozwiązanie uniknie tasowanie jak nie używamy reparation()

---

przykład zająć komentarz

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

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