Distributed Word2Vec Model Training mit Apache Spark 2.0.0 und MLLIB

Question

Ich experimentierte mit Funken und MLLIB, um ein Word2Vec-Modell zu trainieren, aber ich habe nicht die Leistungsvorteile des verteilten maschinellen Lernens auf großen Datensätzen. Mein Verständnis ist, dass, wenn ich w Worker habe, dann, wenn ich eine RDD mit n Anzahl von Partitionen erstellen, wo n> w und ich versuche, ein Word2Vec Modell durch Aufrufen der Fit-Funktion von Word2Vec mit der RDD als Parameter zu erstellen, würde Funke verteilen die Daten einheitlich zu trainieren Word2vec Modelle auf diese w Worker und verwenden Sie eine Art von Reduzierer-Funktion am Ende, um ein einzelnes Ausgabemodell von diesen w-Modellen zu erstellen. Dies würde die Berechnungszeit verringern, statt 1 Chunk, w Chunks von Daten werden gleichzeitig verarbeitet. Der Kompromiss besteht darin, dass je nach der am Ende verwendeten Reduzierfunktion ein gewisser Präzisionsverlust auftreten könnte. Funktioniert Word2Vec in Spark tatsächlich auf diese Weise oder nicht? Ich muss möglicherweise mit den konfigurierbaren Parametern spielen, wenn dies tatsächlich der Fall ist.

EDIT

Hinzufügen der Grund, diese Frage hinter fragen. Ich habe java function word2vec code auf 10 Arbeitsmaschinen ausgeführt und geeignete Werte für executor-memory, driver memory und num-executors, nach Durchlaufen der Dokumentation, für eine 2.5gb Eingabetextdatei gesetzt, die auf rdd-Partitionen gemappt wurde, die dann als verwendet wurden Trainingsdaten für ein mllib word2vec-Modell. Der Trainingsteil dauerte mehrere Stunden. Die Anzahl der Arbeiterknoten scheint keinen großen Einfluss auf die Trainingszeit zu haben. Der gleiche Code läuft erfolgreich auf kleinere Datendateien (in der Größenordnung von 10s von MBs)

-Code

SparkConf conf = new SparkConf().setAppName("SampleWord2Vec"); 
conf.set("spark.serializer", "org.apache.spark.serializer.KryoSerializer"); 
conf.registerKryoClasses(new Class[]{String.class, List.class}); 
JavaSparkContext jsc = new JavaSparkContext(conf); 
JavaRDD<List<String>> jrdd = jsc.textFile(inputFile, 3).map(new Function<String, List<String>>(){    
     @Override 
     public List<String> call(String s) throws Exception { 
      return Arrays.asList(s.split(",")); 
     }   
}); 
jrdd.persist(StorageLevel.MEMORY_AND_DISK()); 
Word2Vec word2Vec = new Word2Vec() 
     .setWindowSize(20) 
     .setMinCount(20); 

Word2VecModel model = word2Vec.fit(jrdd); 
jrdd.unpersist(false); 
model.save(jsc.sc(), outputfile); 
jsc.stop(); 
jsc.close(); 

Answer 1

Gemessen an den Kommentaren, Antworten und downvotes konnte ich meine Frage nicht richtig darstellen. Aber die Antwort auf das, was ich wissen wollte, ist ja, es ist möglich, Ihr word2vec-Modell parallel auf Spark zu trainieren. Die Pull-Anforderung für diese Funktion wurde wieder lange Zeit erstellt:

https://github.com/apache/spark/pull/1719

In Java gibt es eine Setter-Methode (setNumPartitions) für das Word2Vec Objekt in Funken mllib. Dadurch können Sie Ihr word2vec-Modell parallel auf mehreren Executoren trainieren. Gemäß den Kommentaren über den Antrag Pull oben erwähnt:

"Um unsere Implementierung skalierbarer zu machen, wir jede Partition separat und verschmelzen das Modell jeder Partition nach jeder Iteration trainieren, um das Modell genauer, mehr Iterationen zu machen. kann benötigt werden. "

Hoffe, das hilft jemandem.

Answer 2

Ich sehe nichts von Natur aus falsch mit Ihrem Code. Ich würde jedoch sehr empfehlen, dass Sie die API für Datenrahmen berücksichtigen. Als Beispiel ist hier ein kleines Diagramm, das häufig herumgeworfen wird:

Auch ich weiß nicht, wie man „iterieren“ über die Elemente des Datenrahmens kann worden sind (das nicht wirklich ist, wie sie Arbeit). Hier ist ein Beispiel aus dem Spark online docs:

Sie haben die allgemeine Idee ... aber Sie haben Ihre Daten als Datenrahmen parallelisieren, zuerst. Es ist ziemlich trivial, stattdessen Ihre Javardd in einen DataFrame zu übersetzen.

DataFrame fileDF = sqlContext.createDataFrame(jrdd, Model.class); 

Funken läuft ein gerichteter azyklischer Graph (DAG) anstelle von MR, aber das Konzept ist das gleiche. Running 'fit() auf Ihre Daten wird in der Tat über die Daten auf den Arbeitern laufen und dann auf ein einziges Modell reduzieren. Aber dieses Modell wird selbst im Speicher verteilt sein, bis Sie sich entscheiden, es aufzuschreiben.

Wie lange dauert es jedoch, bis Sie dieselbe Datei über die native C++ - Binärdatei NLTK oder Word2Vec ausführen?

Ein letzter Gedanke ... Gibt es einen Grund, warum Sie Speicher und Festplatte beibehalten? Spark hat einen nativen .cache(), der standardmäßig im Arbeitsspeicher verbleibt. Die Stärke von Spark besteht darin, maschinelles Lernen mit Daten im Speicher durchzuführen ... BIG-Daten im Speicher. Wenn Sie auf der Festplatte bleiben, erzeugen Sie selbst bei kryo einen Flaschenhals bei der Festplatten-I/O. IMHO wäre das erste, was zu versuchen wäre, dies loszuwerden und nur in Erinnerung zu bleiben. Wenn die Leistung sich verbessert, ist das großartig, aber Sie werden Sprünge und Grenzen der Leistung finden, indem Sie sich auf die Kraft von Catalyst über DataFrames verlassen.

Eine Sache, die wir nicht besprochen haben, ist Ihr Cluster. Es wäre hilfreich, über Dinge nachzudenken, wie viel Speicher pro Knoten Sie haben ... wie viele Kerne pro Knoten ... ist Ihr Cluster mit anderen Apps virtualisiert, die nach Ressourcen fragen (überprovisioniert wie die meisten vHosts) ... Dein Cluster in der Cloud? Geteilt oder gewidmet?

Haben Sie sich Sparks UI angesehen, um die Laufzeitoperationen des Codes zu analysieren? Was sehen Sie, wenn Sie top auf die Arbeiter laufen lassen, während das Modell passt? Kannst du die volle CPU Auslastung sehen? Haben Sie versucht, --executor-cores anzugeben, um sicherzustellen, dass Sie die CPU voll auslasten?

Ich habe gesehen, dass es viele Male vorkommt, dass die ganze Arbeit an einem Kern auf einem Arbeiterknoten erledigt wird. Es wäre hilfreich, diese Informationen zu haben.

Bei der Fehlersuche der Leistung gibt es viele Orte zu sehen, einschließlich der Spark-Konfigurationsdateien selbst!