Как мога да паралелизирам конвейер от генератори/итератори в Python?

Не можете наистина да паралелизирате четенето от или записването във файлове; те ще бъдат вашето тясно място в крайна сметка. Сигурни ли сте, че вашето тясно място тук е процесорът, а не I/O?

Тъй като вашата обработка не съдържа зависимости (според вас), е тривиално лесно да се използва Многопроцесорна обработка на Python. Пул клас.

Има няколко начина да напишете това, но по-лесният w.r.t. отстраняването на грешки е да се намерят независими критични пътища (най-бавната част от кода), които ще накараме да работят паралелно. Да приемем, че е process_item.

…И това е всъщност. Код:

import multiprocessing.Pool

p = multiprocessing.Pool() # use all available CPUs

input = open("input.txt")
x = (process_line(line) for line in input)
y = p.imap(process_item, x)
z = (generate_output_line(item) + "\n" for item in y)
output = open("output.txt", "w")
output.writelines(z)

Не съм го тествал, но това е основната идея. Imap методът на Pool гарантира, че резултатите се връщат в правилния ред.

има ли някакъв лесен начин да направите така, че няколко реда да могат да бъдат в тръбопровода наведнъж

Написах библиотека, която да прави точно това: https://github.com/michalc/threaded-buffered-pipeline, който итерира всеки итерируем в отделна нишка.

И така, какво беше

input = open("input.txt")

x = (process_line(line) for line in input)
y = (process_item(item) for item in x)
z = (generate_output_line(item) + "\n" for item in y)

output = open("output.txt", "w")
output.writelines(z)

става

from threaded_buffered_pipeline import buffered_pipeline

input = open("input.txt")

buffer_iterable = buffered_pipeline()
x = buffer_iterable((process_line(line) for line in input))
y = buffer_iterable((process_item(item) for item in x))
z = buffer_iterable((generate_output_line(item) + "\n" for item in y))

output = open("output.txt", "w")
output.writelines(z)

Колко действителен паралелизъм добавя това зависи от това какво всъщност се случва във всяка итерируема и колко CPU ядра имате/колко са заети.

Класическият пример е GIL на Python: ако всяка стъпка е сравнително тежка за процесора и използва само Python, тогава няма да се добави много паралелизъм и това може да не е по-бързо от серийната версия. От друга страна, ако всеки от тях е мрежов IO тежък, тогава мисля, че е вероятно да бъде по-бърз.