Thả phân vùng DataFrame trống trong Apache Spark

Question

tôi cố gắng phân vùng lại một DataFrame theo một columnm các các DataFrame có N (hãy nói N=3) giá trị khác nhau trong phân vùng cột x, ví dụ:

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

Những gì tôi muốn đạt được là repartiton myDF bởi x mà không tạo phân vùng trống. Có cách nào tốt hơn làm việc này không?

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

(Nếu tôi không chỉ định numParts trong repartiton, hầu hết các phân vùng của tôi là rỗng (như repartition tạo ra 200 phân vùng) ...)

Answer 1

Tôi muốn nghĩ về giải pháp với iterating qua df phân vùng và tìm nạp số bản ghi trong đó để tìm các phân vùng không trống.

val nonEmptyPart = sparkContext.longAccumulator("nonEmptyPart") 

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

Như chúng tôi đã nhận phân vùng không rỗng (nonEmptyPart), chúng tôi có thể làm sạch các phân vùng trống bằng cách sử dụng coalesce() (check coalesce() vs reparation()).

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

Nó có thể hoặc có thể không phải là tốt nhất, nhưng giải pháp này sẽ tránh xáo trộn như chúng ta không sử dụng reparation()

---

Ví dụ để giải quyết nhận xét

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