Почему этот параллельный поиск и замена не использует 100% ЦП?

Question

У меня очень длинный список твитов (2 миллиона), и я использую регулярные выражения для поиска и замены текста в этих твитах.

Я запускаю его, используя joblib. Параллельная карта (joblib - это параллельный бэкэнд, используемый scikit-learn).

Моя проблема в том, что в диспетчере задач Windows я вижу, что мой скрипт не использует 100% каждого ЦП.Он не использует 100% оперативной памяти и диска.Поэтому я не понимаю, почему это не пойдет быстрее.

Возможно, где-то есть задержки синхронизации, но я не могу найти ни то, ни другое.

Код:

# file main.py
import re
from joblib import delayed, Parallel

def make_tweets():
    tweets = load_from_file()  # this is list of strings

    regex = re.compile(r'a *a|b *b')  # of course more complex IRL, with lookbehind/forward
    mydict = {'aa': 'A', 'bb': 'B'}  

    def handler(match):
        return mydict[match[0].replace(' ', '')]

    def replace_in(tweet)
        return re.sub(regex, handler, tweet)

    # -1 mean all cores
    # I have 6 cores that can run 12 threads
    with Parallel(n_jobs=-1) as parallel:
        tweets2 = parallel(delayed(replace_in)(tweet) for tweet in tweets)

    return tweets2

А вот диспетчер задач:

---

Редактировать : последнее слово

Ответ заключается в том, что рабочие процессы были замедлены синхронизацией joblib: joblib отправляет твиты небольшими порциями (один за другим?) Рабочим, что заставляет их ждать.Использование multiprocessing.Pool.map с размером фрагмента len(tweets)/cpu_count() заставило рабочих использовать 100% ЦП.

При использовании joblib время работы составило около 12 мин.При многопроцессорной обработке это 4мн.При multiprocessing каждый рабочий поток занимал около 50 МБ памяти.

Answer 1

после небольшой игры, я думаю, это потому, что joblib тратит все свое время на координацию параллельного запуска всего, а не на выполнение какой-либо полезной работы.по крайней мере, для меня под OSX и Linux - у меня нет машин с MS Windows

Я начал с загрузки пакетов, загрузки вашего кода и создания фиктивного файла:

from random import choice
import re

from multiprocessing import Pool
from joblib import delayed, Parallel

regex = re.compile(r'a *a|b *b')  # of course more complex IRL, with lookbehind/forward
mydict = {'aa': 'A', 'bb': 'B'}  

def handler(match):
    return mydict[match[0].replace(' ', '')]

def replace_in(tweet):
    return re.sub(regex, handler, tweet)

examples = [
    "Regex replace isn't that computationally expensive... I would suggest using Pandas, though, rather than just a plain loop",
    "Hmm I don't use pandas anywhere else, but if it makes it faster, I'll try! Thanks for the suggestion. Regarding the question: expensive or not, if there is no reason for it to use only 19%, it should use 100%"
    "Well, is tweets a generator, or an actual list?",
    "an actual list of strings",
    "That might be causing the main process to have the 419MB of memory, however, that doesn't mean that list will be copied over to the other processes, which only need to work over slices of the list",
    "I think joblib splits the list in roughly equal chunks and sends these chunks to the worker processes.",
    "Maybe, but if you use something like this code, 2 million lines should be done in less than a minute (assuming an SSD, and reasonable memory speeds).",
    "My point is that you don't need the whole file in memory. You could type tweets.txt | python replacer.py > tweets_replaced.txt, and use the OS's native speeds to replace data line-by-line",
    "I will try this",
    "no, this is actually slower. My code takes 12mn using joblib.parallel and for line in f_in: f_out.write(re.sub(..., line)) takes 21mn. Concerning CPU and memory usage: CPU is same (17%) and memory much lower (60Mb) using files. But I want to minimize time spent, not memory usage.",
    "I moved this to chat because StackOverflow suggested it",
    "I don't have experience with joblib. Could you try the same with Pandas? pandas.pydata.org/pandas-docs/…",
]

with open('tweets.txt', 'w') as fd:
    for i in range(2_000_000):
        print(choice(examples), file=fd)

(посмотрите, можете ли вы угадать, откуда я взял строки!)

в качестве базовой линии, я попытался использовать наивное решение:

with open('tweets.txt') as fin, open('tweets2.txt', 'w') as fout:
    for l in fin:
        fout.write(replace_in(l))

это занимает 14.0s (время настенных часов) на моемНоутбук OSX и 5.15 на моем рабочем столе Linux.Обратите внимание, что изменение вашего определения replace_in для использования do regex.sub(handler, tweet) вместо re.sub(regex, handler, tweet) уменьшает вышеупомянутое значение до 8,6 с на моем ноутбуке, но я не буду использовать это изменение ниже.

Затем я попробовал вашjoblib пакет:

with open('tweets.txt') as fin, open('tweets2.txt', 'w') as fout:
    with Parallel(n_jobs=-1) as parallel:
        for l in parallel(delayed(replace_in)(tweet) for tweet in fin):
            fout.write(l)

, который занимает 1 минуту 16 секунд на моем ноутбуке и 34,2 секунды на моем рабочем столе.Загрузка ЦП была довольно низкой, поскольку все дочерние / рабочие задачи ожидали, что координатор большую часть времени отправлял им работу.

Затем я попытался использовать пакет multiprocessing:

with open('tweets.txt') as fin, open('tweets2.txt', 'w') as fout:
    with Pool() as pool:
        for l in pool.map(replace_in, fin, chunksize=1024):
            fout.write(l)

что заняло 5,95 с на моем ноутбуке и 2,60 с на моем рабочем столе.Я также попробовал с размером фрагмента 8, который занял 22,1 и 8,29 соответственно.размер чанка позволяет пулу отправлять большие порции работы своим детям, поэтому он может тратить меньше времени на координацию и больше времени на выполнение полезной работы.

Поэтому я рискну предположить, что joblib isn 'Это особенно полезно для такого рода использования, так как не имеет понятия размера фрагмента .