element sınıfına göre kıvılcım ayırıcı filtre

Question

Farklı türden elemanlara sahip bir RDD var ve bunları türlerine göre saymak istiyorum, örneğin aşağıdaki kod düzgün çalışacaktır.

scala> val rdd = sc.parallelize(List(1, 2.0, "abc")) 
rdd: org.apache.spark.rdd.RDD[Any] = ParallelCollectionRDD[0] at parallelize at <console>:24 

scala> rdd.filter{case z:Int => true; case _ => false}.count 
res0: Long = 1 

scala> rdd.filter{case z:String => true; case _ => false}.count 
res1: Long = 1 

Şimdi, öğeler kullanıcı tanımlı türden ise, aşağıdaki kod beklendiği gibi çalışmayacaktır.

scala> class TypeA extends Serializable    // this is the base class 
defined class TypeA 

scala> case class TypeB(id:Long) extends TypeA  // derived class 1 
defined class TypeB 

scala> case class TypeC(name:String) extends TypeA // derived class 2 
defined class TypeC 

scala> val rdd1 = sc.parallelize(List(TypeB(123), TypeC("jack"), TypeB(456))) // create an rdd with different types of elements 
rdd1: org.apache.spark.rdd.RDD[TypeA with Product] = ParallelCollectionRDD[3] at parallelize at <console>:29 

scala> rdd1.count    // total size is correct 
res2: Long = 3 

scala> rdd1.filter{case z:TypeB => true; case _ => false}.count // what the hell? 
res3: Long = 0 

scala> rdd1.filter{case z:TypeC => true; case _ => false}.count // again ? 
res4: Long = 0 

scala> rdd1.filter{case z:TypeA => true; case _ => false}.count // only works for the base class? 
res5: Long = 3 

Burada bir şey özledim mi? Yardım lütfen!

Answer 1

Bu, Spark-1199 bir varyasyonu gibi görünüyor ve muhtemelen bir REPL hatası.

Bu verimleri beklenen davranış IDEA içinde yerel olarak çalışan:

import org.apache.spark.SparkContext 

class TypeA extends Serializable 
case class TypeB(id:Long) extends TypeA 
case class TypeC(name:String) extends TypeA 

val sc = new SparkContext("local[*]", "swe") 
val rdd = sc.parallelize(List(TypeB(12), TypeC("Hsa"))) 

rdd.filter { case x: TypeB => true; case _ => false }.count() 

Verim: Nitekim

import org.apache.spark.SparkContext 

defined class TypeA 
defined class TypeB 
defined class TypeC 

sc: org.apache.spark.SparkContext = org.apache.spark.SparkContext@10a1410d 
rdd: org.apache.spark.rdd.RDD[TypeA with Product] = ParallelCollectionRDD[0] at parallelize at <console>:18 

[Stage 0:>....... (0 + 0)/4] 
res0: Long = 1