Dlaczego reduktory/mapa pmap nie używa wszystkich rdzeni procesora?

Question

Próbuję parsować plik z milionem linii, każda linia jest ciągiem json z pewnymi informacjami o książce (autor, zawartość itp.). Używam iota do załadowania pliku, ponieważ mój program zgłasza OutOfMemoryError, jeśli spróbuję użyć slurp. Używam również cheshire do parsowania ciągów. Program po prostu wczytuje plik i zlicza wszystkie słowa we wszystkich książkach.

Moja pierwsza próba obejmowała pmap, aby wykonać ciężką pracę, doszedłem do wniosku, że zasadniczo wykorzystałbym wszystkie moje rdzenie procesora.

(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)))) 

Choć wydaje się wykorzystać wszystkie rdzenie, każdy rdzeń rzadko używa się więcej niż 50% swojej pojemności, domyślam się, że ma do czynienia z wielkości partii pmap i tak natknąłem relatively old question gdzie niektóre komentarze odwołaj się do biblioteki clojure.core.reducers.

postanowiłem przerobić funkcję używając reducers/map:

(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)))) 

Ale użycie procesora jest gorzej, a jeszcze trwa dłużej, aby zakończyć w porównaniu do poprzedniej realizacji:

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 

co ja robić źle? Czy ładowanie mmap + reduktory jest właściwym podejściem podczas analizowania dużego pliku?

EDYTOWANIE: this to plik, którego używam.

EDIT2: Oto czasy z iota/seq zamiast 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

Nie wierzę, że redukcje będą dobrym rozwiązaniem dla Ciebie, ponieważ nie radzą sobie z sekwencjami w leniwych wszystko dobrze (reduktor da prawidłowe wyniki z leniwą sekwencją, ale nie będzie dobrze paralelezy).

Możesz zajrzeć do tego z książki Seven Concurrency Models in Seven Weeks (disclaimer: Jestem autorem), która rozwiązuje podobny problem (licząc ile razy każde słowo pojawia się w Wikipedii).

otrzymali listę stron Wikipedii, funkcja ta liczy słowa sekwencyjnie (get-words zwraca sekwencję słów ze strony):

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

To równoległą wersję używając pmap które działa szybciej, ale tylko około 1,5x szybciej:

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

powodem tylko krąży wokół 1,5x szybciej dlatego, że reduce staje się wąskim gardłem, to dzwoni (partial merge-with +) raz dla każdej strony. Łączenie partii 100 stron poprawia wydajność około 3,2x na maszynie 4-core:

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