Wie zu verwenden mapPartitions Zündkerzen Scala?

Habe ich DocsRDD : RDD[String, String]

val DocsRDD = sc.wholeTextFiles("myDirectory/*" , 2)

DocsRDD:

Doc1.txt , bla bla bla .....\n bla bla bla \n bla ... bla
Doc2.txt , bla bla bla .....bla \n bla bla \n bla ... bla
Doc3.txt , bla bla bla .....\n bla bla bla \n bla ... bla
Doc4.txt , bla bla \n  .....\n bla bla bla bla \n ... bla

Gibt es eine effiziente, elegante Art, Extrakt n-Gramm von diesen mit mapPartitions?
So weit, ich habe alles versucht, ich habe alles gelesen, was ich finden konnte, mindestens 5 mal über und über, über mapPartitions aber ich kann immer noch nicht verstehen, wie es zu benutzen! Es scheint waaay zu schwierig zu manipulieren.
Kurz gesagt: ich will :

val NGramsRDD = DocsRDD.map(x => (x._1 , x._2.sliding(n) ) )

aber effizient mit mapPartitions.
Mein grundlegendes Missverständnis von mapPartitions ist :

OneDocRDD : RDD[String]

 val OneDocRDD = sc.textFile("myDoc1.txt" , 2)
                   .mapPartitions(s1 : Iterator[String] => s2 : Iterator[String])

Kann ich Nicht verstehen! Aus, wenn s1 wurde Iterator[String]? s1 ist der String nach dem sc.Textdatei.

Okay, meine zweite Frage ist : Wird mapPartitions verbessern meine überwinden, gegen map in dieser situation?

Letztes aber nicht am Wenigsten wichtig:
kann f() sein :

     f(Iterator[String]) : Iterator[Something else?]
  • Ihr Aufruf sc.textFile gibt Ihnen einen RDD[String] mit 2 Partitionen. Jedes element in der RDD ist eine Zeile aus der text-Datei. mapPartitions bietet Ihnen einen iterator über alle Zeilen in jeder partition und geben Sie eine Funktion angewendet werden, um jeder dieser Iteratoren. Du bist wieder erwartet einen iterator, der ist dann abgeflacht, wieder in einen RDD.
  • Danke für die Antwort. Dies löscht mir irgendwie das vage Konzept über mapPartitions.
InformationsquelleAutor Spartan | 2016-11-30
  1. 9

    Ich bin mir nicht sicher, dass .mapPartitions helfen (zumindest nicht zum Beispiel), aber mit .mapPartitions würde wie folgt Aussehen: 

    val OneDocRDD = sc.textFile("myDoc1.txt", 2)
  .mapPartitions(iter => {
    //here you can initialize objects that you would need 
    //that you want to create once by worker and not for each x in the map. 
    iter.map(x => (x._1 , x._2.sliding(n)))
  })

    Regel, die Sie verwenden möchten .mapPartitions erstellen/initialisieren eines Objekts, die Sie nicht wollen (Beispiel: zu groß) oder kann nicht serialisiert werden, um den worker-Knoten. Ohne .mapPartitions würden Sie brauchen, um Sie zu schaffen in der .anzeigen, aber das wäre nicht effizient, da das Objekt erstellt werden, für jedes x.

    • Danke für die Antwort! Es disambiguated mir einige unscharfe Wahrnehmung über mapPartitions.
    • Cool! Ich werde diese Methode verwenden, um zu ersetzen, eine Sendung, die aus irgendeinem Grund kryo nicht serialisieren - ich bin gerade dabei zu downloaden, was ich brauchen innerhalb mapPartition eher, dass download der Treiber dann broadcast.
    • Sie verlieren einige der ngrams im Allgemeinen in Grenzen.
    InformationsquelleAutor Pascal Soucy
Schreibe einen Kommentar