Fallanweisungen in Spark

Question

Ich schreibe einen Spark-Code, wo ich eine RDD des Typs (String,(String,String)) in ((String,String),String konvertieren muss).

Ich habe folgende Eingabetextdatei:

Language,Language-code,TotalViewsInThatLang 
English,en,10965376,"Main_Page",2938355 
Russian,ru,1925718,"%D0%97%D0%B0%D0%B3%D0%BB,915495 
Spanish,es,1010810,"Wikipedia:Portada",13603 

ich eine RDD erstellt haben, wie folgt:

val line = sc.textFile(inputFile) 
val nrdd = line.map(x=>(x.split(",")(0),(x.split(",")(1),x.split(",")(2)))) 
nrdd: org.apache.spark.rdd.RDD[(String, (String, String))] = MapPartitionsRDD[2] at map at <console>:26 

Daraus ich case Funktion erstellen RDD vom Typ ((String,String),String) verwenden möchten. Wie kann ich das mit case Anweisungen in map tun?
EDIT
ich die folgende Störung erhalte, wenn ich Fall Funktion ist versucht zu verwenden:

scala> val frdd = nrdd.map({case(x,(y,z))=>((x,y),z))}) 
<console>:1: error: ';' expected but ')' found. 
val frdd = nrdd.map({case(x,(y,z))=>((x,y),z))}) 
              ^  

Answer 1

Wenn ich Ihre Frage falsch verstanden, wollen Sie dies:

val list: List[((String, String), String)] = List((("a1", "b1"), "c1"), (("a2", "b2"), "c2")) 
val res = list.map { case ((a, b), c) => (a, (b, c)) } 

println(res) // List((a1,(b1,c1)), (a2,(b2,c2))) 

Answer 2

Da Ihr RDD Pair RDD, könnten Sie swap von Keyed RDD verwenden.

Beispielcode:

val keyRDD = sc.parallelize(List((("a1", "b1"), "c1"), (("a2", "b2"), "c2")), 2) 
val swappedRDD = keyRDD.map(_.swap) 
swappedRDD.foreach(x => println(x))