Как я могу распараллелить конвейер генераторов/итераторов в Python?

Вы не можете распараллелить чтение или запись в файлы; это будет вашим узким местом, в конечном счете. Вы уверены, что узким местом здесь является ЦП, а не ввод-вывод?

Поскольку ваша обработка не содержит зависимостей (по вашему мнению), использовать тривиально просто. Класс multiprocessing.Pool 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 пула гарантирует, что результаты возвращаются в правильном порядке.

есть ли простой способ сделать так, чтобы несколько строк могли быть в конвейере одновременно

Я написал библиотеку для этого: 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)

Сколько фактического параллелизма это добавляет, зависит от того, что на самом деле происходит в каждой итерации, и от того, сколько ядер ЦП у вас есть/насколько они заняты.

Классическим примером является Python GIL: если каждый шаг довольно сильно загружает процессор и использует только Python, тогда не будет добавлено много параллелизма, и это может быть не быстрее, чем последовательная версия. С другой стороны, если каждый из них является тяжелым сетевым вводом-выводом, то я думаю, что он, вероятно, будет быстрее.