Равномерность класса в Apache Spark

Почему совпадение шаблонов в Spark не работает так же, как в Scala? См. Пример ниже... Функция f() пытается сопоставить шаблон по классу, который работает в Scala REPL, но не работает в Spark и приводит ко всем "???". f2() - это обходной путь, который получает желаемый результат в Spark с помощью .isInstanceOf(), но я понимаю, что это плохая форма в Scala.

Любая помощь по шаблону, соответствующему правильному пути в этом сценарии в Spark, будет с благодарностью.

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)

попытка сопоставления шаблонов, которая работает в Scala REPL, но не работает в Spark

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

который работает в Spark, но плохая форма (я думаю)

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

Используемые версии... Результаты искры запускаются в искровой оболочке (Spark 1.6 на AWS EMR-4.3) Scala REPL в SBT 0.13.9 (Scala 2.10.5)

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

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

Ответ 1