स्पार्क "कोड जेनरेटर: संकलित करने में विफल" Dataset.groupByKey

Question

मैं स्कैला और स्पार्क दोनों के लिए नया हूं, इसलिए उम्मीद है कि कोई मुझे बता सकता है कि मैं यहां गलत कहां जा रहा हूं।

मेरे पास तीन कॉलम डेटासेट (आईडी, नाम, वर्ष) है और मैं प्रत्येक नाम के लिए सबसे हालिया वर्ष खोजना चाहता हूं। दूसरे शब्दों में:

BEFORE           AFTER 
| id_1 | name_1 | 2015 |      | id_2 | name_1 | 2016 | 
| id_2 | name_1 | 2016 |      | id_4 | name_2 | 2015 | 
| id_3 | name_1 | 2014 | 
| id_4 | name_2 | 2015 | 
| id_5 | name_2 | 2014 | 

मैंने सोचा था कि groupByKey और reduceGroups मिल काम किया जाएगा:

val latestYears = ds 
    .groupByKey(_.name) 
    .reduceGroups((left, right) => if (left.year > right.year) left else right) 
    .map(group => group._2) 

लेकिन यह इस त्रुटि देता है, और उत्पन्न जावा कोड का एक बहुत बाहर थूक:

ERROR CodeGenerator: failed to compile: 
org.codehaus.commons.compiler.CompileException: 
File 'generated.java', Line 21, Column 101: Unknown variable or type "value4" 

दिलचस्प बात यह है कि, यदि मैं केवल नाम और वर्ष कॉलम के साथ एक डेटासेट बना देता हूं, तो यह अपेक्षा के अनुसार काम करता है।

object App { 

    case class Record(id: String, name: String, year: Int) 

    def main(args: Array[String]) { 
    val spark = SparkSession.builder().master("local").appName("test").getOrCreate() 
    import spark.implicits._ 

    val ds = spark.createDataset[String](Seq(
     "id_1,name_1,2015", 
     "id_2,name_1,2016", 
     "id_3,name_1,2014", 
     "id_4,name_2,2015", 
     "id_5,name_2,2014" 
    )) 
     .map(line => { 
     val fields = line.split(",") 
     new Record(fields(0), fields(1), fields(2).toInt) 
     }) 

    val latestYears = ds 
     .groupByKey(_.name) 
     .reduceGroups((left, right) => if (left.year > right.year) left else right) 
     .map(group => group._2) 

    latestYears.show() 
    } 


} 

---

संपादित करें: मेरा मानना ​​है कि इस स्पार्क v2.0.1 के साथ एक बग हो सकता है

---

यहाँ पूर्ण कोड मैं चल रहा हूँ है। V2.0.0 में डाउनग्रेड करने के बाद, यह अब नहीं होता है।

Answer 1

आपका groupBy और reduceGroups फ़ंक्शन experimental हैं। reduceByKey (api) का उपयोग क्यों न करें?

सकारात्मक:

• यह कोड आपके पास से अनुवाद करने के लिए आसान होना चाहिए।
• यह अधिक स्थिर है (प्रयोगात्मक नहीं)।
• यह अधिक कुशल होना चाहिए क्योंकि इसे प्रत्येक समूह में सभी वस्तुओं के पूर्ण शफल की आवश्यकता नहीं होती है (जो नेटवर्क I/O धीमा हो सकता है और एक नोड में स्मृति को ओवरफ़्लो कर सकता है)।