Schlüssel/Wert-Paar RDD

Question

Ich habe eine Frage zu Schlüssel/Wert-Paar RDD.

Ich habe fünf Dateien im C:/download/input Ordner, der die Dialoge in den Filmen wie der Inhalt der Dateien hat, wie folgt:

movie_horror_Conjuring.txt 
movie_comedy_eurotrip.txt 
movie_horror_insidious.txt 
movie_sci-fi_Interstellar.txt 
movie_horror_evildead.txt 

Ich versuche, die Dateien im Eingabeordner mit dem sc zu lesen. wholeTextFiles(), wo ich den Schlüssel/Wert erhalten als

(C:/download/input/movie_horror_Conjuring.txt,values) 

folgt ich versuche, eine Operation zu tun, wo ich die Eingabedateien jedes Genre zusammen mit groupByKey() zu einer Gruppe haben. Die Werte aller Horrorfilme zusammen, Comedy-Filme zusammen und so weiter.

Gibt es eine Möglichkeit, die Schlüssel/Wert-Paar auf diese Weise erzeugen kann (horror, values) statt (C:/download/input/movie_horror_Conjuring.txt,values)

val ipfile = sc.wholeTextFiles("C:/download/input") 
val output = ipfile.groupByKey().map(t => (t._1,t._2)) 

Der obige Code mir den Ausgang geben als

(C:/download/input/movie_horror_Conjuring.txt,values) 
(C:/download/input/movie_comedy_eurotrip.txt,values) 
(C:/download/input/movie_horror_Conjuring.txt,values) 
(C:/download/input/movie_sci-fi_Interstellar.txt,values) 
(C:/download/input/movie_horror_evildead.txt,values) 

folgt, wo, wie ich brauche, die Ausgabe wie folgt:

(horror, (values1, values2, values3)) 
(comedy, (values1)) 
(sci-fi, (values1)) 

Ich habe auch versucht Führen Sie einige Map- und Split-Operationen aus, um die Ordnerpfade des Schlüssels zu entfernen, um nur den Dateinamen zu erhalten, aber ich kann die entsprechenden Werte nicht an die Dateien anhängen.

Auch würde ich gerne wissen, wie kann ich die Zeilen zählen in Werten1 bekommen, Werte2, Werte3 usw.

Mein sollte die endgültige Ausgabe sein wie

(Horror, 100)

Dabei ist 100 die Summe der Anzahl der Zeilen in Werten1 = 40 Zeilen, Werte2 = 30 Zeilen und Werte3 = 30 Zeilen usw.

Answer 1

Versuchen Sie Folgendes:

val output = ipfile.map{case (k, v) => (k.split("_")(1),v)}.groupByKey()  
output.collect 

Lassen Sie mich wissen, ob dies für Sie funktioniert!

Update:

Zur Ausgabe im Format (horror, 100) zu erhalten:

val output = ipfile.map{case (k, v) => (k.split("_")(1),v.count(_ == '\n'))}.reduceByKey(_ + _)  
output.collect