Apache Spark中的案例类相等

Question

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结果在spark-shell中运行（在AWS EMR-4.3上为Spark 1.6）在SBT 0.13.9中为Scala REPL（Scala 2.10.5）

Answer 1

这是Spark REPL的已知问题。您可以在SPARK-2620中找到更多详细信息。它会影响Spark REPL中的多种操作，包括上的大多数转换PairwiseRDDs。例如：case class Foo(x: Int)val foos = Seq(Foo(1), Foo(1), Foo(2), Foo(2))foos.distinct.size// Int = 2val 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 = falseFoo.unapply(foosRdd.first) == Foo.unapply(foos.head)// Boolean = true更糟糕的是结果取决于数据分布：sc.parallelize(foos, 1).distinct.count// Long = 2sc.parallelize(foos, 1).map((_, 1)).reduceByKey(_ + _).collect// Array[(Foo, Int)] = Array((Foo(2),2), (Foo(1),2))您可以做的最简单的事情是在REPL之外定义和打包所需的案例类。直接使用提交的任何代码也spark-submit应该起作用。在Scala 2.11+中，您可以使用直接在REPL中创建一个包paste -raw。scala> :paste -raw// Entering paste mode (ctrl-D to finish)package barcase class Bar(x: Int)// Exiting paste mode, now interpreting.scala> import bar.Barimport bar.Barscala> sc.parallelize(Seq(Bar(1), Bar(1), Bar(2), Bar(2))).distinct.collect res1: Array[bar.Bar] = Array(Bar(1), Bar(2))

Answer 2

我看到提到模式匹配在那儿的spark壳中不起作用，但是没有细节...您是说如果我在jar中定义case类，我将能够在REPL中对它们进行模式匹配？再次感谢