Come si inserisce una case case in un rdd e si comporta come una tupla (coppia)?

Question

diciamo per esempio, ho una semplice classe caso

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

Posso ottenere scintilla di riconoscere questo come una tupla ai fini di una cosa del genere, senza convertire ad una tupla in, dire una mappa o keyBy passo .

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

Non mi interessa nemmeno se perde la classe della custodia originale dopo un'operazione del genere. Ho provato il seguente senza fortuna. Sono abbastanza nuovo di Scala, mi manca qualcosa?

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

modifica: Nel codice sopra della classe caso estende Tuple2, questo non produce l'effetto desiderato che la classe e le funzioni RDD non trattano come una tupla e permettono PairRDDFunctions, come mapValues, valori, reduceByKey , ecc.

Answer 1

L'estensione TupleN non è una buona idea per una serie di motivi, con uno dei migliori è il fatto che è deprecato e in 2.11 non è nemmeno possibile estendere TupleN con una classe di case. Anche se rendi il tuo Foo una classe non-case, definendolo in 2.11 con -deprecation mostrerai questo: "avviso: l'ereditarietà della classe Tuple2 nel pacchetto scala è deprecata: le tuple saranno rese definitive in una versione futura.".

Se quello che interessa è la comodità di utilizzo e non ti dispiace il (quasi certamente trascurabile) sovraccarico di conversione ad una tupla, si può arricchire un RDD[Foo] con la sintassi fornita da PairRDDFunctions con una conversione del genere:

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

E poi:

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 ragione per la versione con Foo estensione Tuple2[String, (String, String)] non funziona è che RDD.rddToPairRDDFunctions targe ts an RDD[Tuple2[K, V]] e RDD non è covariante nel suo parametro tipo, quindi un RDD[Foo] non è un RDD[Tuple2[K, V]]. Un esempio più semplice potrebbe rendere più chiaro:

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) 

E poi:

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 
          ^

Ma se fate 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) 

... va tutto bene:

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

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

Non puoi make RDD covariant, quindi definire la propria conversione implicita per aggiungere la sintassi è la soluzione migliore. Personalmente probabilmente sceglierei di fare la conversione esplicitamente, ma questo è un uso relativamente orribile delle conversioni implicite.