Какой лучший способ разделить большие файлы в Python для многопроцессорной обработки?

Question

Я сталкивался с множеством "смущающих параллельных" проектов, которые я хотел бы распараллелить с модулем multiprocessing. Однако они часто включают чтение в огромных файлах (более 2 ГБ), обработку их построчно, выполнение базовых вычислений, а затем запись результатов. Каков наилучший способ разбить файл и обработать его с помощью многопроцессорного модуля Python? Следует ли использовать Queue или JoinableQueue в multiprocessing? Или сам модуль Queue? Или я должен отобразить файл итерируемый по пулу процессов, используя multiprocessing? Я экспериментировал с этими подходами, но накладные расходы огромны при распределении данных построчно. Я остановился на дизайне облегченных конвейерных фильтров, используя cat file | process1 --out-file out1 --num-processes 2 | process2 --out-file out2, который передает определенный процент ввода первого процесса непосредственно на второй ввод (см. в этом посте ), но я бы хотел есть решение, содержащееся полностью в Python.

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

Спасибо, Винс

Дополнительная информация: Время обработки на строку варьируется. Некоторые проблемы бывают быстрыми и практически не связаны с вводом / выводом, некоторые связаны с процессором. Независимые задачи, связанные с ЦП, получат преимущество от распараллеливания, так что даже неэффективные способы назначения данных для функции обработки будут по-прежнему полезны с точки зрения времени настенных часов.

Ярким примером является скрипт, который извлекает поля из строк, проверяет различные побитовые флаги и записывает строки с определенными флагами в новый файл в совершенно новом формате. Это похоже на проблему, связанную с вводом / выводом, но когда я запустил ее с моей дешевой параллельной версией с конвейерами, она была примерно на 20% быстрее. Когда я запускаю его с пулом и картой или в очереди в multiprocessing, это всегда более чем на 100% медленнее.

Answer 1

Одна из лучших архитектур уже входит в состав ОС Linux. Никаких специальных библиотек не требуется.

Вам нужен дизайн с разветвлением.

1. «Основная» программа создает несколько подпроцессов, соединенных трубами.

2. Основная программа читает файл, записывая строки в каналы, выполняя минимальную фильтрацию, необходимую для передачи строк в соответствующие подпроцессы.

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

Вам не нужна структура данных очереди, это именно то, что представляет собой конвейер в памяти - очередь байтов между двумя параллельными процессами.

Answer 2

Одна стратегия состоит в том, чтобы назначить каждому рабочему смещение, так что если у вас есть восемь рабочих процессов, которые вы назначаете, то числа от 0 до 7. Рабочий с номером 0 читает первую запись, обрабатывает, затем пропускает 7 и переходит к обработке 8-й записи и т. Д. Рабочий номер 1 читает вторую запись, затем пропускает 7 и обрабатывает 9-ю запись .........

У этой схемы есть ряд преимуществ. Неважно, насколько большой файл, работа всегда распределяется равномерно, процессы на одном и том же компьютере будут обрабатываться примерно с одинаковой скоростью и использовать одни и те же области буфера, чтобы избежать чрезмерных затрат ввода-вывода. Пока файл не обновлен, вы можете повторно запускать отдельные потоки для восстановления после сбоев.

Answer 3

Вы не упоминаете, как вы обрабатываете строки; возможно, самая важная часть информации.

Является ли каждая строка независимой? Зависит ли расчет от одной строки, предшествующей следующей? Должны ли они быть обработаны в блоках? Сколько времени занимает обработка каждой строки? Есть ли этап обработки, который должен включать «все» данные в конце? Или можно отбросить промежуточные результаты и сохранить только промежуточные итоги? Можно ли изначально разделить файл, разделив его по количеству потоков? Или он растет по мере того, как вы его обрабатываете?

Если строки независимы, а файл не растет, единственное, что вам нужно, это согласовать «начальные адреса» и «длины» для каждого из работников; они могут независимо открываться и искать в файле, а затем вы должны просто согласовать их результаты; возможно, ожидая возвращения результатов N в очередь.

Если строки не являются независимыми, ответ будет сильно зависеть от структуры файла.

Answer 4

Fredrik Lundh's Некоторые заметки о тесте широкого поиска Тима Брея - это интересное чтение об очень похожем сценарии использования с большим количеством полезных советов. Различные другие авторы также реализовали то же самое, некоторые ссылки из этой статьи, но вы можете попробовать поискать в поиске «python wide finder» или что-то еще, чтобы найти что-то еще. (где-то было также решение, основанное на модуле multiprocessing, но оно больше не доступно)

Answer 5

Я знаю, что вы специально спрашивали о Python, но я рекомендую вам взглянуть на Hadoop (http://hadoop.apache.org/):, в котором реализован алгоритм Map and Reduce, который был специально разработан для решения такого рода проблем.

Удачи

Answer 6

Многое зависит от формата вашего файла.

Имеет ли смысл разделить это где-нибудь? Или вам нужно разделить его на новую строку? Или вам нужно убедиться, что вы разбили его в конце определения объекта?

Вместо того, чтобы разбивать файл, вы должны использовать несколько считывателей для одного файла, используя os.lseek для перехода к соответствующей части файла.

Обновление: Постер добавил, что хочет разделить на новые строки. Тогда я предлагаю следующее:

Допустим, у вас есть 4 процесса. Тогда простым решением является os.lseek для 0%, 25%, 50% и 75% файла и чтения байтов до тех пор, пока вы не достигнете первой новой строки. Это ваша отправная точка для каждого процесса. Вам не нужно разбивать файл, чтобы сделать это, просто найдите правильное место в большом файле в каждом процессе и начните чтение оттуда.

Answer 7

Если время выполнения велико, вместо того, чтобы каждый процесс считывал свою следующую строку через Queue, пусть процессы читают пакеты строк. Таким образом, накладные расходы амортизируются по нескольким строкам (например, по тысячам или более).