Почему pmap|redurs/map не использует все ядра процессора?

Question

Я пытаюсь проанализировать файл с миллионом строк, каждая строка представляет собой строку json с некоторой информацией о книге (автор, содержание и т. Д.). Я использую Йоту для загрузки файла, так как моя программа выдает OutOfMemoryError если я попытаюсь использовать slurp, Я также использую Чешир для разбора строк. Программа просто загружает файл и считает все слова во всех книгах.

Моя первая попытка включена pmap чтобы выполнить тяжелую работу, я решил, что в сущности будут использованы все мои ядра процессора.

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

Хотя кажется, что оно использует все ядра, каждое ядро ​​редко использует более 50% своей емкости, но я предполагаю, что это связано с размером пакета pmap, поэтому я наткнулся на относительно старый вопрос, где некоторые комментарии ссылаются на clojure.core.reducers библиотека.

Я решил переписать функцию с помощью 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))))

Но использование процессора хуже, и это займет больше времени по сравнению с предыдущей реализацией:

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

Что я делаю неправильно? Является ли загрузка mmap + редукторы правильным подходом при разборе большого файла?

РЕДАКТИРОВАТЬ: это файл, который я использую.

EDIT2: вот время с iota/seq вместо 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

Я не верю, что редукторы будут правильным решением для вас, так как они не справляются с ленивыми последовательностями вообще (редуктор даст правильные результаты с ленивыми последовательностями, но не будет хорошо распараллеливаться).

Возможно, вы захотите взглянуть на этот пример кода из книги " Семь моделей параллелизма за семь недель" (отказ от ответственности: я являюсь автором), в которой решается аналогичная проблема (подсчитывается, сколько раз каждое слово появляется в Википедии).

Учитывая список страниц Википедии, эта функция считает слова последовательно (get-words возвращает последовательность слов со страницы):

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

Это параллельная версия, использующая pmap который работает быстрее, но только примерно в 1,5 раза быстрее:

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

Причина, по которой он увеличивается примерно в 1,5 раза, заключается в том, что reduce становится узким местом - это зовет (partial merge-with +) один раз для каждой страницы. Объединение пакетов по 100 страниц повышает производительность примерно в 3,2 раза на 4-ядерном компьютере:

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