Wie man mit einer Spark RDD arbeitet, um eine andere RDD

Question

zu erzeugen oder zuzuordnen Ich habe eine Schlüssel-/Wert-RDD Ich möchte das "iterieren" über die Entitäten, Schlüssel/Wert, und erstellen oder zuordnen, zu einem anderen RDD, die mehr oder weniger Einträge als die erste RDD haben könnte.

Beispiel:

Ich habe Aufzeichnungen in accumulo, die Beobachtungen von Farben in Bildern darstellen. Eine Observationsentität/Objekt enthält Daten zum Malnamen und den Farben im Gemälde.

Observation 
public String getPaintingName() {return paintingName;} 
public List<String> getObservedColors() {return colorList} 

ziehe ich die Beobachtungen von accumulo in meinen Code als RDD.

val observationRDD: RDD[(Text, Observation)] = getObservationsFromAccumulo(); 

Ich möchte diese RDD nehmen und eine RDD von der Form (Farbe, paintingName) zu schaffen, wo der Schlüssel ist die Farbe beobachtet und der Wert der Malerei Name, der die Farbe in beobachtet wurde.

val colorToPaintingRDD: RDD[(String, String)] = observationRDD.somefunction({ case (_, observation) => 
    for(String color : observations.getObservedColors()) { 
     // Some how output a entry into a new RDD 
     //output/map (color, observation.getPaintingName) 
}) 

Ich weiß Karte kann nicht funktionieren, weil seine 1 zu 1, dachte ich vielleicht observationRDD.flatmap (einige Funktion), aber kann nicht scheinen, irgendwelche Beispiele zu finden, wie man das schafft, um eine neue, größere oder kleinere zu schaffen , RDD.

Könnte mir jemand helfen und mir sagen, ob Flatmap korrekt ist, und wenn ja, gib mir ein Beispiel mit diesem Beispiel, das ich zur Verfügung gestellt habe, oder sag mir, ob ich weit von der Basis entfernt bin?

Bitte haben Sie Verständnis, dies ist nur ein einfaches Beispiel, es ist nicht der Inhalt, der fragt, wie man eine RDD zu einer RDD mit mehr oder weniger Einträge verwandeln würde.

Answer 1

Sie sollten Flatmap verwenden und ein List [(String, String)] foreach-Element in RDD zurückgeben. FlatMap wird flach das Ergebnis und Sie erhalten eine RDD erhalten [(String, String)]

ich den Code nicht versucht haben, aber es wäre so etwas wie dieses:

val colorToPaintingRDD: RDD[(String, String)] = observationRDD.flatMap { case (_, observation) => 
    observations.getObservedColors().map(color => (color, observation.getPaintingName)) 
} 

Wahrscheinlich wenn getObservedColors Methode In Java müssen Sie JavaConversions importieren und in die Scalaliste wechseln.

import scala.collection.JavaConversions._ 
observations.getObservedColors().toList