अपाचे स्पार्क

Question

में केस क्लास समानता स्पार्क में पैटर्न मिलान क्यों स्केल में समान नहीं है? नीचे उदाहरण देखें ... फ़ंक्शन f() क्लास पर पैटर्न मिलान करने का प्रयास करता है, जो स्कैला आरईपीएल में काम करता है लेकिन स्पार्क में विफल रहता है और परिणामस्वरूप "???" होता है। f2() एक कामकाज है जो .isInstanceOf() का उपयोग कर स्पार्क में वांछित परिणाम प्राप्त करता है, लेकिन मैं समझता हूं कि स्कैला में खराब फॉर्म होना है।

स्पार्क में इस परिदृश्य में सही तरीके से मिलान करने वाले पैटर्न पर कोई भी मदद की सराहना की जाएगी। पैटर्न मिलान में

abstract class a extends Serializable {val a: Int} 
case class b(a: Int) extends a 
case class bNull(a: Int=0) extends a 

val x: List[a] = List(b(0), b(1), bNull()) 
val xRdd = sc.parallelize(x) 

प्रयास जो स्काला आरईपीएल में काम करता है लेकिन स्पार्क में विफल रहता है

def f(x: a) = x match { 
    case b(n) => "b" 
    case bNull(n) => "bnull" 
    case _ => "???" 
} 

वैकल्पिक हल है कि स्पार्क में काम करता है, लेकिन बुरा रूप है (मुझे लगता है कि)

def f2(x: a) = { 
    if (x.isInstanceOf[b]) { 
     "b" 
    } else if (x.isInstanceOf[bNull]) { 
     "bnull" 
    } else { 
     "???" 
    } 
} 

परिणाम देखें

xRdd.map(f).collect     //does not work in Spark 
             // result: Array("???", "???", "???") 
xRdd.map(f2).collect     // works in Spark 
             // resut: Array("b", "b", "bnull") 
x.map(f(_))       // works in Scala REPL  
             // result: List("b", "b", "bnull") 

संस्करण इस्तेमाल किया ... स्पार्क परिणाम चिंगारी से खोल एसबीटी 0.13.9 में स्काला आरईपीएल (एडब्ल्यूएस ईएमआर-4.3 पर 1.6 स्पार्क) में चलने वाले (स्काला 2.10.5)

Answer 1

यह स्पार्क आरईपीएल साथ एक ज्ञात समस्या है। आप SPARK-2620 में अधिक जानकारी प्राप्त कर सकते हैं। यह स्पार्क आरईपीएल में कई परिचालनों को प्रभावित करता है जिनमें PairwiseRDDs पर अधिकांश परिवर्तन शामिल हैं। उदाहरण के लिए:

case class Foo(x: Int) 

val foos = Seq(Foo(1), Foo(1), Foo(2), Foo(2)) 
foos.distinct.size 
// Int = 2 

val foosRdd = sc.parallelize(foos, 4) 
foosRdd.distinct.count 
// Long = 4 

foosRdd.map((_, 1)).reduceByKey(_ + _).collect 
// Array[(Foo, Int)] = Array((Foo(1),1), (Foo(1),1), (Foo(2),1), (Foo(2),1)) 

foosRdd.first == foos.head 
// Boolean = false 

Foo.unapply(foosRdd.first) == Foo.unapply(foos.head) 
// Boolean = true 

क्या यह और भी बदतर बना देता है कि परिणाम डेटा वितरण पर निर्भर है:

sc.parallelize(foos, 1).distinct.count 
// Long = 2 

sc.parallelize(foos, 1).map((_, 1)).reduceByKey(_ + _).collect 
// Array[(Foo, Int)] = Array((Foo(2),2), (Foo(1),2)) 

सरल बात आप कर सकते हैं आरईपीएल बाहर परिभाषित करने और पैकेज की आवश्यकता मामले वर्गों है। spark-submit का उपयोग करके सीधे सबमिट किए गए किसी भी कोड को भी काम करना चाहिए।

स्कैला 2.11+ में आप paste -raw के साथ सीधे आरईपीएल में एक पैकेज बना सकते हैं।

scala> :paste -raw 
// Entering paste mode (ctrl-D to finish) 

package bar 

case class Bar(x: Int) 


// Exiting paste mode, now interpreting. 

scala> import bar.Bar 
import bar.Bar 

scala> sc.parallelize(Seq(Bar(1), Bar(1), Bar(2), Bar(2))).distinct.collect 
res1: Array[bar.Bar] = Array(Bar(1), Bar(2))