Warum verwendet pmap | reducers/map nicht alle CPU-Kerne?

Question

Ich versuche, eine Datei mit einer Million Zeilen zu analysieren, jede Zeile ist eine JSON-Zeichenfolge mit einigen Informationen über ein Buch (Autor, Inhalt usw.). Ich verwende iota, um die Datei zu laden, da mein Programm OutOfMemoryError wirft, wenn ich versuche, slurp zu verwenden. Ich verwende auch cheshire, um die Zeichenfolgen zu analysieren. Das Programm lädt einfach eine Datei und zählt alle Wörter in allen Büchern.

Mein erster Versuch enthalten pmap, um die schwere Arbeit zu tun, ich dachte, das würde im Wesentlichen alle meine CPU-Kerne nutzen.

(ns multicore-parsing.core 
    (:require [cheshire.core :as json] 
      [iota :as io] 
      [clojure.string :as string] 
      [clojure.core.reducers :as r])) 


(defn words-pmap 
    [filename] 
    (letfn [(parse-with-keywords [str] 
      (json/parse-string str true)) 
      (words [book] 
      (string/split (:contents book) #"\s+"))] 
    (->> 
    (io/vec filename) 
    (pmap parse-with-keywords) 
    (pmap words) 
    (r/reduce #(apply conj %1 %2) #{}) 
    (count)))) 

Während es alle Kerne zu verwenden scheint, jeder Kern nur selten verwendet mehr als 50% seiner Kapazität, meine Vermutung ist, dass es mit Losgröße pmap zu tun hat, und so stolperte ich über relatively old question wo einige Kommentare Verweise auf die clojure.core.reducers Bibliothek.

habe ich beschlossen, die Funktion mit reducers/map neu zu schreiben:

(defn words-reducers 
    [filename] 
    (letfn [(parse-with-keywords [str] 
      (json/parse-string str true)) 
      (words [book] 
      (string/split (:contents book) #"\s+"))] 
    (->> 
    (io/vec filename) 
    (r/map parse-with-keywords) 
    (r/map words) 
    (r/reduce #(apply conj %1 %2) #{}) 
    (count)))) 

Aber die CPU-Auslastung ist schlimmer, und es dauert noch länger im Vergleich zu der vorherigen Ausführung zu beenden:

multicore-parsing.core=> (time (words-pmap "./dummy_data.txt")) 
"Elapsed time: 20899.088919 msecs" 
546 
multicore-parsing.core=> (time (words-reducers "./dummy_data.txt")) 
"Elapsed time: 28790.976455 msecs" 
546 

Was bin ich falsch machen? Ist mmap loading + die richtige Vorgehensweise beim Parsen einer großen Datei?

EDIT: this ist die Datei, die ich benutze.

EDIT2: Hier sind die Timings mit iota/seq statt iota/vec:

multicore-parsing.core=> (time (words-reducers "./dummy_data.txt")) 
"Elapsed time: 160981.224565 msecs" 
546 
multicore-parsing.core=> (time (words-pmap "./dummy_data.txt")) 
"Elapsed time: 160296.482722 msecs" 
546 

Answer 1

Ich glaube nicht, dass Reduzierungen für Sie die richtige Lösung sein werden, wie sie fertig werden nicht mit faulen Sequenzen an alles gut (ein Reducer wird korrekte Ergebnisse mit einer faulen Sequenz ergeben, aber wird nicht gut parallelisieren).

Sie können einen Blick auf diese sample code aus dem Buch Seven Concurrency Models in Seven Weeks nehmen wollen (Disclaimer: Ich bin der Autor), die ein ähnliches Problem löst (Zählen der Anzahl, wie oft jedes Wort auf Wikipedia erscheint).

eine Liste von Wikipedia-Seiten gegeben, zählt diese Funktion die Worte sequentiell (get-words gibt eine Sequenz von Wörtern von einer Seite):

(defn count-words-sequential [pages] 
    (frequencies (mapcat get-words pages))) 

Dies ist eine parallele Version pmap verwendet, die schneller laufen wird, aber nur um 1,5x schneller:

(defn count-words-parallel [pages] 
    (reduce (partial merge-with +) 
    (pmap #(frequencies (get-words %)) pages))) 

der Grund, es geht nur 1.5x um schneller ist, weil die reduce ein Engpass-es wird für jede Seite (partial merge-with +) einmal anruft. Zusammenführen von Chargen von 100 Seiten verbessert die Leistung auf rund 3.2x auf einer 4-Core-Maschine:

(defn count-words [pages] 
    (reduce (partial merge-with +) 
    (pmap count-words-sequential (partition-all 100 pages))))