scala - que - spark ejemplo

¿Cómo pongo una clase de caso en un rdd y hago que actúe como una tupla(par)?

Question

Digamos, por ejemplo, tengo una clase de caso simple

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

¿Puedo obtener chispa para reconocer esto como una tupla con el propósito de algo como esto, sin convertir a una tupla en, digamos un mapa o una tecla Por paso.

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

Ni siquiera me importa si pierde la clase de caja original después de tal operación. Intenté lo siguiente sin suerte. Soy bastante nuevo para Scala, ¿me estoy perdiendo algo?

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

editar: en el fragmento anterior, la clase caso extiende Tuple2, esto no produce el efecto deseado de que la clase RDD y las funciones no lo tratan como una tupla y permiten PairRDDFunctions, como mapValues, values, reduceByKey, etc.

Answer 1

Extender TupleN no es una buena idea por varias razones, una de las mejores es el hecho de que está obsoleto, y en 2.11 ni siquiera es posible extender TupleN con una clase de caso. Incluso si convierte su Foo una clase no -deprecation , definirla en 2.11 con -deprecation le mostrará esto: "warning: la herencia de la clase Tuple2 en el paquete scala está en desuso: Tuples se hará definitiva en una versión futura".

Si lo que te importa es la comodidad de uso y no te importa la sobrecarga (casi sin duda) de la conversión a una tupla, puedes enriquecer un RDD[Foo] con la sintaxis proporcionada por PairRDDFunctions con una conversión como esta:

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

Y entonces:

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)

La razón por la cual su versión con Foo extiende Tuple2[String, (String, String)] no funciona es que RDD.rddToPairRDDFunctions dirige a un RDD[Tuple2[K, V]] y el RDD no es covariante en su parámetro de tipo, por lo que RDD[Foo] no es un RDD[Tuple2[K, V]] . Un ejemplo más simple podría aclarar esto:

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)

Y entonces:

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
                              ^

Pero si haces Box covariante ...

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)

…todo está bien:

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

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

Sin embargo, no puede hacer que RDD covariante, por lo que definir su propia conversión implícita para agregar la sintaxis es su mejor opción. Personalmente, probablemente elegiría hacer la conversión explícitamente, pero este es un uso relativamente poco horrible de las conversiones implícitas.