गिराने अपाचे स्पार्क में खाली DataFrame विभाजन

Question

मैं एक columnm के अनुसार एक DataFrame के पुनर्विभाजन की कोशिश DataFrame है N विभिन्न मूल्यों विभाजन स्तंभ x में जैसे (N=3 मान लीजिए),:

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

क्या मैं प्राप्त करने के लिए myDFx द्वारा रिक्त विभाजन तैयार किए बिना पुनः प्राप्त करना है। क्या ऐसा करने का कोई बेहतर तरीका है?

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

(मैं repartiton में numParts निर्दिष्ट नहीं करते हैं, मेरे विभाजन के सबसे खाली हैं (जैसा कि repartition 200 विभाजन बनाता है) ...)

Answer 1

मैं df से अधिक बार-बार दोहराना के साथ समाधान के लगता है कि विभाजन और गैर-खाली विभाजन खोजने के लिए में रिकॉर्ड गिनती प्राप्त करना।

val nonEmptyPart = sparkContext.longAccumulator("nonEmptyPart") 

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

हम गैर खाली विभाजन (nonEmptyPart) मिल गया के रूप में, हम coalesce() (check coalesce() vs reparation()) का उपयोग करके खाली विभाजन साफ ​​कर सकते हैं।

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

यह कर सकते हैं या सबसे अच्छा नहीं हो सकता है, लेकिन इस समाधान हम उपयोग नहीं कर रहे के रूप में reparation()

---

उदाहरण टिप्पणी को संबोधित करने के

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}") 

आउटपुट फेरबदल से बचने जाएगा

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