如何在rdd中放置一个case类并让它像一个元组(对)？

Question

比方说,我有一个简单的案例类

case class Foo(k:String, v1:String, v2:String)

为了这样的目的,我可以获得火花来识别这个元组,而不是转换为元组,例如地图或keyBy步骤.

val rdd = sc.parallelize(List(Foo("k", "v1", "v2")))
// Swap values
rdd.mapValues(v => (v._2, v._1))

在这样的操作之后,我甚至不关心它是否会失去原来的案例类.我试过以下没有运气.我对Scala很新,我错过了什么吗？

case class Foo(k:String, v1:String, v2:String)
  extends Tuple2[String, (String, String)](k, (v1, v2))

编辑:在上面的代码片段中,case类扩展了Tuple2,这不会产生RDD类和函数不像元组那样对待并允许PairRDDFunctions,例如mapValues,values,reduceByKey等所需的效果.

Answer 1

TupleN由于许多原因,扩展并不是一个好主意,其中最好的一个是它被弃用的事实,而在2.11上,它甚至不可能TupleN用case类扩展.即使你创建Foo了一个非case类,在2.11上定义它-deprecation也会显示:"警告:不推荐使用包scala中的类Tuple2继承:在将来的版本中,元组将成为最终版本."

如果您关心的是使用的便利性,并且您不介意转换为元组的(几乎肯定可以忽略不计)开销,您可以使用以下转换RDD[Foo]提供的语法来丰富a PairRDDFunctions:

import org.apache.spark.rdd.{ PairRDDFunctions, RDD }

case class Foo(k: String, v1: String, v2: String)

implicit def fooToPairRDDFunctions[K, V]
  (rdd: RDD[Foo]): PairRDDFunctions[String, (String, String)] =
    new PairRDDFunctions(
      rdd.map {
        case Foo(k, v1, v2) => k -> (v1, v2)
      }
    )

然后:

scala> val rdd = sc.parallelize(List(Foo("a", "b", "c"), Foo("d", "e", "f")))
rdd: org.apache.spark.rdd.RDD[Foo] = ParallelCollectionRDD[6] at parallelize at <console>:34

scala> rdd.mapValues(_._1).first
res0: (String, String) = (a,b)

你的Foo扩展版本Tuple2[String, (String, String)]不起作用的原因是它的类型参数中的RDD.rddToPairRDDFunctions目标是RDD[Tuple2[K, V]]并且RDD不是协变的,所以a RDD[Foo]不是a RDD[Tuple2[K, V]].一个更简单的例子可能会使这更清楚:

case class Box[A](a: A)

class Foo(k: String, v: String) extends Tuple2[String, String](k, v)

class PairBoxFunctions(box: Box[(String, String)]) {
  def pairValue: String = box.a._2
}

implicit def toPairBoxFunctions(box: Box[(String, String)]): PairBoxFunctions =
  new PairBoxFunctions(box)

然后:

scala> Box(("a", "b")).pairValue
res0: String = b

scala> Box(new Foo("a", "b")).pairValue
<console>:16: error: value pairValue is not a member of Box[Foo]
       Box(new Foo("a", "b")).pairValue
                              ^

但如果你做Box协变......

case class Box[+A](a: A)

class Foo(k: String, v: String) extends Tuple2[String, String](k, v)

class PairBoxFunctions(box: Box[(String, String)]) {
  def pairValue: String = box.a._2
}

implicit def toPairBoxFunctions(box: Box[(String, String)]): PairBoxFunctions =
  new PairBoxFunctions(box)

…一切安好:

scala> Box(("a", "b")).pairValue
res0: String = b

scala> Box(new Foo("a", "b")).pairValue
res1: String = b

RDD但是,您无法进行协变,因此定义自己的隐式转换以添加语法是最好的选择.就个人而言,我可能会选择明确地进行转换,但这是对隐式转换的相对不可靠的使用.